pub mod active_backlog;
pub mod agents_config;
pub mod agy_ask;
pub mod claims;
pub mod claude_adopt;
pub mod claude_ask;
pub mod claude_attach;
pub mod claude_drive;
pub mod claude_roster;
pub mod client;
pub mod client_verbs;
pub mod codex_ask;
pub mod codex_inject;
mod completion_output;
pub mod daemon;
pub mod delivery_completion;
pub mod digest;
pub mod drift;
pub mod envelope;
pub mod events;
pub mod finalize;
pub mod gc;
pub mod gemini_ask;
mod identity;
pub mod kill_criteria;
pub mod logs;
pub mod logs_client;
pub mod loop_dispatch;
pub mod loop_runtime;
pub mod loop_target;
pub mod loopcheck;
pub mod mail_inject;
pub mod manifest;
pub mod needs;
pub mod nudge;
pub mod opencode_ask;
pub mod osc;
pub mod paths;
pub mod protocol;
pub mod provider;
pub mod readiness;
pub mod scrape;
pub mod screen;
pub mod spawn_gate;
pub mod state;
pub mod stream_worker;
pub mod subprocess_ask;
pub mod subscribe;
pub mod supervisor;
pub mod terminal_stop;
pub mod verify_evidence;
pub mod version;
pub mod wait;
pub mod write_queue;
use serde::{Deserialize, Serialize};
use std::time::Duration;
#[derive(Debug, thiserror::Error, PartialEq, Eq)]
pub enum ShortIdError {
#[error("short id must be non-empty")]
Empty,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct ShortId(pub(crate) String);
impl ShortId {
pub fn new(s: impl Into<String>) -> Result<Self, ShortIdError> {
let s = s.into();
if s.is_empty() {
return Err(ShortIdError::Empty);
}
Ok(ShortId(s))
}
pub fn as_str(&self) -> &str {
&self.0
}
}
impl std::fmt::Display for ShortId {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(&self.0)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AgentStatus {
Spawning,
Ready,
Idle,
Busy,
Live,
Restarting,
Orphaned,
Failed,
Exited,
PermanentDead,
}
impl AgentStatus {
pub fn is_drive_eligible(&self) -> bool {
matches!(
self,
AgentStatus::Ready | AgentStatus::Idle | AgentStatus::Busy | AgentStatus::Live
)
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum ParsedEvent {
SessionCreated {
session_id: String,
},
OutputChunk {
text: String,
},
ReplyComplete {
text: String,
duration_ms: u64,
},
ToolUse {
name: String,
args: Option<serde_json::Value>,
},
ProviderError {
message: String,
},
Unknown {
raw: String,
},
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub struct MonotonicTimestamp(u64);
impl MonotonicTimestamp {
pub fn now() -> Self {
MonotonicTimestamp(raw_monotonic_nanos())
}
pub fn duration_since(&self, earlier: MonotonicTimestamp) -> Duration {
Duration::from_nanos(self.0.saturating_sub(earlier.0))
}
pub fn elapsed(&self) -> Duration {
MonotonicTimestamp::now().duration_since(*self)
}
pub fn as_nanos(&self) -> u64 {
self.0
}
pub fn from_nanos(nanos: u64) -> Self {
MonotonicTimestamp(nanos)
}
}
#[cfg(target_os = "linux")]
fn raw_monotonic_nanos() -> u64 {
let mut ts = libc::timespec {
tv_sec: 0,
tv_nsec: 0,
};
let rc = unsafe { libc::clock_gettime(libc::CLOCK_BOOTTIME, &mut ts) };
if rc != 0 {
tracing::error!("clock_gettime(CLOCK_BOOTTIME) failed; monotonic reading degraded to 0");
return 0;
}
(ts.tv_sec as u64)
.saturating_mul(1_000_000_000)
.saturating_add(ts.tv_nsec.max(0) as u64)
}
#[cfg(target_os = "macos")]
fn raw_monotonic_nanos() -> u64 {
#[repr(C)]
struct MachTimebaseInfo {
numer: u32,
denom: u32,
}
extern "C" {
fn mach_continuous_time() -> u64;
fn mach_timebase_info(info: *mut MachTimebaseInfo) -> libc::c_int;
}
use std::sync::OnceLock;
static TIMEBASE: OnceLock<(u64, u64)> = OnceLock::new();
let (numer, denom) = *TIMEBASE.get_or_init(|| {
let mut info = MachTimebaseInfo { numer: 0, denom: 0 };
let rc = unsafe { mach_timebase_info(&mut info) };
if rc != 0 || info.denom == 0 {
(1, 1)
} else {
(info.numer as u64, info.denom as u64)
}
});
let ticks = unsafe { mach_continuous_time() };
((ticks as u128 * numer as u128) / denom as u128) as u64
}
#[cfg(not(any(target_os = "linux", target_os = "macos")))]
fn raw_monotonic_nanos() -> u64 {
let mut ts = libc::timespec {
tv_sec: 0,
tv_nsec: 0,
};
let rc = unsafe { libc::clock_gettime(libc::CLOCK_MONOTONIC, &mut ts) };
if rc != 0 {
tracing::error!("clock_gettime(CLOCK_MONOTONIC) failed; monotonic reading degraded to 0");
return 0;
}
(ts.tv_sec as u64)
.saturating_mul(1_000_000_000)
.saturating_add(ts.tv_nsec.max(0) as u64)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn short_id_rejects_empty() {
assert_eq!(ShortId::new(""), Err(ShortIdError::Empty));
let ok = ShortId::new("wkA").unwrap();
assert_eq!(ok.as_str(), "wkA");
}
#[test]
fn agent_status_serde_roundtrip_is_snake_case() {
let json = serde_json::to_string(&AgentStatus::PermanentDead).unwrap();
assert_eq!(json, "\"permanent_dead\"");
let back: AgentStatus = serde_json::from_str(&json).unwrap();
assert_eq!(back, AgentStatus::PermanentDead);
}
#[test]
fn drive_eligibility_matches_ld28() {
assert!(AgentStatus::Ready.is_drive_eligible());
assert!(AgentStatus::Idle.is_drive_eligible());
assert!(AgentStatus::Busy.is_drive_eligible());
assert!(!AgentStatus::Restarting.is_drive_eligible());
assert!(!AgentStatus::Exited.is_drive_eligible());
assert!(!AgentStatus::PermanentDead.is_drive_eligible());
}
#[test]
fn parsed_event_tagged_serde() {
let ev = ParsedEvent::ReplyComplete {
text: "hi".into(),
duration_ms: 42,
};
let json = serde_json::to_string(&ev).unwrap();
assert!(json.contains("\"kind\":\"reply_complete\""));
let back: ParsedEvent = serde_json::from_str(&json).unwrap();
assert_eq!(ev, back);
}
#[test]
fn parsed_event_unknown_preserves_raw() {
let ev = ParsedEvent::Unknown {
raw: "{\"new_event\":1}".into(),
};
let json = serde_json::to_string(&ev).unwrap();
let back: ParsedEvent = serde_json::from_str(&json).unwrap();
assert_eq!(ev, back);
}
#[test]
fn monotonic_clock_is_nondecreasing_and_measures_elapsed() {
let t0 = MonotonicTimestamp::now();
std::thread::sleep(Duration::from_millis(20));
let t1 = MonotonicTimestamp::now();
assert!(t1 >= t0, "monotonic clock went backwards");
let elapsed = t1.duration_since(t0);
assert!(
elapsed >= Duration::from_millis(15),
"elapsed too small: {elapsed:?}"
);
assert!(
elapsed < Duration::from_secs(5),
"elapsed implausibly large: {elapsed:?}"
);
}
#[test]
fn duration_since_future_saturates_to_zero() {
let t0 = MonotonicTimestamp::now();
std::thread::sleep(Duration::from_millis(5));
let t1 = MonotonicTimestamp::now();
assert_eq!(t0.duration_since(t1), Duration::ZERO);
}
#[test]
fn every_production_emit_kind_is_registered() {
use std::collections::BTreeSet;
let known: BTreeSet<&str> = KNOWN_EVENT_KINDS.iter().copied().collect();
let src_root = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src");
let mut files = Vec::new();
collect_rs_files(&src_root, &mut files);
assert!(!files.is_empty(), "found no .rs files under {src_root:?}");
let mut unregistered: Vec<String> = Vec::new();
let mut production_kinds: BTreeSet<String> = BTreeSet::new();
let mut scanned_calls = 0usize;
for file in &files {
let text = std::fs::read_to_string(file).expect("read source file");
let prod = match text.find("#[cfg(test)]") {
Some(i) => &text[..i],
None => &text[..],
};
let file_name = file.file_name().unwrap().to_string_lossy();
for (kind, line) in scan_emit_kinds(prod) {
scanned_calls += 1;
production_kinds.insert(kind.clone());
if !known.contains(kind.as_str()) {
unregistered.push(format!(
"{file_name}:{line}: .emit(\"{kind}\") not in KNOWN_EVENT_KINDS"
));
}
}
}
assert!(
scanned_calls > 0,
"scanner found zero emit call sites - the scan pattern likely broke"
);
const TEST_ONLY_EMIT_KINDS: &[&str] = &["tick", "heartbeat", "foo", "x"];
let test_only: BTreeSet<&str> = TEST_ONLY_EMIT_KINDS.iter().copied().collect();
let mut below_only: Vec<String> = Vec::new();
for file in &files {
let text = std::fs::read_to_string(file).expect("read source file");
let boundary = match text.find("#[cfg(test)]") {
Some(i) => i,
None => continue,
};
let base_line = text[..boundary].bytes().filter(|&c| c == b'\n').count();
let file_name = file.file_name().unwrap().to_string_lossy();
for (kind, line) in scan_emit_kinds(&text[boundary..]) {
if production_kinds.contains(&kind) || test_only.contains(kind.as_str()) {
continue;
}
below_only.push(format!(
"{file_name}:{}: .emit(\"{kind}\") appears only below #[cfg(test)] \
(not emitted in production, not a known test-only fixture)",
base_line + line
));
}
}
assert!(
below_only.is_empty(),
"emit kinds found only below a #[cfg(test)] boundary - the truncation \
assumption (all production emits precede the test module) may be \
violated. If a kind below is a real production emit, register it in \
KNOWN_EVENT_KINDS and move it above the test module; if it is \
test-only, add it to TEST_ONLY_EMIT_KINDS:\n {}",
below_only.join("\n ")
);
let synthetic = "x.emit(\"agent_spawned\", &p);\n y.emit_fields(\n \"definitely_not_a_real_kind\", m);\n z.emit (\"another_fake_kind\");";
let scanned = scan_emit_kinds(synthetic);
assert!(
scanned.iter().any(|(k, l)| k == "agent_spawned" && *l == 1),
"scanner missed a single-line emit kind (or wrong line)"
);
assert!(
scanned
.iter()
.any(|(k, _)| k == "definitely_not_a_real_kind"),
"scanner missed a multi-line emit_fields kind"
);
assert!(
scanned.iter().any(|(k, _)| k == "another_fake_kind"),
"scanner missed a `.emit (` call with whitespace before the paren"
);
assert!(
!known.contains("definitely_not_a_real_kind") && !known.contains("another_fake_kind"),
"the synthetic drift kinds must not be real registered kinds"
);
assert!(
unregistered.is_empty(),
"production emit kinds missing from KNOWN_EVENT_KINDS:\n {}",
unregistered.join("\n ")
);
}
fn collect_rs_files(dir: &std::path::Path, out: &mut Vec<std::path::PathBuf>) {
let entries = match std::fs::read_dir(dir) {
Ok(e) => e,
Err(_) => return,
};
for entry in entries.flatten() {
let path = entry.path();
if path.is_dir() {
collect_rs_files(&path, out);
} else if path.extension().and_then(|e| e.to_str()) == Some("rs") {
out.push(path);
}
}
}
fn scan_emit_kinds(src: &str) -> Vec<(String, usize)> {
let bytes = src.as_bytes();
let mut kinds = Vec::new();
for needle in [".emit", ".emit_fields"] {
let nb = needle.as_bytes();
let mut from = 0usize;
while let Some(rel) = find_sub(&bytes[from..], nb) {
let pos = from + rel;
let mut j = pos + nb.len();
if needle == ".emit" && j < bytes.len() && bytes[j] == b'_' {
from = j;
continue;
}
while j < bytes.len() && (bytes[j] as char).is_whitespace() {
j += 1;
}
if j < bytes.len() && bytes[j] == b'(' {
j += 1;
while j < bytes.len() && (bytes[j] as char).is_whitespace() {
j += 1;
}
if j < bytes.len() && bytes[j] == b'"' {
let start = j + 1;
let mut k = start;
while k < bytes.len() && bytes[k] != b'"' {
k += 1;
}
if k < bytes.len() {
let kind = String::from_utf8_lossy(&bytes[start..k]).into_owned();
let line = src[..pos].bytes().filter(|&c| c == b'\n').count() + 1;
kinds.push((kind, line));
}
}
}
from = pos + nb.len();
}
}
kinds
}
fn find_sub(haystack: &[u8], needle: &[u8]) -> Option<usize> {
if needle.is_empty() || haystack.len() < needle.len() {
return None;
}
haystack.windows(needle.len()).position(|w| w == needle)
}
}
pub const KNOWN_EVENT_KINDS: &[&str] = &[
"agent_spawned",
"agent_stopped",
"agent_exited",
"agent_removed",
"agent_inconsistent",
"agent_ask_done",
"agent_create_no_session",
"agent_orphan_reaped",
"agent_orphan_state_archived",
"agent_row_reaped",
"node_failed",
"bg_worker_terminal_stopped",
"agent_spawn_failed",
"agent_stop_error",
"agent_spawn_cwd_fallback",
"agent_stream_claim_unavailable",
"channel_registered",
"daemon_started",
"daemon_exited",
"daemon_idle_pending_exit",
"daemon_shutting_down",
"daemon_state",
"daemon_recovery_error",
"daemon_exe_fingerprint_unavailable",
"drive_attached",
"drive_detached",
"drive_crashed",
"drive_force_close_timeout",
"drive_keystroke_stepped",
"drive_refused_busy_elsewhere",
"drive_takeover_after_stale",
"drive_watch_input_rejected",
"reconcile_deferred",
"reconcile_done",
"reconcile_error",
"startup_reconcile_done",
"startup_reconcile_failed",
"agent_deliver_injected",
"agent_deliver_demoted",
"agent_deliver_status_write_failed",
"active_backlog_task_crashed",
"active_backlog_mission_retired",
"dispatch_deferred",
"event_payload_too_large",
"inside_leg_report",
"inside_leg_report_dropped",
"inside_leg_completed",
"inside_leg_report_buffered",
"inside_leg_buffer_flushed",
"screen_state_change",
];
pub fn emit_schema_json() -> serde_json::Value {
use serde_json::json;
json!({
"envelope": {
"$comment": "Unified events.jsonl envelope (x-2901). Emitted by crates/fno-agents/src/events.rs; structurally equal to schemas/events-v3.json after doc-key stripping (the parity gate diffs them).",
"type": "object",
"required": ["ts", "type", "source", "data"],
"properties": {
"ts": {
"type": "string",
"description": "UTC RFC3339 timestamp with millisecond precision and Z suffix"
},
"type": {
"type": "string",
"description": "Event type name; the daemon kinds live in KNOWN_EVENT_KINDS (see event_kinds below)"
},
"source": {
"type": "string",
"anyOf": [
{ "enum": ["active-backlog", "approvals", "backlog", "daemon", "fno-loop", "hook", "megatron", "megawalk", "migration", "observer", "skill_diff", "subagent", "target", "test"] },
{ "pattern": "^(worker|stream-worker):.+$" }
],
"description": "Producer identity: a fixed-string source or a per-agent worker (worker:<id> / stream-worker:<id>)"
},
"data": {
"type": "object",
"description": "Per-type payload object"
}
},
"additionalProperties": true
},
"status": {
"$comment": "AgentState schema v1. Derived from crates/fno-agents/src/state.rs AgentState struct.",
"type": "object",
"required": ["schema_version", "short_id", "status"],
"properties": {
"schema_version": {
"type": "integer",
"const": 1
},
"short_id": {
"type": "string"
},
"status": {
"type": "string",
"enum": [
"spawning", "ready", "idle", "busy", "live",
"restarting", "orphaned", "failed", "exited", "permanent_dead"
]
},
"ready": {
"type": "boolean",
"default": false
},
"last_message_at": {
"type": ["string", "null"]
},
"last_reply": {
"type": ["string", "null"]
},
"restart_count": {
"type": "integer",
"minimum": 0,
"default": 0
},
"last_restart_at": {
"type": ["string", "null"]
},
"pty": {
"oneOf": [
{ "type": "null" },
{
"type": "object",
"required": ["active", "drive_active"],
"properties": {
"active": { "type": "boolean" },
"drive_active": { "type": "boolean", "default": false },
"drive_session_id": { "type": ["string", "null"] },
"drive_mode": { "type": ["string", "null"] },
"last_heartbeat_at_monotonic_ns": { "type": ["integer", "null"] }
},
"additionalProperties": false,
"if": {
"properties": { "drive_active": { "const": true } },
"required": ["drive_active"]
},
"then": {
"required": ["drive_session_id", "drive_mode"],
"properties": {
"drive_session_id": { "type": "string" },
"drive_mode": { "type": "string" }
}
}
}
]
}
},
"additionalProperties": false
},
"event_kinds": KNOWN_EVENT_KINDS
})
}