Skip to main content

fno_agents/
lib.rs

1//! `fno-agents` substrate crate (Phase 6, ab-a09e1eaf).
2//!
3//! This crate is the Rust substrate for PTY-managed agents (codex / gemini /
4//! future OpenCode). It is split per the design's Locked Decisions:
5//!
6//! - shared types (this module): [`ShortId`], [`AgentStatus`], [`ParsedEvent`]
7//!   (LD9, sealed enum), [`MonotonicTimestamp`] (count-during-sleep clock).
8//! - [`pty`]: PTY spawn + bounded-ring output drainer (LD31).
9//! - [`write_queue`]: bounded-backpressure stdin queue + [`write_queue::WriteMsg`].
10//! - [`supervisor`]: [`supervisor::RestartPolicy`] state machine + hard ceiling (LD36).
11//! - [`readiness`]: [`readiness::ReadinessDetector`] trait + `UnknownReadinessSignal`
12//!   (Open Question #9: no generic byte-count fallback; per-CLI signal mandatory).
13//!
14//! ## Scope of Wave 1 (this PR)
15//!
16//! Wave 0's smoke prototype (`cli/scripts/smoke/pty-survival/`) refuted the
17//! "direct daemon-owned PTY survives daemon restart" assertion: a child on a
18//! PTY whose master the supervisor owns is SIGHUP'd and dies the instant the
19//! master closes. The locked outcome (Outcome B) is a per-agent worker process
20//! that owns the master and outlives the daemon. The substrate in this crate
21//! is therefore written **worker-side**: [`pty::PtySession`] is what a worker
22//! owns; the daemon (Wave 3) reconnects to workers over their sockets.
23//!
24//! Deliberately deferred (documented seams, not gaps):
25//! - `alacritty_terminal` grid wiring + per-CLI [`readiness::ReadinessDetector`]
26//!   impls -> Wave 2, alongside the smoke captures that define the grid patterns
27//!   (the trait operates over [`readiness::ScreenView`] so Wave 2 only adds impls).
28//! - `tokio` runtime integration -> Wave 3 (the daemon is its only consumer; the
29//!   substrate stays runtime-agnostic and is driven from `spawn_blocking`).
30//!
31//! ## Scope of Wave 2 (this PR)
32//!
33//! Wave 2 fills the seams Wave 1 left:
34//! - [`provider`]: [`provider::Provider`] + [`provider::ProviderWithPty`] traits
35//!   (LD8) and the three impls ([`provider::ClaudeProvider`] shellout,
36//!   [`provider::CodexProvider`] / [`provider::GeminiProvider`] PTY-managed).
37//! - [`envelope`]: [`envelope::Envelope`] structural anti-injection wrapper (LD15).
38//! - [`screen`]: the terminal-grid construction behind [`readiness::ScreenView`]
39//!   (the per-CLI [`readiness::ReadinessDetector`] impls now live in
40//!   [`readiness`]).
41
42pub mod active_backlog;
43pub mod agents_config;
44pub mod claude_ask;
45pub mod client;
46pub mod client_verbs;
47pub mod codex_ask;
48pub mod daemon;
49pub mod drift;
50pub mod drive;
51pub mod drive_client;
52pub mod envelope;
53pub mod events;
54pub mod finalize;
55pub mod gemini_ask;
56pub mod grid;
57pub mod kill_criteria;
58pub mod logs;
59pub mod logs_client;
60pub mod loop_dispatch;
61pub mod loop_megatron;
62pub mod loop_megawalk;
63pub mod loop_runtime;
64pub mod loop_target;
65pub mod loopcheck;
66pub mod nudge;
67pub mod paths;
68pub mod protocol;
69pub mod provider;
70pub mod pty;
71pub mod readiness;
72pub mod screen;
73pub mod state;
74pub mod stream_worker;
75pub mod subprocess_ask;
76pub mod supervisor;
77pub mod verify_evidence;
78pub mod worker;
79pub mod write_queue;
80
81use serde::{Deserialize, Serialize};
82use std::time::Duration;
83
84/// A short, opaque agent identifier (e.g. `wkA`). Stored in the registry and
85/// used to name per-agent state directories. Validation is intentionally light
86/// at this layer; dispatch-layer validation (US1 invariant) owns argv rules.
87#[derive(Debug, thiserror::Error, PartialEq, Eq)]
88pub enum ShortIdError {
89    #[error("short id must be non-empty")]
90    Empty,
91}
92
93#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
94pub struct ShortId(pub(crate) String);
95
96impl ShortId {
97    /// Construct a short id. The field is crate-private and this is the only
98    /// constructor, so a zero-length registry key (which would collapse
99    /// per-agent state directory paths) cannot be built at any call site.
100    /// Charset rules beyond non-empty remain the dispatch layer's
101    /// responsibility (US1 argv validation).
102    pub fn new(s: impl Into<String>) -> Result<Self, ShortIdError> {
103        let s = s.into();
104        if s.is_empty() {
105            return Err(ShortIdError::Empty);
106        }
107        Ok(ShortId(s))
108    }
109
110    pub fn as_str(&self) -> &str {
111        &self.0
112    }
113}
114
115impl std::fmt::Display for ShortId {
116    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
117        f.write_str(&self.0)
118    }
119}
120
121/// Agent lifecycle status. `state.status` is canonical; `registry.status` is a
122/// denormalized projection of it (LD10). Serialized snake_case for the JSON
123/// state files and the cross-language schemas.
124#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
125#[serde(rename_all = "snake_case")]
126pub enum AgentStatus {
127    /// PTY spawned, not yet confirmed ready for input.
128    Spawning,
129    /// Confirmed ready for input (readiness_detector reported ready).
130    Ready,
131    /// Alive and waiting (equivalent to `Ready` for drive-eligibility, LD28).
132    Idle,
133    /// Mid-reply / actively processing.
134    Busy,
135    /// Live shorthand used by the registry projection.
136    Live,
137    /// Restart policy is backing off before re-spawn.
138    Restarting,
139    /// Reachability probe failed; needs reconcile or rm.
140    Orphaned,
141    /// Per-agent task panicked (provider parse panic, etc.); restart policy applies.
142    Failed,
143    /// Child exited; registry entry retained until rm.
144    Exited,
145    /// Restart hard ceiling hit (LD36); will not restart again.
146    PermanentDead,
147}
148
149impl AgentStatus {
150    /// Drive is accepted only for these statuses (LD28). `Idle`/`Live` are
151    /// equivalent to `Ready` for drive purposes.
152    pub fn is_drive_eligible(&self) -> bool {
153        matches!(
154            self,
155            AgentStatus::Ready | AgentStatus::Idle | AgentStatus::Busy | AgentStatus::Live
156        )
157    }
158}
159
160/// Sealed event vocabulary every provider parses INTO (LD9). Variant additions
161/// are a one-line crate-wide change; no per-provider enums. `#[serde(tag="kind")]`
162/// matches the wire shape in the design's Architecture section.
163#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
164#[serde(tag = "kind", rename_all = "snake_case")]
165pub enum ParsedEvent {
166    SessionCreated {
167        session_id: String,
168    },
169    OutputChunk {
170        text: String,
171    },
172    ReplyComplete {
173        text: String,
174        duration_ms: u64,
175    },
176    ToolUse {
177        name: String,
178        args: Option<serde_json::Value>,
179    },
180    ProviderError {
181        message: String,
182    },
183    /// A line the provider's parser did not recognize. Tee'd to timeline.jsonl
184    /// as `unknown_stream_event` rather than dropped, so a provider version bump
185    /// degrades gracefully (Silent-Failure-Hunter finding).
186    Unknown {
187        raw: String,
188    },
189}
190
191/// A monotonic timestamp that **counts during system sleep**, used for
192/// drive-window heartbeat math (LD17 + Domain Pitfall: macOS/Linux suspend
193/// divergence).
194///
195/// Rust's `std::time::Instant` is inconsistent across platforms for the
196/// sleep case: on macOS it uses `mach_continuous_time` (counts sleep), on
197/// Linux it uses `CLOCK_MONOTONIC` (does NOT count sleep). A laptop-sleep
198/// during a drive window must EXPIRE the window, so we standardize on the
199/// count-during-sleep semantic on both:
200///
201/// - Linux: `clock_gettime(CLOCK_BOOTTIME)`.
202/// - macOS: `mach_continuous_time()` converted to ns via `mach_timebase_info`.
203///
204/// Stored as nanoseconds since an unspecified epoch; only differences are
205/// meaningful. Wall-clock `ts` for human audit lives in events.jsonl, tracked
206/// independently (LD17).
207#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
208pub struct MonotonicTimestamp(u64);
209
210impl MonotonicTimestamp {
211    /// Read the current count-during-sleep monotonic clock.
212    pub fn now() -> Self {
213        MonotonicTimestamp(raw_monotonic_nanos())
214    }
215
216    /// Nanoseconds elapsed since `earlier`. Saturates at 0 if `earlier` is in
217    /// the future (clock readings are monotonic, so this only guards against a
218    /// caller passing a later timestamp as `earlier`).
219    pub fn duration_since(&self, earlier: MonotonicTimestamp) -> Duration {
220        Duration::from_nanos(self.0.saturating_sub(earlier.0))
221    }
222
223    /// Convenience: elapsed since this timestamp until now.
224    pub fn elapsed(&self) -> Duration {
225        MonotonicTimestamp::now().duration_since(*self)
226    }
227
228    /// Raw nanoseconds, for persisting the heartbeat baseline to state.json.
229    pub fn as_nanos(&self) -> u64 {
230        self.0
231    }
232
233    /// Reconstruct from raw nanoseconds previously read via [`as_nanos`]. Used
234    /// by the daemon (Wave 3) to restore a persisted heartbeat baseline. Only
235    /// meaningful when paired with a `now()` from the same daemon incarnation's
236    /// clock (the value is epoch-relative to the running clock).
237    ///
238    /// [`as_nanos`]: MonotonicTimestamp::as_nanos
239    pub fn from_nanos(nanos: u64) -> Self {
240        MonotonicTimestamp(nanos)
241    }
242}
243
244#[cfg(target_os = "linux")]
245fn raw_monotonic_nanos() -> u64 {
246    // CLOCK_BOOTTIME includes time spent suspended (unlike CLOCK_MONOTONIC).
247    let mut ts = libc::timespec {
248        tv_sec: 0,
249        tv_nsec: 0,
250    };
251    // SAFETY: `ts` is a valid, owned timespec; CLOCK_BOOTTIME is a valid clock
252    // id on Linux >= 2.6.39.
253    let rc = unsafe { libc::clock_gettime(libc::CLOCK_BOOTTIME, &mut ts) };
254    if rc != 0 {
255        // clock_gettime on a standard clock id effectively never fails on a
256        // supported kernel, so treat it as a should-be-impossible fault and
257        // make it LOUD rather than silent. Returning 0 is NOT a universal
258        // fail-safe: if a *baseline* read failed, elapsed over-reports (window
259        // expires early - safe); if a *current* read fails, elapsed under-
260        // reports toward 0 (window could hang open - unsafe). We accept that
261        // residual risk only because the failure cannot occur in practice, and
262        // log so it never passes unnoticed.
263        tracing::error!("clock_gettime(CLOCK_BOOTTIME) failed; monotonic reading degraded to 0");
264        return 0;
265    }
266    (ts.tv_sec as u64)
267        .saturating_mul(1_000_000_000)
268        .saturating_add(ts.tv_nsec.max(0) as u64)
269}
270
271#[cfg(target_os = "macos")]
272fn raw_monotonic_nanos() -> u64 {
273    // mach_continuous_time() counts during sleep; convert mach ticks -> ns via
274    // the timebase ratio (1/1 on current Apple hardware, but we must not assume
275    // it). `libc` deprecated its mach timebase helpers and dropped
276    // mach_continuous_time entirely (it lives in the `mach2` crate now), so we
277    // declare the two libSystem symbols directly to avoid a macOS-only crate
278    // dependency. Both are part of libSystem, linked by default on macOS.
279    #[repr(C)]
280    struct MachTimebaseInfo {
281        numer: u32,
282        denom: u32,
283    }
284    extern "C" {
285        fn mach_continuous_time() -> u64;
286        fn mach_timebase_info(info: *mut MachTimebaseInfo) -> libc::c_int;
287    }
288    use std::sync::OnceLock;
289    static TIMEBASE: OnceLock<(u64, u64)> = OnceLock::new();
290    let (numer, denom) = *TIMEBASE.get_or_init(|| {
291        let mut info = MachTimebaseInfo { numer: 0, denom: 0 };
292        // SAFETY: `info` is a valid, owned, repr(C) struct matching the C ABI;
293        // mach_timebase_info fills it and returns a kern_return_t.
294        let rc = unsafe { mach_timebase_info(&mut info) };
295        if rc != 0 || info.denom == 0 {
296            (1, 1)
297        } else {
298            (info.numer as u64, info.denom as u64)
299        }
300    });
301    // SAFETY: no arguments; returns a monotonic tick count that counts sleep.
302    let ticks = unsafe { mach_continuous_time() };
303    // ns = ticks * numer / denom, computed in u128 to avoid overflow.
304    ((ticks as u128 * numer as u128) / denom as u128) as u64
305}
306
307#[cfg(not(any(target_os = "linux", target_os = "macos")))]
308fn raw_monotonic_nanos() -> u64 {
309    // Other POSIX targets are not shipped by Phase 6 (Windows is Phase 7+).
310    // Fall back to CLOCK_MONOTONIC so the crate still compiles for dev on
311    // such hosts; the suspend semantic is undefined there and not relied on.
312    let mut ts = libc::timespec {
313        tv_sec: 0,
314        tv_nsec: 0,
315    };
316    // SAFETY: valid owned timespec; CLOCK_MONOTONIC is POSIX-standard.
317    let rc = unsafe { libc::clock_gettime(libc::CLOCK_MONOTONIC, &mut ts) };
318    if rc != 0 {
319        tracing::error!("clock_gettime(CLOCK_MONOTONIC) failed; monotonic reading degraded to 0");
320        return 0;
321    }
322    (ts.tv_sec as u64)
323        .saturating_mul(1_000_000_000)
324        .saturating_add(ts.tv_nsec.max(0) as u64)
325}
326
327#[cfg(test)]
328mod tests {
329    use super::*;
330
331    #[test]
332    fn short_id_rejects_empty() {
333        assert_eq!(ShortId::new(""), Err(ShortIdError::Empty));
334        let ok = ShortId::new("wkA").unwrap();
335        assert_eq!(ok.as_str(), "wkA");
336    }
337
338    #[test]
339    fn agent_status_serde_roundtrip_is_snake_case() {
340        let json = serde_json::to_string(&AgentStatus::PermanentDead).unwrap();
341        assert_eq!(json, "\"permanent_dead\"");
342        let back: AgentStatus = serde_json::from_str(&json).unwrap();
343        assert_eq!(back, AgentStatus::PermanentDead);
344    }
345
346    #[test]
347    fn drive_eligibility_matches_ld28() {
348        assert!(AgentStatus::Ready.is_drive_eligible());
349        assert!(AgentStatus::Idle.is_drive_eligible());
350        assert!(AgentStatus::Busy.is_drive_eligible());
351        assert!(!AgentStatus::Restarting.is_drive_eligible());
352        assert!(!AgentStatus::Exited.is_drive_eligible());
353        assert!(!AgentStatus::PermanentDead.is_drive_eligible());
354    }
355
356    #[test]
357    fn parsed_event_tagged_serde() {
358        let ev = ParsedEvent::ReplyComplete {
359            text: "hi".into(),
360            duration_ms: 42,
361        };
362        let json = serde_json::to_string(&ev).unwrap();
363        assert!(json.contains("\"kind\":\"reply_complete\""));
364        let back: ParsedEvent = serde_json::from_str(&json).unwrap();
365        assert_eq!(ev, back);
366    }
367
368    #[test]
369    fn parsed_event_unknown_preserves_raw() {
370        let ev = ParsedEvent::Unknown {
371            raw: "{\"new_event\":1}".into(),
372        };
373        let json = serde_json::to_string(&ev).unwrap();
374        let back: ParsedEvent = serde_json::from_str(&json).unwrap();
375        assert_eq!(ev, back);
376    }
377
378    #[test]
379    fn monotonic_clock_is_nondecreasing_and_measures_elapsed() {
380        let t0 = MonotonicTimestamp::now();
381        std::thread::sleep(Duration::from_millis(20));
382        let t1 = MonotonicTimestamp::now();
383        assert!(t1 >= t0, "monotonic clock went backwards");
384        let elapsed = t1.duration_since(t0);
385        assert!(
386            elapsed >= Duration::from_millis(15),
387            "elapsed too small: {elapsed:?}"
388        );
389        assert!(
390            elapsed < Duration::from_secs(5),
391            "elapsed implausibly large: {elapsed:?}"
392        );
393    }
394
395    #[test]
396    fn duration_since_future_saturates_to_zero() {
397        let t0 = MonotonicTimestamp::now();
398        std::thread::sleep(Duration::from_millis(5));
399        let t1 = MonotonicTimestamp::now();
400        // Passing the later ts as `earlier` must not panic or underflow.
401        assert_eq!(t0.duration_since(t1), Duration::ZERO);
402    }
403
404    // ── cv-114f75cc: production emit-kind completeness guard ──────────────
405    // KNOWN_EVENT_KINDS is hand-maintained and feeds both the Branch B `kind`
406    // schema enum and the cross-language parity gate, so a new `.emit("foo")`
407    // whose kind was never added to the constant would silently drift those
408    // surfaces. This test scans every production call site and fails on drift.
409
410    #[test]
411    fn every_production_emit_kind_is_registered() {
412        use std::collections::BTreeSet;
413
414        let known: BTreeSet<&str> = KNOWN_EVENT_KINDS.iter().copied().collect();
415        let src_root = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src");
416
417        let mut files = Vec::new();
418        collect_rs_files(&src_root, &mut files);
419        assert!(!files.is_empty(), "found no .rs files under {src_root:?}");
420
421        let mut unregistered: Vec<String> = Vec::new();
422        let mut production_kinds: BTreeSet<String> = BTreeSet::new();
423        let mut scanned_calls = 0usize;
424        for file in &files {
425            let text = std::fs::read_to_string(file).expect("read source file");
426            // Production code only: truncate at the first `#[cfg(test)]` marker.
427            // Tests live at the bottom by Rust convention, so test fixtures like
428            // `.emit("tick")` are excluded. (Verified: every production emit in
429            // this crate precedes its file's first `#[cfg(test)]`.)
430            let prod = match text.find("#[cfg(test)]") {
431                Some(i) => &text[..i],
432                None => &text[..],
433            };
434            let file_name = file.file_name().unwrap().to_string_lossy();
435            for (kind, line) in scan_emit_kinds(prod) {
436                scanned_calls += 1;
437                production_kinds.insert(kind.clone());
438                if !known.contains(kind.as_str()) {
439                    unregistered.push(format!(
440                        "{file_name}:{line}: .emit(\"{kind}\") not in KNOWN_EVENT_KINDS"
441                    ));
442                }
443            }
444        }
445
446        assert!(
447            scanned_calls > 0,
448            "scanner found zero emit call sites - the scan pattern likely broke"
449        );
450
451        // cv-2801ed8a: enforce the truncation assumption rather than just
452        // documenting it. The scan above trusts that every production emit
453        // precedes its file's first `#[cfg(test)]`. Verify it: scan BELOW each
454        // boundary too, and require every kind found there to be either also
455        // emitted in production (so the registration guard above already saw
456        // it) or a known test-only fixture. A production-looking kind that
457        // lives only below a boundary would otherwise escape the guard
458        // silently. `production_kinds` must be complete across ALL files before
459        // this check (a kind can be production in one file and test-only in
460        // another), so this is a second pass.
461        //
462        // `tick`/`heartbeat` are test fixture emits; `foo`/`x` are `.emit(...)`
463        // examples inside doc comments in the test module that the byte-level
464        // scanner picks up. (Escaped `.emit(\"...\")` in the scanner self-check
465        // string is NOT matched: the char after `(` is a backslash, not `"`.)
466        const TEST_ONLY_EMIT_KINDS: &[&str] = &["tick", "heartbeat", "foo", "x"];
467        let test_only: BTreeSet<&str> = TEST_ONLY_EMIT_KINDS.iter().copied().collect();
468
469        let mut below_only: Vec<String> = Vec::new();
470        for file in &files {
471            let text = std::fs::read_to_string(file).expect("read source file");
472            let boundary = match text.find("#[cfg(test)]") {
473                Some(i) => i,
474                None => continue,
475            };
476            // scan_emit_kinds reports lines relative to its input slice; add the
477            // newline count before the boundary so the message points at the
478            // real file line.
479            let base_line = text[..boundary].bytes().filter(|&c| c == b'\n').count();
480            let file_name = file.file_name().unwrap().to_string_lossy();
481            for (kind, line) in scan_emit_kinds(&text[boundary..]) {
482                if production_kinds.contains(&kind) || test_only.contains(kind.as_str()) {
483                    continue;
484                }
485                below_only.push(format!(
486                    "{file_name}:{}: .emit(\"{kind}\") appears only below #[cfg(test)] \
487                     (not emitted in production, not a known test-only fixture)",
488                    base_line + line
489                ));
490            }
491        }
492
493        assert!(
494            below_only.is_empty(),
495            "emit kinds found only below a #[cfg(test)] boundary - the truncation \
496             assumption (all production emits precede the test module) may be \
497             violated. If a kind below is a real production emit, register it in \
498             KNOWN_EVENT_KINDS and move it above the test module; if it is \
499             test-only, add it to TEST_ONLY_EMIT_KINDS:\n  {}",
500            below_only.join("\n  ")
501        );
502
503        // Self-check: the scanner extracts a single-line `.emit(` kind, a
504        // multi-line `.emit_fields(` kind, AND a whitespace-before-paren
505        // `.emit (` kind (valid Rust), so a genuine unregistered kind cannot
506        // slip past this guard silently. Also asserts the reported line number.
507        let synthetic = "x.emit(\"agent_spawned\", &p);\n  y.emit_fields(\n    \"definitely_not_a_real_kind\", m);\n z.emit (\"another_fake_kind\");";
508        let scanned = scan_emit_kinds(synthetic);
509        assert!(
510            scanned.iter().any(|(k, l)| k == "agent_spawned" && *l == 1),
511            "scanner missed a single-line emit kind (or wrong line)"
512        );
513        assert!(
514            scanned
515                .iter()
516                .any(|(k, _)| k == "definitely_not_a_real_kind"),
517            "scanner missed a multi-line emit_fields kind"
518        );
519        assert!(
520            scanned.iter().any(|(k, _)| k == "another_fake_kind"),
521            "scanner missed a `.emit (` call with whitespace before the paren"
522        );
523        assert!(
524            !known.contains("definitely_not_a_real_kind") && !known.contains("another_fake_kind"),
525            "the synthetic drift kinds must not be real registered kinds"
526        );
527        assert!(
528            unregistered.is_empty(),
529            "production emit kinds missing from KNOWN_EVENT_KINDS:\n  {}",
530            unregistered.join("\n  ")
531        );
532    }
533
534    fn collect_rs_files(dir: &std::path::Path, out: &mut Vec<std::path::PathBuf>) {
535        let entries = match std::fs::read_dir(dir) {
536            Ok(e) => e,
537            Err(_) => return,
538        };
539        for entry in entries.flatten() {
540            let path = entry.path();
541            if path.is_dir() {
542                collect_rs_files(&path, out);
543            } else if path.extension().and_then(|e| e.to_str()) == Some("rs") {
544                out.push(path);
545            }
546        }
547    }
548
549    /// Extract `(kind, line)` for every `.emit` / `.emit_fields` call with a
550    /// string-literal kind. Whitespace tolerant on both sides: `.emit ("x")`
551    /// and a newline between `(` and the opening quote both parse (valid Rust).
552    /// A call whose first argument is not a string literal is skipped - the
553    /// kind is dynamic and not statically checkable. The line number (1-based)
554    /// is reported so a drift failure points straight at the offending call.
555    fn scan_emit_kinds(src: &str) -> Vec<(String, usize)> {
556        let bytes = src.as_bytes();
557        let mut kinds = Vec::new();
558        for needle in [".emit", ".emit_fields"] {
559            let nb = needle.as_bytes();
560            let mut from = 0usize;
561            while let Some(rel) = find_sub(&bytes[from..], nb) {
562                let pos = from + rel;
563                let mut j = pos + nb.len();
564                // `.emit` must not match inside `.emit_fields` (next char `_`).
565                if needle == ".emit" && j < bytes.len() && bytes[j] == b'_' {
566                    from = j;
567                    continue;
568                }
569                while j < bytes.len() && (bytes[j] as char).is_whitespace() {
570                    j += 1;
571                }
572                if j < bytes.len() && bytes[j] == b'(' {
573                    j += 1;
574                    while j < bytes.len() && (bytes[j] as char).is_whitespace() {
575                        j += 1;
576                    }
577                    if j < bytes.len() && bytes[j] == b'"' {
578                        let start = j + 1;
579                        let mut k = start;
580                        while k < bytes.len() && bytes[k] != b'"' {
581                            k += 1;
582                        }
583                        if k < bytes.len() {
584                            let kind = String::from_utf8_lossy(&bytes[start..k]).into_owned();
585                            let line = src[..pos].bytes().filter(|&c| c == b'\n').count() + 1;
586                            kinds.push((kind, line));
587                        }
588                    }
589                }
590                from = pos + nb.len();
591            }
592        }
593        kinds
594    }
595
596    fn find_sub(haystack: &[u8], needle: &[u8]) -> Option<usize> {
597        if needle.is_empty() || haystack.len() < needle.len() {
598            return None;
599        }
600        haystack.windows(needle.len()).position(|w| w == needle)
601    }
602}
603
604// ---------------------------------------------------------------------------
605// W7: Cross-language schema introspection
606// ---------------------------------------------------------------------------
607
608/// All real operator-facing event kinds emitted by the Rust supervisor.
609/// Excludes test-only kinds (tick, heartbeat).
610///
611/// This const is the authoritative list for `--emit-schema` output and must
612/// stay in sync with every `.emit(kind, ...)` / `.emit_fields(kind, ...)`
613/// call site in the crate. The parity check script compares this list against
614/// the Python side for global uniqueness.
615///
616/// **How to regenerate when adding a new event kind:**
617/// ```text
618/// grep -rn '\.emit\b\|\.emit_fields\b' crates/fno-agents/src/ \
619///   | grep -v '//' \
620///   | grep -oP '"[a-z_]+"' \
621///   | sort | uniq
622/// ```
623/// Then cross-check the output against this list. Test-only kinds (tick,
624/// heartbeat) and value fields (reason, backend, ...) will appear in the grep
625/// output; only include kinds that appear as the first string argument to an
626/// emit call in non-test production code.
627pub const KNOWN_EVENT_KINDS: &[&str] = &[
628    // Agent lifecycle (daemon-emitted)
629    "agent_spawned",
630    "agent_stopped",
631    "agent_exited",
632    "agent_removed",
633    "agent_inconsistent",
634    "agent_ask_done",
635    "agent_create_no_session",
636    "agent_orphan_reaped",
637    "agent_orphan_state_archived",
638    "agent_spawn_failed",
639    "agent_stop_error",
640    "agent_spawn_cwd_fallback",
641    // Claude stream-json adoption front door (daemon-emitted, ab-734fcd6c):
642    // advisory note that the single-writer claim substrate could not be consulted
643    // before spawning, so the adopt proceeded fail-open (the registry one-host
644    // re-check is the authoritative guard).
645    "agent_stream_claim_unavailable",
646    // Channel (daemon-emitted)
647    "channel_registered",
648    // Daemon lifecycle (daemon-emitted)
649    "daemon_started",
650    "daemon_exited",
651    "daemon_idle_pending_exit",
652    "daemon_shutting_down",
653    "daemon_state",
654    "daemon_recovery_error",
655    // Binary-version drift (daemon-emitted, plan ab-1891cdff): advisory note that
656    // the daemon could not fingerprint its own executable at startup, so every
657    // client drift check fails safe to Unknown.
658    "daemon_exe_fingerprint_unavailable",
659    // Drive (daemon-emitted)
660    "drive_attached",
661    "drive_detached",
662    "drive_crashed",
663    "drive_force_close_timeout",
664    "drive_keystroke_stepped",
665    "drive_takeover_after_stale",
666    "drive_watch_input_rejected",
667    // Reconcile (daemon-emitted)
668    "reconcile_deferred",
669    "reconcile_done",
670    "reconcile_error",
671    // Startup reconcile sweep (daemon-emitted, plan ab-70faa65b Architecture B)
672    "startup_reconcile_done",
673    "startup_reconcile_failed",
674    // Deliver (daemon-emitted, Task 2.2 US4)
675    "agent_deliver_injected",
676    "agent_deliver_demoted",
677    // Active-backlog drain supervisor (daemon-emitted, node x-c070): the drain
678    // tick panicked and the supervisor is restarting it with backoff. The drain
679    // decision events (active_backlog_dispatched / _yield / _parked / _skip) are
680    // loop-stream events via Journal::append, NOT daemon emits, so they are
681    // exempt from this registry by design.
682    "active_backlog_task_crashed",
683    // Meta (daemon/worker-emitted)
684    "event_payload_too_large",
685];
686
687/// Build the Branch B (Rust/fno-agents) envelope JSON Schema and the
688/// `status-v1` AgentState schema as static JSON objects.
689///
690/// This mirrors `docs/architecture/schemas/events-v3.json` Branch B and
691/// `docs/architecture/schemas/status-v1.json`. The hand-rolled approach is
692/// chosen to avoid pulling in `schemars`; it MUST be accompanied by the
693/// struct-drift unit test in `src/bin/client.rs` that asserts every
694/// `AgentState` field key is present in the emitted status schema properties.
695///
696/// Returns a JSON object suitable for printing via `--emit-schema`:
697/// ```json
698/// {
699///   "envelope": { <Branch B schema> },
700///   "status": { <status-v1 schema> },
701///   "event_kinds": ["agent_spawned", ...]
702/// }
703/// ```
704pub fn emit_schema_json() -> serde_json::Value {
705    use serde_json::json;
706    json!({
707        "envelope": {
708            "$comment": "Branch B: Rust/fno-agents supervisor envelope. Emitted by crates/fno-agents/src/events.rs.",
709            "type": "object",
710            "required": ["ts", "kind", "source"],
711            "properties": {
712                "ts": {
713                    "type": "string",
714                    "description": "UTC RFC3339 timestamp with millisecond precision and Z suffix"
715                },
716                "kind": {
717                    "type": "string",
718                    "enum": KNOWN_EVENT_KINDS,
719                    "description": "Event kind name from KNOWN_EVENT_KINDS"
720                },
721                "source": {
722                    "type": "string",
723                    "pattern": "^(daemon|worker:.+)$",
724                    "description": "Producer identity: 'daemon' or 'worker:<short_id>'"
725                }
726            },
727            "not": { "required": ["type"] },
728            "additionalProperties": true
729        },
730        "status": {
731            "$comment": "AgentState schema v1. Derived from crates/fno-agents/src/state.rs AgentState struct.",
732            "type": "object",
733            "required": ["schema_version", "short_id", "status"],
734            "properties": {
735                "schema_version": {
736                    "type": "integer",
737                    "const": 1
738                },
739                "short_id": {
740                    "type": "string"
741                },
742                "status": {
743                    "type": "string",
744                    "enum": [
745                        "spawning", "ready", "idle", "busy", "live",
746                        "restarting", "orphaned", "failed", "exited", "permanent_dead"
747                    ]
748                },
749                "ready": {
750                    "type": "boolean",
751                    "default": false
752                },
753                "last_message_at": {
754                    "type": ["string", "null"]
755                },
756                "last_reply": {
757                    "type": ["string", "null"]
758                },
759                "restart_count": {
760                    "type": "integer",
761                    "minimum": 0,
762                    "default": 0
763                },
764                "last_restart_at": {
765                    "type": ["string", "null"]
766                },
767                "pty": {
768                    "oneOf": [
769                        { "type": "null" },
770                        {
771                            "type": "object",
772                            "required": ["active", "drive_active"],
773                            "properties": {
774                                "active": { "type": "boolean" },
775                                "drive_active": { "type": "boolean", "default": false },
776                                "drive_session_id": { "type": ["string", "null"] },
777                                "drive_mode": { "type": ["string", "null"] },
778                                "last_heartbeat_at_monotonic_ns": { "type": ["integer", "null"] }
779                            },
780                            "additionalProperties": false,
781                            "if": {
782                                "properties": { "drive_active": { "const": true } },
783                                "required": ["drive_active"]
784                            },
785                            "then": {
786                                "required": ["drive_session_id", "drive_mode"],
787                                "properties": {
788                                    "drive_session_id": { "type": "string" },
789                                    "drive_mode": { "type": "string" }
790                                }
791                            }
792                        }
793                    ]
794                }
795            },
796            "additionalProperties": false
797        },
798        "event_kinds": KNOWN_EVENT_KINDS
799    })
800}