fxrs_process/
monitor.rs

1//! Executor-neutral terminal-monitor state machine.
2//!
3//! This module deliberately contains no threads, clocks, sockets, filesystem
4//! access, or process inspection. A native session, tmux recovery host, or
5//! detached supervisor supplies observations and persists [`Monitor`] values;
6//! all three therefore share validation, scheduling, matching, and event
7//! semantics.
8
9use serde::{Deserialize, Serialize};
10
11pub const MINIMUM_SCHEDULE_MS: u64 = 10;
12pub const MAXIMUM_SCHEDULE_MS: u64 = 24 * 60 * 60 * 1_000;
13pub const MAXIMUM_LIFETIME_MS: u64 = 365 * 24 * 60 * 60 * 1_000;
14pub const MAXIMUM_PATTERN_BYTES: usize = 256;
15pub const MAXIMUM_MONITOR_ID_BYTES: usize = 128;
16pub const MAXIMUM_AUTHORITY_TEXT_BYTES: usize = 4_096;
17pub const MAXIMUM_COMMAND_BYTES: usize = 64 * 1_024;
18const PATTERN_WORDS: usize = (MAXIMUM_PATTERN_BYTES + 1).div_ceil(64);
19
20#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
21#[serde(rename_all = "snake_case")]
22pub enum MonitorSignal {
23    Hangup,
24    Interrupt,
25    Quit,
26    Terminate,
27    Kill,
28}
29
30#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
31#[serde(tag = "kind", rename_all = "snake_case")]
32pub enum MonitorCondition {
33    ProcessExit,
34    ExitCode { exit_code: i32 },
35    Signal { signal: MonitorSignal },
36    OutputContains { pattern: String },
37    OutputMatches { pattern: String },
38    OutputQuiet { duration_ms: u64 },
39    ScreenMatches { pattern: String },
40    TcpReady { host: String, port: u16 },
41    HttpReady { pattern: String },
42    PathExists { path: String },
43    PathChanged { path: String },
44    PathSize { path: String, minimum_bytes: u64 },
45    CustomProbe { command: String, cwd: String },
46}
47
48impl MonitorCondition {
49    pub fn requires_polling(&self) -> bool {
50        matches!(
51            self,
52            Self::TcpReady { .. }
53                | Self::HttpReady { .. }
54                | Self::PathExists { .. }
55                | Self::PathChanged { .. }
56                | Self::PathSize { .. }
57                | Self::CustomProbe { .. }
58        )
59    }
60
61    fn validate(&self) -> Result<(), MonitorError> {
62        match self {
63            Self::ProcessExit | Self::ExitCode { .. } | Self::Signal { .. } => Ok(()),
64            Self::OutputContains { pattern }
65            | Self::OutputMatches { pattern }
66            | Self::ScreenMatches { pattern } => valid_text(
67                pattern,
68                MAXIMUM_PATTERN_BYTES,
69                MonitorError::InvalidCondition,
70            ),
71            Self::OutputQuiet { duration_ms } => {
72                validate_schedule(*duration_ms).map_err(|_| MonitorError::InvalidCondition)
73            }
74            Self::TcpReady { host, port } => {
75                valid_text(
76                    host,
77                    MAXIMUM_AUTHORITY_TEXT_BYTES,
78                    MonitorError::InvalidCondition,
79                )?;
80                if *port == 0 {
81                    return Err(MonitorError::InvalidCondition);
82                }
83                Ok(())
84            }
85            Self::HttpReady { pattern } => valid_text(
86                pattern,
87                MAXIMUM_AUTHORITY_TEXT_BYTES,
88                MonitorError::InvalidCondition,
89            ),
90            Self::PathExists { path }
91            | Self::PathChanged { path }
92            | Self::PathSize { path, .. } => valid_text(
93                path,
94                MAXIMUM_AUTHORITY_TEXT_BYTES,
95                MonitorError::InvalidCondition,
96            ),
97            Self::CustomProbe { command, cwd } => {
98                valid_text(
99                    command,
100                    MAXIMUM_COMMAND_BYTES,
101                    MonitorError::InvalidCondition,
102                )?;
103                valid_text(
104                    cwd,
105                    MAXIMUM_AUTHORITY_TEXT_BYTES,
106                    MonitorError::InvalidCondition,
107                )
108            }
109        }
110    }
111
112    fn is_gap_sensitive(&self) -> bool {
113        matches!(
114            self,
115            Self::OutputContains { .. }
116                | Self::OutputMatches { .. }
117                | Self::OutputQuiet { .. }
118                | Self::ScreenMatches { .. }
119        )
120    }
121}
122
123#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
124#[serde(tag = "kind", rename_all = "snake_case")]
125pub enum NotifySchedule {
126    OnMatch,
127    OnStateChange,
128    OnExit,
129    EveryCheck,
130    EveryNChecks { count: u32 },
131    Interval { interval_ms: u64 },
132}
133
134#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
135#[serde(tag = "kind", rename_all = "snake_case")]
136pub enum MonitorLifetime {
137    UntilMatch,
138    UntilSessionEnd,
139    Duration { duration_ms: u64 },
140}
141
142#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
143#[serde(deny_unknown_fields)]
144pub struct MonitorDefinition {
145    pub condition: MonitorCondition,
146    #[serde(default, skip_serializing_if = "Option::is_none")]
147    pub check_interval_ms: Option<u64>,
148    pub notify: NotifySchedule,
149    pub lifetime: MonitorLifetime,
150}
151
152impl MonitorDefinition {
153    pub fn validate(&self) -> Result<(), MonitorError> {
154        self.condition.validate()?;
155        match &self.notify {
156            NotifySchedule::EveryNChecks { count } if *count == 0 || *count > 1_000_000 => {
157                return Err(MonitorError::InvalidSchedule);
158            }
159            NotifySchedule::Interval { interval_ms } => validate_schedule(*interval_ms)?,
160            _ => {}
161        }
162        if let MonitorLifetime::Duration { duration_ms } = self.lifetime
163            && (duration_ms == 0 || duration_ms > MAXIMUM_LIFETIME_MS)
164        {
165            return Err(MonitorError::InvalidLifetime);
166        }
167        match (self.condition.requires_polling(), self.check_interval_ms) {
168            (true, Some(interval)) => validate_schedule(interval),
169            (true, None) => Err(MonitorError::MissingCheckSchedule),
170            (false, Some(_)) => Err(MonitorError::UnexpectedCheckSchedule),
171            (false, None) => Ok(()),
172        }
173    }
174}
175
176#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
177#[serde(tag = "kind", rename_all = "snake_case")]
178pub enum MonitorOperation {
179    Add {
180        definition: MonitorDefinition,
181    },
182    Update {
183        monitor_id: String,
184        definition: MonitorDefinition,
185    },
186    Pause {
187        monitor_id: String,
188    },
189    Resume {
190        monitor_id: String,
191    },
192    Remove {
193        monitor_id: String,
194    },
195}
196
197impl MonitorOperation {
198    pub fn validate(&self) -> Result<(), MonitorError> {
199        match self {
200            Self::Add { definition } => definition.validate(),
201            Self::Update {
202                monitor_id,
203                definition,
204            } => {
205                validate_monitor_id(monitor_id)?;
206                definition.validate()
207            }
208            Self::Pause { monitor_id }
209            | Self::Resume { monitor_id }
210            | Self::Remove { monitor_id } => validate_monitor_id(monitor_id),
211        }
212    }
213}
214
215#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
216#[serde(rename_all = "snake_case")]
217pub enum MonitorState {
218    Active,
219    Paused,
220    Matched,
221    Degraded,
222}
223
224#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
225#[serde(rename_all = "snake_case")]
226pub enum EventReason {
227    Matched,
228    StateChanged,
229    SessionExit,
230    Check,
231    Interval,
232    Expired,
233    Removed,
234    Paused,
235    Resumed,
236    Updated,
237}
238
239#[derive(Clone, Copy, Debug, Eq, PartialEq)]
240pub enum Observation {
241    Check,
242    Output,
243    Screen,
244    Quiet,
245    Exit,
246    SessionExit,
247}
248
249#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
250pub struct MonitorRuntime {
251    pub state: MonitorState,
252    pub generation: u64,
253    pub created_at_ms: i64,
254    pub lifetime_deadline_ms: Option<i64>,
255    pub next_check_ms: Option<i64>,
256    pub next_notification_ms: Option<i64>,
257    pub check_count: u64,
258    pub notification_count: u64,
259    pub last_event_id: u64,
260    pub last_event_reason: Option<EventReason>,
261    pub condition_matched: bool,
262    pub path_baseline: Option<PathBaseline>,
263    pub probe_cwd_fingerprint: Option<[u8; 32]>,
264    matcher_states: [u64; PATTERN_WORDS],
265}
266
267#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
268pub struct PathBaseline {
269    pub exists: bool,
270    pub size: u64,
271    pub modified_ns: i128,
272}
273
274#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
275pub struct Monitor {
276    pub monitor_id: String,
277    pub definition: MonitorDefinition,
278    pub runtime: MonitorRuntime,
279}
280
281#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
282pub struct Decision {
283    pub notify: Option<EventReason>,
284    pub remove: bool,
285    pub state_changed: bool,
286}
287
288impl Monitor {
289    pub fn new(
290        monitor_id: impl Into<String>,
291        definition: MonitorDefinition,
292        now_ms: i64,
293    ) -> Result<Self, MonitorError> {
294        definition.validate()?;
295        if now_ms < 0 {
296            return Err(MonitorError::InvalidClock);
297        }
298        let monitor_id = monitor_id.into();
299        validate_monitor_id(&monitor_id)?;
300        let lifetime_deadline_ms = match definition.lifetime {
301            MonitorLifetime::Duration { duration_ms } => Some(deadline(now_ms, duration_ms)?),
302            MonitorLifetime::UntilMatch | MonitorLifetime::UntilSessionEnd => None,
303        };
304        let next_check_ms = match (&definition.condition, definition.check_interval_ms) {
305            (_, Some(interval)) => Some(deadline(now_ms, interval)?),
306            (MonitorCondition::OutputQuiet { duration_ms }, None) => {
307                Some(deadline(now_ms, *duration_ms)?)
308            }
309            _ => None,
310        };
311        let next_notification_ms = match definition.notify {
312            NotifySchedule::Interval { interval_ms } => Some(deadline(now_ms, interval_ms)?),
313            _ => None,
314        };
315        Ok(Self {
316            monitor_id,
317            definition,
318            runtime: MonitorRuntime {
319                state: MonitorState::Active,
320                generation: 1,
321                created_at_ms: now_ms,
322                lifetime_deadline_ms,
323                next_check_ms,
324                next_notification_ms,
325                check_count: 0,
326                notification_count: 0,
327                last_event_id: 0,
328                last_event_reason: None,
329                condition_matched: false,
330                path_baseline: None,
331                probe_cwd_fingerprint: None,
332                matcher_states: [0; PATTERN_WORDS],
333            },
334        })
335    }
336
337    pub fn observe(
338        &mut self,
339        observation: Observation,
340        condition_matches: bool,
341        now_ms: i64,
342    ) -> Result<Decision, MonitorError> {
343        if matches!(
344            self.runtime.state,
345            MonitorState::Paused | MonitorState::Degraded
346        ) {
347            return Ok(Decision::default());
348        }
349        if self.expired(now_ms)? {
350            return Ok(Decision {
351                notify: self.state_notification(EventReason::Expired),
352                remove: true,
353                state_changed: true,
354            });
355        }
356        if observation == Observation::SessionExit {
357            return Ok(Decision {
358                notify: if self.definition.notify == NotifySchedule::OnExit {
359                    Some(EventReason::SessionExit)
360                } else {
361                    self.state_notification(EventReason::SessionExit)
362                },
363                remove: true,
364                state_changed: true,
365            });
366        }
367        self.runtime.check_count = self
368            .runtime
369            .check_count
370            .checked_add(1)
371            .ok_or(MonitorError::CounterOverflow)?;
372        if observation == Observation::Check
373            && let Some(interval) = self.definition.check_interval_ms
374        {
375            let due = self.runtime.next_check_ms.unwrap_or(now_ms);
376            self.runtime.next_check_ms = Some(advance_deadline(due, interval, now_ms)?);
377        }
378        if observation == Observation::Quiet
379            && let MonitorCondition::OutputQuiet { duration_ms } = self.definition.condition
380        {
381            self.runtime.next_check_ms = Some(deadline(now_ms, duration_ms)?);
382        }
383
384        let newly_matched = condition_matches && !self.runtime.condition_matched;
385        if newly_matched {
386            self.runtime.condition_matched = true;
387            self.runtime.state = MonitorState::Matched;
388        }
389        let notify = match self.definition.notify {
390            NotifySchedule::OnMatch if newly_matched => Some(EventReason::Matched),
391            NotifySchedule::OnStateChange if newly_matched => Some(EventReason::StateChanged),
392            NotifySchedule::EveryCheck => Some(EventReason::Check),
393            NotifySchedule::EveryNChecks { count }
394                if self.runtime.check_count.is_multiple_of(u64::from(count)) =>
395            {
396                Some(EventReason::Check)
397            }
398            _ => None,
399        };
400        Ok(Decision {
401            notify,
402            remove: newly_matched && self.definition.lifetime == MonitorLifetime::UntilMatch,
403            state_changed: newly_matched,
404        })
405    }
406
407    pub fn timer_decision(&mut self, now_ms: i64) -> Result<Decision, MonitorError> {
408        if self.expired(now_ms)? {
409            return Ok(Decision {
410                notify: self.state_notification(EventReason::Expired),
411                remove: true,
412                state_changed: true,
413            });
414        }
415        if matches!(
416            self.runtime.state,
417            MonitorState::Paused | MonitorState::Degraded
418        ) {
419            return Ok(Decision::default());
420        }
421        let NotifySchedule::Interval { interval_ms } = self.definition.notify else {
422            return Ok(Decision::default());
423        };
424        let due = self
425            .runtime
426            .next_notification_ms
427            .ok_or(MonitorError::InvalidState)?;
428        if now_ms < due {
429            return Ok(Decision::default());
430        }
431        self.runtime.next_notification_ms = Some(advance_deadline(due, interval_ms, now_ms)?);
432        Ok(Decision {
433            notify: Some(EventReason::Interval),
434            ..Decision::default()
435        })
436    }
437
438    pub fn note_output(&mut self, now_ms: i64) -> Result<(), MonitorError> {
439        if let MonitorCondition::OutputQuiet { duration_ms } = self.definition.condition {
440            self.runtime.next_check_ms = Some(deadline(now_ms, duration_ms)?);
441        }
442        Ok(())
443    }
444
445    pub fn quiet_due(&self, now_ms: i64) -> bool {
446        self.runtime.state != MonitorState::Paused
447            && matches!(
448                self.definition.condition,
449                MonitorCondition::OutputQuiet { .. }
450            )
451            && self.runtime.next_check_ms.is_some_and(|due| now_ms >= due)
452    }
453
454    pub fn polling_due(&self, now_ms: i64) -> bool {
455        !matches!(
456            self.runtime.state,
457            MonitorState::Paused | MonitorState::Degraded
458        ) && self.definition.condition.requires_polling()
459            && self.runtime.next_check_ms.is_some_and(|due| now_ms >= due)
460    }
461
462    pub fn next_deadline(&self) -> Option<i64> {
463        let mut result = self.runtime.lifetime_deadline_ms;
464        if !matches!(
465            self.runtime.state,
466            MonitorState::Paused | MonitorState::Degraded
467        ) {
468            result = earlier(result, self.runtime.next_check_ms);
469            result = earlier(result, self.runtime.next_notification_ms);
470        }
471        result
472    }
473
474    pub fn pause(&mut self) -> bool {
475        if matches!(
476            self.runtime.state,
477            MonitorState::Paused | MonitorState::Degraded
478        ) {
479            return false;
480        }
481        self.runtime.state = MonitorState::Paused;
482        true
483    }
484
485    pub fn resume(&mut self, now_ms: i64) -> Result<bool, MonitorError> {
486        if self.runtime.state != MonitorState::Paused {
487            return Ok(false);
488        }
489        self.runtime.state = if self.runtime.condition_matched {
490            MonitorState::Matched
491        } else {
492            MonitorState::Active
493        };
494        if let Some(interval) = self.definition.check_interval_ms {
495            self.runtime.next_check_ms = Some(deadline(now_ms, interval)?);
496        } else if let MonitorCondition::OutputQuiet { duration_ms } = self.definition.condition {
497            self.runtime.next_check_ms = Some(deadline(now_ms, duration_ms)?);
498        }
499        if let NotifySchedule::Interval { interval_ms } = self.definition.notify {
500            self.runtime.next_notification_ms = Some(deadline(now_ms, interval_ms)?);
501        }
502        Ok(true)
503    }
504
505    pub fn degrade_for_raw_gap(&mut self) -> Result<bool, MonitorError> {
506        if !self.definition.condition.is_gap_sensitive()
507            || self.runtime.state == MonitorState::Degraded
508        {
509            return Ok(false);
510        }
511        self.runtime.state = MonitorState::Degraded;
512        self.runtime.matcher_states.fill(0);
513        self.runtime.next_check_ms = None;
514        self.runtime.next_notification_ms = None;
515        self.bump_generation()?;
516        Ok(true)
517    }
518
519    pub fn bump_generation(&mut self) -> Result<(), MonitorError> {
520        self.runtime.generation = self
521            .runtime
522            .generation
523            .checked_add(1)
524            .ok_or(MonitorError::CounterOverflow)?;
525        Ok(())
526    }
527
528    pub fn note_notification(
529        &mut self,
530        event_id: u64,
531        reason: EventReason,
532    ) -> Result<(), MonitorError> {
533        if event_id == 0 || event_id <= self.runtime.last_event_id {
534            return Err(MonitorError::InvalidEventId);
535        }
536        self.runtime.last_event_id = event_id;
537        self.runtime.last_event_reason = Some(reason);
538        self.runtime.notification_count = self
539            .runtime
540            .notification_count
541            .checked_add(1)
542            .ok_or(MonitorError::CounterOverflow)?;
543        Ok(())
544    }
545
546    pub fn feed_output(&mut self, bytes: &[u8]) -> Result<bool, MonitorError> {
547        let (pattern, wildcard) = match &self.definition.condition {
548            MonitorCondition::OutputContains { pattern } => (pattern.as_bytes(), false),
549            MonitorCondition::OutputMatches { pattern } => (pattern.as_bytes(), true),
550            _ => return Ok(false),
551        };
552        pattern_feed(pattern, wildcard, &mut self.runtime.matcher_states, bytes)
553    }
554
555    pub fn validate(&self) -> Result<(), MonitorError> {
556        validate_monitor_id(&self.monitor_id)?;
557        self.definition.validate()?;
558        if self.runtime.generation == 0
559            || self.runtime.created_at_ms < 0
560            || (self.runtime.last_event_id > 0 && self.runtime.notification_count == 0)
561            || ((self.runtime.last_event_id == 0) != self.runtime.last_event_reason.is_none())
562            || (self.runtime.condition_matched && self.runtime.state == MonitorState::Active)
563        {
564            return Err(MonitorError::InvalidState);
565        }
566        if self
567            .runtime
568            .lifetime_deadline_ms
569            .is_some_and(|deadline| deadline <= self.runtime.created_at_ms)
570        {
571            return Err(MonitorError::InvalidState);
572        }
573        Ok(())
574    }
575
576    fn expired(&self, now_ms: i64) -> Result<bool, MonitorError> {
577        if now_ms < 0 {
578            return Err(MonitorError::InvalidClock);
579        }
580        Ok(self
581            .runtime
582            .lifetime_deadline_ms
583            .is_some_and(|deadline| now_ms >= deadline))
584    }
585
586    fn state_notification(&self, reason: EventReason) -> Option<EventReason> {
587        (self.definition.notify == NotifySchedule::OnStateChange).then_some(reason)
588    }
589}
590
591pub fn stable_id(sequence: u64) -> Result<String, MonitorError> {
592    if sequence == 0 {
593        return Err(MonitorError::IdExhausted);
594    }
595    Ok(format!("monitor-{sequence}"))
596}
597
598pub fn pattern_matches(pattern: &[u8], wildcard: bool, bytes: &[u8]) -> Result<bool, MonitorError> {
599    let mut states = [0; PATTERN_WORDS];
600    pattern_feed(pattern, wildcard, &mut states, bytes)
601}
602
603fn pattern_feed(
604    pattern: &[u8],
605    wildcard: bool,
606    states: &mut [u64; PATTERN_WORDS],
607    bytes: &[u8],
608) -> Result<bool, MonitorError> {
609    if pattern.is_empty() || pattern.len() > MAXIMUM_PATTERN_BYTES {
610        return Err(MonitorError::InvalidPattern);
611    }
612    set_bit(states, 0);
613    epsilon_closure(pattern, wildcard, states);
614    for byte in bytes {
615        let mut next = [0; PATTERN_WORDS];
616        set_bit(&mut next, 0);
617        for (index, token) in pattern.iter().copied().enumerate() {
618            if !bit_is_set(states, index) {
619                continue;
620            }
621            if wildcard && token == b'*' {
622                set_bit(&mut next, index);
623            } else if token == *byte || (wildcard && token == b'?') {
624                set_bit(&mut next, index + 1);
625            }
626        }
627        epsilon_closure(pattern, wildcard, &mut next);
628        *states = next;
629        if bit_is_set(states, pattern.len()) {
630            return Ok(true);
631        }
632    }
633    Ok(bit_is_set(states, pattern.len()))
634}
635
636fn validate_schedule(interval_ms: u64) -> Result<(), MonitorError> {
637    if !(MINIMUM_SCHEDULE_MS..=MAXIMUM_SCHEDULE_MS).contains(&interval_ms) {
638        return Err(MonitorError::InvalidSchedule);
639    }
640    Ok(())
641}
642
643fn validate_monitor_id(value: &str) -> Result<(), MonitorError> {
644    valid_text(
645        value,
646        MAXIMUM_MONITOR_ID_BYTES,
647        MonitorError::InvalidMonitor,
648    )
649}
650
651fn valid_text(value: &str, maximum: usize, error: MonitorError) -> Result<(), MonitorError> {
652    if value.is_empty() || value.len() > maximum || value.contains('\0') {
653        return Err(error);
654    }
655    Ok(())
656}
657
658fn deadline(now_ms: i64, duration_ms: u64) -> Result<i64, MonitorError> {
659    let duration = i64::try_from(duration_ms).map_err(|_| MonitorError::DeadlineOverflow)?;
660    now_ms
661        .checked_add(duration)
662        .ok_or(MonitorError::DeadlineOverflow)
663}
664
665fn advance_deadline(current: i64, interval_ms: u64, now_ms: i64) -> Result<i64, MonitorError> {
666    let interval = i64::try_from(interval_ms).map_err(|_| MonitorError::DeadlineOverflow)?;
667    if interval <= 0 {
668        return Err(MonitorError::InvalidSchedule);
669    }
670    if current > now_ms {
671        return Ok(current);
672    }
673    let elapsed = now_ms
674        .checked_sub(current)
675        .ok_or(MonitorError::DeadlineOverflow)?;
676    let steps = elapsed / interval + 1;
677    current
678        .checked_add(
679            steps
680                .checked_mul(interval)
681                .ok_or(MonitorError::DeadlineOverflow)?,
682        )
683        .ok_or(MonitorError::DeadlineOverflow)
684}
685
686fn earlier(left: Option<i64>, right: Option<i64>) -> Option<i64> {
687    match (left, right) {
688        (None, value) | (value, None) => value,
689        (Some(left), Some(right)) => Some(left.min(right)),
690    }
691}
692
693fn epsilon_closure(pattern: &[u8], wildcard: bool, states: &mut [u64; PATTERN_WORDS]) {
694    if !wildcard {
695        return;
696    }
697    for (index, token) in pattern.iter().copied().enumerate() {
698        if token == b'*' && bit_is_set(states, index) {
699            set_bit(states, index + 1);
700        }
701    }
702}
703
704fn set_bit(states: &mut [u64; PATTERN_WORDS], index: usize) {
705    states[index / 64] |= 1_u64 << (index % 64);
706}
707
708fn bit_is_set(states: &[u64; PATTERN_WORDS], index: usize) -> bool {
709    states[index / 64] & (1_u64 << (index % 64)) != 0
710}
711
712#[derive(Clone, Copy, Debug, Eq, PartialEq)]
713pub enum MonitorError {
714    InvalidCondition,
715    InvalidLifetime,
716    InvalidSchedule,
717    MissingCheckSchedule,
718    UnexpectedCheckSchedule,
719    InvalidMonitor,
720    InvalidState,
721    InvalidClock,
722    InvalidPattern,
723    InvalidEventId,
724    CounterOverflow,
725    DeadlineOverflow,
726    IdExhausted,
727}
728
729impl std::fmt::Display for MonitorError {
730    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
731        write!(
732            formatter,
733            "{}",
734            match self {
735                Self::InvalidCondition => "invalid monitor condition",
736                Self::InvalidLifetime => "invalid monitor lifetime",
737                Self::InvalidSchedule => "invalid monitor schedule",
738                Self::MissingCheckSchedule => "polling monitor requires check_interval_ms",
739                Self::UnexpectedCheckSchedule =>
740                    "event-driven monitor cannot set check_interval_ms",
741                Self::InvalidMonitor => "invalid monitor",
742                Self::InvalidState => "invalid monitor state",
743                Self::InvalidClock => "invalid monitor clock",
744                Self::InvalidPattern => "invalid monitor pattern",
745                Self::InvalidEventId => "invalid monitor event id",
746                Self::CounterOverflow => "monitor counter overflow",
747                Self::DeadlineOverflow => "monitor deadline overflow",
748                Self::IdExhausted => "monitor id exhausted",
749            }
750        )
751    }
752}
753
754impl std::error::Error for MonitorError {}
755
756#[cfg(test)]
757mod tests {
758    use super::*;
759
760    fn definition(
761        condition: MonitorCondition,
762        notify: NotifySchedule,
763        lifetime: MonitorLifetime,
764    ) -> MonitorDefinition {
765        let check_interval_ms = condition.requires_polling().then_some(25);
766        MonitorDefinition {
767            condition,
768            check_interval_ms,
769            notify,
770            lifetime,
771        }
772    }
773
774    #[test]
775    fn bounded_matcher_preserves_literal_and_wildcard_state_across_chunks() {
776        let mut literal = Monitor::new(
777            "monitor-1",
778            definition(
779                MonitorCondition::OutputContains {
780                    pattern: "ready".into(),
781                },
782                NotifySchedule::OnMatch,
783                MonitorLifetime::UntilMatch,
784            ),
785            0,
786        )
787        .unwrap();
788        assert!(!literal.feed_output(b"re").unwrap());
789        assert!(literal.feed_output(b"ady").unwrap());
790
791        let mut wildcard = Monitor::new(
792            "monitor-2",
793            definition(
794                MonitorCondition::OutputMatches {
795                    pattern: "sta*t?d".into(),
796                },
797                NotifySchedule::OnMatch,
798                MonitorLifetime::UntilMatch,
799            ),
800            0,
801        )
802        .unwrap();
803        assert!(!wildcard.feed_output(b"xxsta").unwrap());
804        assert!(!wildcard.feed_output(b"ble-t").unwrap());
805        assert!(wildcard.feed_output(b"edyy").unwrap());
806        assert!(pattern_matches(b"*status?ok*", true, b"prefix status-ok suffix").unwrap());
807    }
808
809    #[test]
810    fn validation_rejects_schedule_floods_mismatches_and_overflow() {
811        let mut polling = definition(
812            MonitorCondition::PathExists {
813                path: "/workspace/ready".into(),
814            },
815            NotifySchedule::OnMatch,
816            MonitorLifetime::UntilMatch,
817        );
818        polling.check_interval_ms = Some(1);
819        assert_eq!(polling.validate(), Err(MonitorError::InvalidSchedule));
820        polling.check_interval_ms = None;
821        assert_eq!(polling.validate(), Err(MonitorError::MissingCheckSchedule));
822
823        let event = MonitorDefinition {
824            condition: MonitorCondition::ProcessExit,
825            check_interval_ms: Some(25),
826            notify: NotifySchedule::OnMatch,
827            lifetime: MonitorLifetime::UntilMatch,
828        };
829        assert_eq!(event.validate(), Err(MonitorError::UnexpectedCheckSchedule));
830        assert_eq!(deadline(i64::MAX, 1), Err(MonitorError::DeadlineOverflow));
831    }
832
833    #[test]
834    fn decisions_keep_match_exit_interval_and_lifetime_independent() {
835        let mut matched = Monitor::new(
836            "monitor-1",
837            definition(
838                MonitorCondition::ProcessExit,
839                NotifySchedule::OnMatch,
840                MonitorLifetime::UntilMatch,
841            ),
842            100,
843        )
844        .unwrap();
845        assert_eq!(
846            matched.observe(Observation::Exit, true, 110).unwrap(),
847            Decision {
848                notify: Some(EventReason::Matched),
849                remove: true,
850                state_changed: true,
851            }
852        );
853
854        let mut on_exit = Monitor::new(
855            "monitor-2",
856            definition(
857                MonitorCondition::ProcessExit,
858                NotifySchedule::OnExit,
859                MonitorLifetime::UntilSessionEnd,
860            ),
861            100,
862        )
863        .unwrap();
864        assert_eq!(
865            on_exit
866                .observe(Observation::SessionExit, false, 110)
867                .unwrap()
868                .notify,
869            Some(EventReason::SessionExit)
870        );
871
872        let mut interval = Monitor::new(
873            "monitor-3",
874            definition(
875                MonitorCondition::ProcessExit,
876                NotifySchedule::Interval { interval_ms: 25 },
877                MonitorLifetime::UntilSessionEnd,
878            ),
879            100,
880        )
881        .unwrap();
882        assert_eq!(
883            interval.timer_decision(125).unwrap().notify,
884            Some(EventReason::Interval)
885        );
886    }
887
888    #[test]
889    fn quiet_pause_resume_event_dedup_and_expiry_are_deterministic() {
890        let mut quiet = Monitor::new(
891            "monitor-9",
892            definition(
893                MonitorCondition::OutputQuiet { duration_ms: 50 },
894                NotifySchedule::OnStateChange,
895                MonitorLifetime::Duration { duration_ms: 100 },
896            ),
897            1_000,
898        )
899        .unwrap();
900        quiet.note_output(1_025).unwrap();
901        assert!(!quiet.quiet_due(1_074));
902        assert!(quiet.quiet_due(1_075));
903        assert!(quiet.pause());
904        assert_eq!(quiet.next_deadline(), Some(1_100));
905        assert!(quiet.resume(1_080).unwrap());
906        quiet.note_notification(4, EventReason::Matched).unwrap();
907        assert_eq!(
908            quiet.note_notification(4, EventReason::Matched),
909            Err(MonitorError::InvalidEventId)
910        );
911        assert!(quiet.timer_decision(1_100).unwrap().remove);
912        assert_eq!(stable_id(42).unwrap(), "monitor-42");
913    }
914
915    #[test]
916    fn raw_gaps_degrade_output_monitors_but_not_process_monitors() {
917        let mut output = Monitor::new(
918            "monitor-gap",
919            definition(
920                MonitorCondition::OutputContains {
921                    pattern: "needle".into(),
922                },
923                NotifySchedule::OnMatch,
924                MonitorLifetime::UntilSessionEnd,
925            ),
926            100,
927        )
928        .unwrap();
929        assert!(output.degrade_for_raw_gap().unwrap());
930        assert_eq!(output.runtime.state, MonitorState::Degraded);
931        assert_eq!(output.runtime.generation, 2);
932
933        let mut process = Monitor::new(
934            "monitor-process",
935            definition(
936                MonitorCondition::ProcessExit,
937                NotifySchedule::OnMatch,
938                MonitorLifetime::UntilSessionEnd,
939            ),
940            100,
941        )
942        .unwrap();
943        assert!(!process.degrade_for_raw_gap().unwrap());
944        assert_eq!(process.runtime.state, MonitorState::Active);
945    }
946
947    #[test]
948    fn monitor_json_uses_the_public_flat_kind_vocabulary() {
949        let definition = definition(
950            MonitorCondition::TcpReady {
951                host: "127.0.0.1".into(),
952                port: 3_000,
953            },
954            NotifySchedule::EveryNChecks { count: 2 },
955            MonitorLifetime::Duration { duration_ms: 500 },
956        );
957        let value = serde_json::to_value(&definition).unwrap();
958        assert_eq!(value["condition"]["kind"], "tcp_ready");
959        assert_eq!(value["condition"]["port"], 3_000);
960        assert_eq!(value["notify"]["kind"], "every_n_checks");
961        assert_eq!(value["lifetime"]["kind"], "duration");
962        assert_eq!(
963            serde_json::from_value::<MonitorDefinition>(value).unwrap(),
964            definition
965        );
966    }
967}
fxrs_process/monitor.rs

fxrs_process/
monitor.rs
