1#![allow(
6 clippy::missing_errors_doc,
7 clippy::missing_panics_doc,
8 clippy::needless_pass_by_value,
9 clippy::too_many_lines
10)]
11
12use sha2::{Digest as _, Sha256};
13
14pub const WATCH_SESSION_CONFIGURATION_V1_SCHEMA: &str = "presolve.watch-session-configuration";
15pub const WATCH_CHANGE_BATCH_V1_SCHEMA: &str = "presolve.watch-change-batch";
16pub const WATCH_EXECUTION_PLAN_V1_SCHEMA: &str = "presolve.watch-execution-plan";
17pub const WATCH_EVENT_V1_SCHEMA: &str = "presolve.watch-event";
18pub const WATCH_SESSION_SNAPSHOT_V1_SCHEMA: &str = "presolve.watch-session-snapshot";
19pub const WATCH_EXECUTION_REPORT_V1_SCHEMA: &str = "presolve.watch-execution-report";
20pub const MAX_WATCH_SESSIONS: usize = 64;
21pub const MAX_WATCH_EVENTS: usize = 1024;
22pub const MAX_OBSERVED_CHANGES: usize = 4096;
23pub const MAX_LOGICAL_PATH_BYTES: usize = 4096;
24pub const MAX_DEBOUNCE_MILLISECONDS: u64 = 86_400_000;
25
26#[derive(Debug, Clone, Copy, PartialEq, Eq)]
27pub struct WatchDebounceV1 {
28 pub quiet_period_milliseconds: u64,
29 pub maximum_delay_milliseconds: u64,
30}
31#[derive(Debug, Clone, PartialEq, Eq)]
32pub struct WatchSessionConfigurationV1 {
33 pub schema: String,
34 pub version: u32,
35 pub watch_session_id: String,
36 pub workspace_id: String,
37 pub debounce: WatchDebounceV1,
38 pub supersession_policy: String,
39 pub event_detail: String,
40 pub event_journal_capacity: usize,
41}
42#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
43pub struct ObservedChangeV1 {
44 pub kind: String,
45 pub logical_path: String,
46 pub previous_logical_path: Option<String>,
47}
48#[derive(Debug, Clone, PartialEq, Eq)]
49pub struct WatchCandidateMetadataV1 {
50 pub candidate_id: String,
51 pub fingerprint: String,
52}
53#[derive(Debug, Clone, PartialEq, Eq)]
54pub struct WatchChangeBatchMetadataV1 {
55 pub schema: String,
56 pub version: u32,
57 pub watch_session_id: String,
58 pub sequence: u64,
59 pub observed_changes: Vec<ObservedChangeV1>,
60 pub candidate: WatchCandidateMetadataV1,
61}
62#[derive(Debug, Clone, PartialEq, Eq)]
63pub struct WatchExecutionPlanV1 {
64 pub schema: &'static str,
65 pub version: u32,
66 pub watch_session_id: String,
67 pub workspace_id: String,
68 pub first_sequence: u64,
69 pub last_sequence: u64,
70 pub coalesced_batch_count: u64,
71 pub candidate_fingerprint: String,
72 pub observed_changes: Vec<ObservedChangeV1>,
73 pub eligibility: String,
74 pub plan_identity: String,
75}
76#[derive(Debug, Clone, PartialEq, Eq)]
77pub struct WatchEventV1 {
78 pub schema: &'static str,
79 pub version: u32,
80 pub watch_session_id: String,
81 pub event_sequence: u64,
82 pub kind: String,
83 pub first_sequence: Option<u64>,
84 pub last_sequence: Option<u64>,
85 pub candidate_fingerprint: Option<String>,
86 pub workspace_result_identity: Option<String>,
87 pub code: Option<String>,
88 pub event_identity: String,
89}
90#[derive(Debug, Clone, PartialEq, Eq)]
91pub struct WatchSessionSnapshotV1 {
92 pub schema: &'static str,
93 pub version: u32,
94 pub watch_session_id: String,
95 pub workspace_id: String,
96 pub state: String,
97 pub last_accepted_sequence: Option<u64>,
98 pub active_candidate_fingerprint: Option<String>,
99 pub pending_candidate_fingerprint: Option<String>,
100 pub pending_sequence_range: Option<(u64, u64)>,
101 pub last_successful_candidate_fingerprint: Option<String>,
102 pub last_successful_workspace_result_identity: Option<String>,
103 pub next_event_sequence: u64,
104 pub retained_event_cursor_range: Option<(u64, u64)>,
105 pub pending_count: usize,
106 pub active_count: usize,
107}
108#[derive(Debug, Clone, PartialEq, Eq)]
109pub struct WatchExecutionReportV1 {
110 pub plan: WatchExecutionPlanV1,
111 pub outcome: String,
112 pub workspace_result_identity: Option<String>,
113}
114#[derive(Debug, Clone, PartialEq, Eq)]
115pub struct WatchAcceptanceV1 {
116 pub classification: &'static str,
117 pub retained_pending: bool,
118 pub obsolete_active: bool,
119}
120
121#[derive(Debug, Clone, PartialEq, Eq)]
122pub enum WatchErrorV1 {
123 Code(&'static str),
124}
125impl WatchErrorV1 {
126 #[must_use]
127 pub const fn code(&self) -> &'static str {
128 match self {
129 Self::Code(c) => c,
130 }
131 }
132}
133
134#[derive(Debug, Clone)]
135struct Pending {
136 first: u64,
137 last: u64,
138 batches: u64,
139 window_start: u64,
140 quiet_deadline: u64,
141 maximum_deadline: u64,
142 changes: Vec<ObservedChangeV1>,
143 candidate: WatchCandidateMetadataV1,
144}
145#[derive(Debug, Clone)]
146struct Active {
147 plan: WatchExecutionPlanV1,
148 obsolete: bool,
149}
150#[derive(Debug, Clone)]
151pub struct WatchSessionV1 {
152 configuration: WatchSessionConfigurationV1,
153 last_sequence: Option<u64>,
154 pending: Option<Pending>,
155 active: Option<Active>,
156 last_success: Option<(String, String, u64)>,
157 events: Vec<WatchEventV1>,
158 next_event: u64,
159 open: bool,
160}
161fn hash(value: impl AsRef<[u8]>) -> String {
162 format!("sha256:{:x}", Sha256::digest(value))
163}
164fn normalize_text(value: &str) -> Result<String, WatchErrorV1> {
165 let value = value.trim();
166 if value.is_empty() || value.bytes().any(|b| b.is_ascii_control()) {
167 Err(WatchErrorV1::Code("L8W004_INVALID_WATCH_CONFIGURATION"))
168 } else {
169 Ok(value.into())
170 }
171}
172fn normalize_path(value: &str) -> Result<String, WatchErrorV1> {
173 if value.is_empty()
174 || value.len() > MAX_LOGICAL_PATH_BYTES
175 || value.starts_with('/')
176 || value
177 .split('/')
178 .any(|p| p.is_empty() || p == "." || p == "..")
179 {
180 return Err(WatchErrorV1::Code("L8W007_INVALID_CHANGE_BATCH"));
181 }
182 Ok(value.into())
183}
184fn canonical_changes(changes: &mut Vec<ObservedChangeV1>) -> Result<(), WatchErrorV1> {
185 if changes.len() > MAX_OBSERVED_CHANGES {
186 return Err(WatchErrorV1::Code("L8W010_WATCH_RESOURCE_LIMIT_EXCEEDED"));
187 }
188 for change in changes.iter_mut() {
189 if !matches!(
190 change.kind.as_str(),
191 "created" | "modified" | "deleted" | "renamed"
192 ) {
193 return Err(WatchErrorV1::Code("L8W007_INVALID_CHANGE_BATCH"));
194 }
195 change.logical_path = normalize_path(&change.logical_path)?;
196 if change.kind == "renamed" {
197 change.previous_logical_path = Some(normalize_path(
198 change
199 .previous_logical_path
200 .as_deref()
201 .ok_or(WatchErrorV1::Code("L8W007_INVALID_CHANGE_BATCH"))?,
202 )?);
203 } else if change.previous_logical_path.is_some() {
204 return Err(WatchErrorV1::Code("L8W007_INVALID_CHANGE_BATCH"));
205 }
206 }
207 changes.sort();
208 changes.dedup();
209 Ok(())
210}
211impl WatchSessionConfigurationV1 {
212 pub fn normalize_validate(&self) -> Result<Self, WatchErrorV1> {
213 if self.schema != WATCH_SESSION_CONFIGURATION_V1_SCHEMA || self.version != 1 {
214 return Err(WatchErrorV1::Code("L8W005_UNSUPPORTED_WATCH_SCHEMA"));
215 }
216 if self.debounce.maximum_delay_milliseconds < self.debounce.quiet_period_milliseconds
217 || self.debounce.maximum_delay_milliseconds > MAX_DEBOUNCE_MILLISECONDS
218 || self.supersession_policy != "cancel_obsolete"
219 || !matches!(self.event_detail.as_str(), "summary" | "full")
220 {
221 return Err(WatchErrorV1::Code("L8W004_INVALID_WATCH_CONFIGURATION"));
222 }
223 if self.event_journal_capacity == 0 || self.event_journal_capacity > MAX_WATCH_EVENTS {
224 return Err(WatchErrorV1::Code("L8W010_WATCH_RESOURCE_LIMIT_EXCEEDED"));
225 }
226 let mut out = self.clone();
227 out.watch_session_id = normalize_text(&out.watch_session_id)?;
228 out.workspace_id = normalize_text(&out.workspace_id)?;
229 Ok(out)
230 }
231}
232impl WatchSessionV1 {
233 pub fn new(configuration: WatchSessionConfigurationV1) -> Result<Self, WatchErrorV1> {
234 let mut out = Self {
235 configuration: configuration.normalize_validate()?,
236 last_sequence: None,
237 pending: None,
238 active: None,
239 last_success: None,
240 events: vec![],
241 next_event: 1,
242 open: true,
243 };
244 out.event("watch_started", None, None, None, None, None);
245 Ok(out)
246 }
247 #[must_use]
248 pub fn configuration(&self) -> &WatchSessionConfigurationV1 {
249 &self.configuration
250 }
251 fn event(
252 &mut self,
253 kind: &str,
254 first: Option<u64>,
255 last: Option<u64>,
256 fingerprint: Option<String>,
257 result: Option<String>,
258 code: Option<&str>,
259 ) {
260 let sequence = self.next_event;
261 self.next_event += 1;
262 let identity = hash(format!(
263 "{}|{}|{}|{:?}|{:?}|{:?}|{:?}",
264 self.configuration.watch_session_id, sequence, kind, first, last, fingerprint, result
265 ));
266 self.events.push(WatchEventV1 {
267 schema: WATCH_EVENT_V1_SCHEMA,
268 version: 1,
269 watch_session_id: self.configuration.watch_session_id.clone(),
270 event_sequence: sequence,
271 kind: kind.into(),
272 first_sequence: first,
273 last_sequence: last,
274 candidate_fingerprint: fingerprint,
275 workspace_result_identity: result,
276 code: code.map(Into::into),
277 event_identity: identity,
278 });
279 if self.events.len() > self.configuration.event_journal_capacity {
280 self.events.remove(0);
281 }
282 }
283 pub fn accept(
284 &mut self,
285 mut batch: WatchChangeBatchMetadataV1,
286 now: u64,
287 ) -> Result<WatchAcceptanceV1, WatchErrorV1> {
288 if !self.open {
289 return Err(WatchErrorV1::Code("L8W009_WATCH_SESSION_CLOSED"));
290 }
291 if batch.schema != WATCH_CHANGE_BATCH_V1_SCHEMA || batch.version != 1 {
292 return Err(WatchErrorV1::Code("L8W005_UNSUPPORTED_WATCH_SCHEMA"));
293 }
294 if batch.watch_session_id != self.configuration.watch_session_id || batch.sequence == 0 {
295 return Err(WatchErrorV1::Code("L8W007_INVALID_CHANGE_BATCH"));
296 }
297 if self
298 .last_sequence
299 .is_some_and(|last| batch.sequence <= last)
300 {
301 self.event(
302 "candidate_rejected",
303 Some(batch.sequence),
304 Some(batch.sequence),
305 None,
306 None,
307 Some("L8W006_NON_MONOTONIC_CHANGE_SEQUENCE"),
308 );
309 return Err(WatchErrorV1::Code("L8W006_NON_MONOTONIC_CHANGE_SEQUENCE"));
310 }
311 batch.candidate.candidate_id = normalize_text(&batch.candidate.candidate_id)?;
312 batch.candidate.fingerprint = normalize_text(&batch.candidate.fingerprint)?;
313 canonical_changes(&mut batch.observed_changes)?;
314 self.last_sequence = Some(batch.sequence);
315 if self
316 .last_success
317 .as_ref()
318 .is_some_and(|(f, _, _)| *f == batch.candidate.fingerprint)
319 {
320 self.event(
321 "candidate_noop",
322 Some(batch.sequence),
323 Some(batch.sequence),
324 Some(batch.candidate.fingerprint),
325 None,
326 Some("L8R005_EQUAL_TO_PUBLISHED_CANDIDATE"),
327 );
328 return Ok(WatchAcceptanceV1 {
329 classification: "L8R005_EQUAL_TO_PUBLISHED_CANDIDATE",
330 retained_pending: false,
331 obsolete_active: false,
332 });
333 }
334 if self
335 .active
336 .as_ref()
337 .is_some_and(|a| a.plan.candidate_fingerprint == batch.candidate.fingerprint)
338 {
339 self.event(
340 "candidate_noop",
341 Some(batch.sequence),
342 Some(batch.sequence),
343 Some(batch.candidate.fingerprint),
344 None,
345 Some("L8R004_EQUAL_TO_ACTIVE_CANDIDATE"),
346 );
347 return Ok(WatchAcceptanceV1 {
348 classification: "L8R004_EQUAL_TO_ACTIVE_CANDIDATE",
349 retained_pending: false,
350 obsolete_active: false,
351 });
352 }
353 let (first, batches, window_start, mut changes) = match self.pending.take() {
354 Some(p) => (p.first, p.batches + 1, p.window_start, p.changes),
355 None => (batch.sequence, 1, now, vec![]),
356 };
357 changes.extend(batch.observed_changes);
358 canonical_changes(&mut changes)?;
359 let maximum =
360 window_start.saturating_add(self.configuration.debounce.maximum_delay_milliseconds);
361 let quiet = now.saturating_add(self.configuration.debounce.quiet_period_milliseconds);
362 let replaced = first != batch.sequence;
363 self.pending = Some(Pending {
364 first,
365 last: batch.sequence,
366 batches,
367 window_start,
368 quiet_deadline: quiet,
369 maximum_deadline: maximum,
370 changes,
371 candidate: batch.candidate.clone(),
372 });
373 let obsolete = if let Some(active) = self.active.as_mut() {
374 active.obsolete = true;
375 true
376 } else {
377 false
378 };
379 self.event(
380 if replaced {
381 "candidate_coalesced"
382 } else {
383 "candidate_accepted"
384 },
385 Some(first),
386 Some(batch.sequence),
387 Some(batch.candidate.fingerprint.clone()),
388 None,
389 Some(if obsolete {
390 "L8R003_ACTIVE_CANDIDATE_SUPERSEDED"
391 } else if replaced {
392 "L8R002_PENDING_CANDIDATE_REPLACED"
393 } else {
394 "L8R001_FIRST_CANDIDATE"
395 }),
396 );
397 if obsolete {
398 self.event(
399 "candidate_superseded",
400 Some(first),
401 Some(batch.sequence),
402 Some(batch.candidate.fingerprint),
403 None,
404 Some("L8R003_ACTIVE_CANDIDATE_SUPERSEDED"),
405 );
406 }
407 Ok(WatchAcceptanceV1 {
408 classification: if obsolete {
409 "L8R003_ACTIVE_CANDIDATE_SUPERSEDED"
410 } else if replaced {
411 "L8R002_PENDING_CANDIDATE_REPLACED"
412 } else {
413 "L8R001_FIRST_CANDIDATE"
414 },
415 retained_pending: true,
416 obsolete_active: obsolete,
417 })
418 }
419 pub fn scheduler_turn(&mut self, now: u64) -> Option<WatchExecutionPlanV1> {
420 if !self.open || self.active.is_some() {
421 return None;
422 }
423 let pending = self.pending.as_ref()?;
424 if now < pending.quiet_deadline && now < pending.maximum_deadline {
425 return None;
426 }
427 let pending = self.pending.take()?;
428 let eligibility = if self.configuration.debounce.quiet_period_milliseconds == 0
429 && self.configuration.debounce.maximum_delay_milliseconds == 0
430 {
431 "immediate_zero_debounce"
432 } else if now >= pending.maximum_deadline
433 && pending.maximum_deadline <= pending.quiet_deadline
434 {
435 "maximum_delay_elapsed"
436 } else {
437 "quiet_period_elapsed"
438 };
439 let plan_identity = hash(format!(
440 "{}|{}|{}|{}|{}|{:?}|{}",
441 self.configuration.watch_session_id,
442 pending.first,
443 pending.last,
444 pending.batches,
445 pending.candidate.fingerprint,
446 pending.changes,
447 eligibility
448 ));
449 let plan = WatchExecutionPlanV1 {
450 schema: WATCH_EXECUTION_PLAN_V1_SCHEMA,
451 version: 1,
452 watch_session_id: self.configuration.watch_session_id.clone(),
453 workspace_id: self.configuration.workspace_id.clone(),
454 first_sequence: pending.first,
455 last_sequence: pending.last,
456 coalesced_batch_count: pending.batches,
457 candidate_fingerprint: pending.candidate.fingerprint,
458 observed_changes: pending.changes,
459 eligibility: eligibility.into(),
460 plan_identity,
461 };
462 self.event(
463 "build_scheduled",
464 Some(plan.first_sequence),
465 Some(plan.last_sequence),
466 Some(plan.candidate_fingerprint.clone()),
467 None,
468 Some(match eligibility {
469 "maximum_delay_elapsed" => "L8R007_MAXIMUM_DELAY_ELAPSED",
470 "immediate_zero_debounce" => "L8R012_ZERO_DEBOUNCE_COALESCED",
471 _ => "L8R006_QUIET_PERIOD_ELAPSED",
472 }),
473 );
474 self.event(
475 "build_started",
476 Some(plan.first_sequence),
477 Some(plan.last_sequence),
478 Some(plan.candidate_fingerprint.clone()),
479 None,
480 None,
481 );
482 self.active = Some(Active {
483 plan: plan.clone(),
484 obsolete: false,
485 });
486 Some(plan)
487 }
488 pub fn flush(&mut self) -> Result<(), WatchErrorV1> {
489 if !self.open {
490 return Err(WatchErrorV1::Code("L8W009_WATCH_SESSION_CLOSED"));
491 }
492 let Some(p) = self.pending.as_mut() else {
493 return Err(WatchErrorV1::Code("L8W011_FLUSH_WITHOUT_PENDING_CANDIDATE"));
494 };
495 p.quiet_deadline = 0;
496 p.maximum_deadline = 0;
497 let (first, last, fingerprint) = (p.first, p.last, p.candidate.fingerprint.clone());
498 self.event(
499 "watch_flushed",
500 Some(first),
501 Some(last),
502 Some(fingerprint),
503 None,
504 Some("L8R008_MANUAL_FLUSH"),
505 );
506 Ok(())
507 }
508 pub fn finish(
509 &mut self,
510 success: bool,
511 workspace_result_identity: Option<String>,
512 ) -> Option<WatchExecutionReportV1> {
513 let active = self.active.take()?;
514 let outcome = if active.obsolete {
515 if success {
516 self.event(
517 "obsolete_result_discarded",
518 Some(active.plan.first_sequence),
519 Some(active.plan.last_sequence),
520 Some(active.plan.candidate_fingerprint.clone()),
521 workspace_result_identity.clone(),
522 Some("L8R011_OBSOLETE_SUCCESS_DISCARDED"),
523 );
524 "obsolete_discarded"
525 } else {
526 self.event(
527 "build_superseded",
528 Some(active.plan.first_sequence),
529 Some(active.plan.last_sequence),
530 Some(active.plan.candidate_fingerprint.clone()),
531 None,
532 None,
533 );
534 "superseded"
535 }
536 } else if success {
537 let id = workspace_result_identity.clone().unwrap_or_default();
538 self.last_success = Some((
539 active.plan.candidate_fingerprint.clone(),
540 id,
541 active.plan.last_sequence,
542 ));
543 self.event(
544 "build_succeeded",
545 Some(active.plan.first_sequence),
546 Some(active.plan.last_sequence),
547 Some(active.plan.candidate_fingerprint.clone()),
548 workspace_result_identity.clone(),
549 None,
550 );
551 "succeeded"
552 } else {
553 self.event(
554 "build_failed",
555 Some(active.plan.first_sequence),
556 Some(active.plan.last_sequence),
557 Some(active.plan.candidate_fingerprint.clone()),
558 None,
559 Some("L8R009_BUILD_FAILED_RETAINING_PRIOR_SUCCESS"),
560 );
561 "failed"
562 };
563 Some(WatchExecutionReportV1 {
564 plan: active.plan,
565 outcome: outcome.into(),
566 workspace_result_identity,
567 })
568 }
569 pub fn poll(&self, after: u64) -> Result<Vec<WatchEventV1>, WatchErrorV1> {
570 if let Some(first) = self.events.first().map(|e| e.event_sequence) {
571 if after.saturating_add(1) < first {
572 return Err(WatchErrorV1::Code("L8W012_EVENT_CURSOR_OUT_OF_RANGE"));
573 }
574 }
575 Ok(self
576 .events
577 .iter()
578 .filter(|e| e.event_sequence > after)
579 .cloned()
580 .collect())
581 }
582 #[must_use]
583 pub fn snapshot(&self) -> WatchSessionSnapshotV1 {
584 let range = self
585 .events
586 .first()
587 .zip(self.events.last())
588 .map(|(a, b)| (a.event_sequence, b.event_sequence));
589 WatchSessionSnapshotV1 {
590 schema: WATCH_SESSION_SNAPSHOT_V1_SCHEMA,
591 version: 1,
592 watch_session_id: self.configuration.watch_session_id.clone(),
593 workspace_id: self.configuration.workspace_id.clone(),
594 state: if self.open { "open" } else { "closed" }.into(),
595 last_accepted_sequence: self.last_sequence,
596 active_candidate_fingerprint: self
597 .active
598 .as_ref()
599 .map(|a| a.plan.candidate_fingerprint.clone()),
600 pending_candidate_fingerprint: self
601 .pending
602 .as_ref()
603 .map(|p| p.candidate.fingerprint.clone()),
604 pending_sequence_range: self.pending.as_ref().map(|p| (p.first, p.last)),
605 last_successful_candidate_fingerprint: self.last_success.as_ref().map(|x| x.0.clone()),
606 last_successful_workspace_result_identity: self
607 .last_success
608 .as_ref()
609 .map(|x| x.1.clone()),
610 next_event_sequence: self.next_event,
611 retained_event_cursor_range: range,
612 pending_count: usize::from(self.pending.is_some()),
613 active_count: usize::from(self.active.is_some()),
614 }
615 }
616 pub fn stop(&mut self) {
617 self.pending = None;
618 self.open = false;
619 self.event("watch_stopped", None, None, None, None, None);
620 }
621}
622
623#[cfg(test)]
624mod tests {
625 use super::*;
626 fn config() -> WatchSessionConfigurationV1 {
627 WatchSessionConfigurationV1 {
628 schema: WATCH_SESSION_CONFIGURATION_V1_SCHEMA.into(),
629 version: 1,
630 watch_session_id: "watch-a".into(),
631 workspace_id: "workspace-a".into(),
632 debounce: WatchDebounceV1 {
633 quiet_period_milliseconds: 10,
634 maximum_delay_milliseconds: 30,
635 },
636 supersession_policy: "cancel_obsolete".into(),
637 event_detail: "summary".into(),
638 event_journal_capacity: 8,
639 }
640 }
641 fn batch(sequence: u64, fingerprint: &str) -> WatchChangeBatchMetadataV1 {
642 WatchChangeBatchMetadataV1 {
643 schema: WATCH_CHANGE_BATCH_V1_SCHEMA.into(),
644 version: 1,
645 watch_session_id: "watch-a".into(),
646 sequence,
647 observed_changes: vec![ObservedChangeV1 {
648 kind: "modified".into(),
649 logical_path: format!("src/{sequence}.ts"),
650 previous_logical_path: None,
651 }],
652 candidate: WatchCandidateMetadataV1 {
653 candidate_id: format!("c{sequence}"),
654 fingerprint: fingerprint.into(),
655 },
656 }
657 }
658 #[test]
659 fn fake_clock_debounce_coalesces_latest() {
660 let mut s = WatchSessionV1::new(config()).unwrap();
661 s.accept(batch(1, "one"), 0).unwrap();
662 s.accept(batch(2, "two"), 5).unwrap();
663 assert!(s.scheduler_turn(14).is_none());
664 let p = s.scheduler_turn(15).unwrap();
665 assert_eq!(
666 (
667 p.first_sequence,
668 p.last_sequence,
669 p.candidate_fingerprint.as_str()
670 ),
671 (1, 2, "two")
672 );
673 }
674 #[test]
675 fn maximum_delay_is_not_reset() {
676 let mut s = WatchSessionV1::new(config()).unwrap();
677 s.accept(batch(1, "one"), 0).unwrap();
678 s.accept(batch(2, "two"), 25).unwrap();
679 assert_eq!(
680 s.scheduler_turn(30).unwrap().eligibility,
681 "maximum_delay_elapsed"
682 );
683 }
684 #[test]
685 fn zero_debounce_waits_for_scheduler_turn() {
686 let mut c = config();
687 c.debounce = WatchDebounceV1 {
688 quiet_period_milliseconds: 0,
689 maximum_delay_milliseconds: 0,
690 };
691 let mut s = WatchSessionV1::new(c).unwrap();
692 s.accept(batch(1, "one"), 0).unwrap();
693 s.accept(batch(2, "two"), 0).unwrap();
694 assert_eq!(s.scheduler_turn(0).unwrap().last_sequence, 2);
695 }
696 #[test]
697 fn obsolete_success_is_discarded() {
698 let mut s = WatchSessionV1::new(config()).unwrap();
699 s.accept(batch(1, "one"), 0).unwrap();
700 s.scheduler_turn(10).unwrap();
701 assert!(s.accept(batch(2, "two"), 10).unwrap().obsolete_active);
702 assert_eq!(
703 s.finish(true, Some("old".into())).unwrap().outcome,
704 "obsolete_discarded"
705 );
706 }
707 #[test]
708 fn equal_published_candidate_is_a_noop_and_sequences_remain_monotonic() {
709 let mut s = WatchSessionV1::new(config()).unwrap();
710 s.accept(batch(4, "same"), 0).unwrap();
711 s.scheduler_turn(10).unwrap();
712 s.finish(true, Some("result".into())).unwrap();
713 assert_eq!(
714 s.accept(batch(8, "same"), 11).unwrap().classification,
715 "L8R005_EQUAL_TO_PUBLISHED_CANDIDATE"
716 );
717 assert_eq!(
718 s.accept(batch(8, "later"), 12).unwrap_err().code(),
719 "L8W006_NON_MONOTONIC_CHANGE_SEQUENCE"
720 );
721 }
722 #[test]
723 fn flush_and_event_eviction_are_source_free() {
724 let mut c = config();
725 c.event_journal_capacity = 3;
726 let mut s = WatchSessionV1::new(c).unwrap();
727 assert_eq!(
728 s.flush().unwrap_err().code(),
729 "L8W011_FLUSH_WITHOUT_PENDING_CANDIDATE"
730 );
731 s.accept(batch(1, "one"), 1).unwrap();
732 s.flush().unwrap();
733 s.scheduler_turn(1).unwrap();
734 s.finish(false, None).unwrap();
735 let events = s.poll(3).unwrap();
736 assert!(events
737 .iter()
738 .all(|event| !format!("{event:?}").contains("source text")));
739 assert_eq!(s.snapshot().retained_event_cursor_range, Some((4, 6)));
740 assert_eq!(
741 s.poll(0).unwrap_err().code(),
742 "L8W012_EVENT_CURSOR_OUT_OF_RANGE"
743 );
744 }
745 #[test]
746 fn stop_releases_pending_and_rejects_future_batches() {
747 let mut s = WatchSessionV1::new(config()).unwrap();
748 s.accept(batch(1, "one"), 0).unwrap();
749 s.stop();
750 assert_eq!(s.snapshot().pending_count, 0);
751 assert_eq!(
752 s.accept(batch(2, "two"), 1).unwrap_err().code(),
753 "L8W009_WATCH_SESSION_CLOSED"
754 );
755 }
756 #[test]
757 fn deterministic_twenty_fresh_runs() {
758 let mut expected = None;
759 for _ in 0..20 {
760 let mut s = WatchSessionV1::new(config()).unwrap();
761 s.accept(batch(1, "one"), 0).unwrap();
762 s.accept(batch(2, "two"), 4).unwrap();
763 let p = s.scheduler_turn(14).unwrap();
764 let _r = s.finish(true, Some("result".into())).unwrap();
765 let got = format!("{:?}{:?}", p, s.poll(0).unwrap());
766 if let Some(old) = &expected {
767 assert_eq!(old, &got);
768 } else {
769 expected = Some(got);
770 }
771 }
772 }
773}