diff --git a/Cargo.lock b/Cargo.lock index 8379e4c9..b607adb7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -186,6 +186,9 @@ dependencies = [ "thiserror 1.0.69", "tokio", "toml", + "tracing", + "tracing-subscriber", + "uuid", "walkdir", "watchman_client", "zstd", @@ -1912,6 +1915,7 @@ dependencies = [ "tokio-stream", "tonic", "tonic-build", + "tracing", "ulid", "uuid", ] @@ -3125,6 +3129,15 @@ dependencies = [ "digest", ] +[[package]] +name = "sharded-slab" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6" +dependencies = [ + "lazy_static", +] + [[package]] name = "shell-words" version = "1.1.1" @@ -3402,6 +3415,15 @@ dependencies = [ "syn 2.0.118", ] +[[package]] +name = "thread_local" +version = "1.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ad99c4c6d32803332c548b1af0540b357b3f5fc0be8f6c6bfe8b2e6ae784070" +dependencies = [ + "cfg-if", +] + [[package]] name = "tinystr" version = "0.8.3" @@ -3702,6 +3724,17 @@ dependencies = [ "once_cell", ] +[[package]] +name = "tracing-subscriber" +version = "0.3.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319" +dependencies = [ + "sharded-slab", + "thread_local", + "tracing-core", +] + [[package]] name = "tree-sitter" version = "0.25.10" diff --git a/Cargo.toml b/Cargo.toml index cfa22eba..95af0eb4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -121,6 +121,8 @@ ignore = "0.4" tempfile = "3.10" parking_lot = "0.12" log = "0.4" +tracing = "0.1" +tracing-subscriber = { version = "0.3", default-features = false, features = ["registry", "std"] } env_logger = "0.11" xxhash-rust = { version = "0.8", features = ["xxh3"] } diff --git a/README.md b/README.md index 1e897d7c..952ca5ba 100644 --- a/README.md +++ b/README.md @@ -403,10 +403,15 @@ Every change stores two parallel representations: |---------|-------------| | `atomic agent enable` | Install agent hooks (auto-detect or `--agent claude-code`) | | `atomic agent disable` | Remove agent hooks | -| `atomic agent status` | Show active sessions and hook status (`--json` for tooling) | +| `atomic agent status` | Show sessions, hook liveness, and failures (`--json` for tooling) | | `atomic agent explain ` | Generate AI reasoning summary for a session | | `atomic agent attest` | List and inspect attestations | +Hook health is computed from session-scoped logs, so concurrent agents retain +their own outcomes and a child's success cannot hide a parent's failure. See +[Hook Health and Session Logs](docs/hook-health.md) for diagnosis, JSON fields, +and parent-session correlation. + ### Vault Commands | Command | Description | diff --git a/atomic-agent/Cargo.toml b/atomic-agent/Cargo.toml index 4101eba7..44b3d4f5 100644 --- a/atomic-agent/Cargo.toml +++ b/atomic-agent/Cargo.toml @@ -44,6 +44,9 @@ anyhow = { workspace = true } # Logging log = { workspace = true } +tracing = { workspace = true } +tracing-subscriber = { workspace = true } +uuid = { workspace = true } # Filesystem traversal (fallback watcher when Watchman unavailable) walkdir = { workspace = true } diff --git a/atomic-agent/README.md b/atomic-agent/README.md index 55f1e6cc..02f4665e 100644 --- a/atomic-agent/README.md +++ b/atomic-agent/README.md @@ -65,7 +65,33 @@ The recording workflow on each turn end: 2. **Add** — track any new files the agent created (filtering out `node_modules`, `target`, etc.) 3. **Record** — create an Atomic change with AI provenance metadata -No daemon. No Watchman required. No git. Each hook invocation is a standalone process that opens the repo, does its work, and exits. Codex is the one lifecycle-specific exception: because Codex caps `SessionEnd` hooks at three seconds, the hook hands finalization to a short-lived background worker after safely forwarding the event payload. +No Watchman or Git is required. Each hook invocation is a standalone process. Database operations use the shared service handlers: local mode runs them in-process; reactor mode calls the daemon. Codex is the one lifecycle-specific exception: because Codex caps `SessionEnd` hooks at three seconds, the hook hands finalization to a short-lived background worker after safely forwarding the event payload. + +## Hook Health and Session Logs + +`atomic agent status` reports the last successful hook dispatch and outstanding +failures. `--json` includes aggregate `hook_health`, session-scoped `hook_sessions` +(with parent IDs and per-verb outcomes), and `hook_health_error` when the log cannot +be loaded. A child's successful hook cannot clear its parent's failure. + +Native hooks, typed `agent receive`, and the shared dispatch handler emit +structured Rust `tracing` events. The service appends immutable observations to +the session log in `.atomic/atomic.redb`, independently of provenance turn state. +Event IDs make append retries idempotent; transactional writes retain concurrent +parent/child activity. Health is projected from the log rather than a shared +`hook-health.json` snapshot. Existing plugin and git-shadow errors are also read +from the bounded tail of `hook-errors.log`. + +Session IDs come from hook payloads. Bridges can explicitly carry correlation +across processes with `ATOMIC_SESSION_ID` and `ATOMIC_PARENT_SESSION_ID`; typed +dispatch forwards the resolved parent ID on the service request. Malformed input +without an identifiable session remains an unassigned diagnostic. During a +service outage, each observation is spooled to its own immutable file under +`.atomic/hook-log-pending/`; a successful append replays pending observations +idempotently. Status can report pending observations while the service is down. + +See [Hook Health and Session Logs](../docs/hook-health.md) for the status JSON +fields and troubleshooting workflow. ## Agent Identity diff --git a/atomic-agent/src/hook_health.rs b/atomic-agent/src/hook_health.rs new file mode 100644 index 00000000..0145bbef --- /dev/null +++ b/atomic-agent/src/hook_health.rs @@ -0,0 +1,973 @@ +//! Hook health projected from immutable, session-scoped operational logs. +//! Legacy plugin/git-shadow errors remain readable from `hook-errors.log`. + +use std::collections::BTreeMap; +use std::io::{Seek, SeekFrom}; +use std::path::Path; + +use serde::{Deserialize, Serialize}; + +/// Version of the computed health report. +const HEALTH_SCHEMA_VERSION: u32 = 1; + +/// How many bytes of `hook-errors.log` tail to read. +/// +/// The log is unbounded, so the reader seeks to the end and parses only the +/// last chunk. Large enough to cover a busy day of errors on a repo like +/// `atomic`; small enough that `agent status` stays instant. +const ERROR_LOG_TAIL_BYTES: u64 = 256 * 1024; + +/// Longest error message retained per entry, in the aggregated report. +const MAX_DETAIL_LEN: usize = 200; + +/// Prefix the agent plugins use for session ids in `hook-errors.log`. +/// +/// It is the one token the two writers' grammars can be told apart by: the +/// plugin writer emits it as the second field, the git-shadow writer emits +/// its own tag there instead. +const SESSION_ID_PREFIX: &str = "ses_"; + +// --------------------------------------------------------------------------- +// Health record +// --------------------------------------------------------------------------- + +/// How a single hook dispatch ended. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum FireOutcome { + /// The event dispatched successfully; it may not have needed a new change. + Ok, + /// The event reached the dispatcher but failed. + Error, +} + +impl FireOutcome { + /// The wire name, used in human output. + pub fn as_str(self) -> &'static str { + match self { + FireOutcome::Ok => "ok", + FireOutcome::Error => "error", + } + } +} + +/// The most recent dispatch of one verb for one agent. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct VerbHealth { + /// RFC3339 timestamp of the most recent dispatch, successful or not. + pub last_fired: String, + /// Outcome of that most recent dispatch. + pub outcome: FireOutcome, + /// Failure detail when `outcome` is `error`; truncated. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub detail: Option, + /// Timestamp of the most recent *successful* dispatch, if any. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub last_ok: Option, +} + +/// One agent's hook health. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct AgentHealth { + pub display_name: String, + /// Keyed by hook verb (`session-start`, `before-tool`, …). + pub verbs: BTreeMap, +} + +/// Computed report; persistence belongs to the operational event log. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct HookHealth { + pub schema_version: u32, + /// RFC3339 timestamp of the newest observation. + pub updated_at: String, + pub agents: BTreeMap, + pub sessions: BTreeMap, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SessionHookHealth { + pub agent: String, + pub session_id: Option, + pub parent_session_id: Option, + pub health: AgentHealth, +} + +impl HookHealth { + /// An empty document, ready for its first [`HookHealth::record`]. + pub fn empty(now: &str) -> Self { + Self { + schema_version: HEALTH_SCHEMA_VERSION, + updated_at: now.to_string(), + agents: BTreeMap::new(), + sessions: BTreeMap::new(), + } + } + + /// Fold one dispatch into the record, preserving `last_ok` across a + /// later failure so a hook that has *never* succeeded is + /// distinguishable from one that worked and then broke. + pub fn record( + &mut self, + agent: &str, + display_name: &str, + verb: &str, + outcome: FireOutcome, + detail: Option<&str>, + now: &str, + ) { + let entry = self + .agents + .entry(agent.to_string()) + .or_insert_with(|| AgentHealth { + display_name: display_name.to_string(), + verbs: BTreeMap::new(), + }); + entry.display_name = display_name.to_string(); + + fold_verb(&mut entry.verbs, verb, outcome, detail, now); + if observed_at(now) >= observed_at(&self.updated_at) { + self.updated_at = now.to_string(); + } + } + + /// Build a report from observations, including out-of-order outage replay. + pub fn from_events(events: Vec) -> Option { + let mut report = Self::empty(&events.first()?.at); + for event in events { + report.record_event(&event); + } + Some(report) + } + + /// Incremental projection, safe for pages and older events replayed after + /// an outage. Only a success in the same session can heal that session. + pub fn record_event(&mut self, event: &crate::hook_log::SessionLogEvent) { + let outcome = if event.outcome == "ok" { + FireOutcome::Ok + } else { + FireOutcome::Error + }; + self.record( + &event.agent, + &event.display_name, + &event.verb, + outcome, + event.detail.as_deref(), + &event.at, + ); + let key = serde_json::to_string(&(&event.agent, &event.session_id)) + .expect("string tuple serializes"); + let session = self + .sessions + .entry(key) + .or_insert_with(|| SessionHookHealth { + agent: event.agent.clone(), + session_id: event.session_id.clone(), + parent_session_id: event.parent_session_id.clone(), + health: AgentHealth { + display_name: event.display_name.clone(), + verbs: BTreeMap::new(), + }, + }); + if event.parent_session_id.is_some() { + session.parent_session_id = event.parent_session_id.clone(); + } + fold_verb( + &mut session.health.verbs, + &event.verb, + outcome, + event.detail.as_deref(), + &event.at, + ); + } +} + +fn observed_at(at: &str) -> Option> { + chrono::DateTime::parse_from_rfc3339(at).ok() +} + +fn fold_verb( + verbs: &mut BTreeMap, + verb: &str, + outcome: FireOutcome, + detail: Option<&str>, + at: &str, +) { + let previous = verbs.get(verb); + let last_ok = previous.and_then(|state| state.last_ok.clone()); + let last_ok = if outcome == FireOutcome::Ok + && last_ok + .as_deref() + .is_none_or(|ok| observed_at(at) >= observed_at(ok)) + { + Some(at.to_string()) + } else { + last_ok + }; + if let Some(previous) = verbs + .get_mut(verb) + .filter(|state| observed_at(&state.last_fired) > observed_at(at)) + { + previous.last_ok = last_ok; + return; + } + verbs.insert( + verb.to_string(), + VerbHealth { + last_fired: at.to_string(), + outcome, + detail: detail.map(truncate_detail), + last_ok, + }, + ); +} + +/// Flatten a health record into one row per agent, for display and JSON. +/// +/// Agents with no recorded fires are omitted entirely: absence is already +/// reported by the `hooks_installed` flag, and a row of empty timestamps +/// would read as a second, contradictory answer. +pub fn summarize(health: &HookHealth) -> Vec { + let mut out: Vec = health + .agents + .iter() + .map(|(name, agent)| { + let last_fired = agent + .verbs + .values() + .map(|v| v.last_fired.as_str()) + .max_by_key(|at| observed_at(at)) + .map(str::to_string); + let last_ok = agent + .verbs + .values() + .filter_map(|v| v.last_ok.as_deref()) + .max_by_key(|at| observed_at(at)) + .map(str::to_string); + let last_error = agent + .verbs + .values() + .filter(|v| v.outcome == FireOutcome::Error) + .max_by_key(|v| observed_at(&v.last_fired)) + .map(|v| v.last_fired.clone()); + let failing: Vec = agent + .verbs + .iter() + .filter(|(_, v)| v.outcome == FireOutcome::Error) + .map(|(k, _)| k.clone()) + .collect(); + + AgentHookSummary { + name: name.clone(), + display_name: agent.display_name.clone(), + last_fired, + last_ok, + last_error, + verbs_recorded: agent.verbs.len(), + failing_verbs: failing, + } + }) + .collect(); + + out.sort_by(|a, b| a.name.cmp(&b.name)); + // A successful child must not clear a parent's outstanding failure. + if !health.sessions.is_empty() { + for row in &mut out { + let mut failing = std::collections::BTreeSet::new(); + let mut last_error = None; + for session in health + .sessions + .values() + .filter(|session| session.agent == row.name) + { + for (verb, state) in &session.health.verbs { + if state.outcome == FireOutcome::Error { + failing.insert(verb.clone()); + if last_error + .as_deref() + .is_none_or(|at| observed_at(&state.last_fired) >= observed_at(at)) + { + last_error = Some(state.last_fired.clone()); + } + } + } + } + row.failing_verbs = failing.into_iter().collect(); + row.last_error = last_error; + } + } + out +} + +/// One agent's health, flattened for display. +#[derive(Debug, Clone, Serialize)] +pub struct AgentHookSummary { + pub name: String, + pub display_name: String, + pub last_fired: Option, + pub last_ok: Option, + pub last_error: Option, + pub verbs_recorded: usize, + pub failing_verbs: Vec, +} + +impl AgentHookSummary { + /// Describe the age of the last success relative to the current time. + pub fn liveness(&self) -> String { + self.liveness_at(&chrono::Utc::now().to_rfc3339()) + } + + fn liveness_at(&self, now: &str) -> String { + match (&self.last_fired, &self.last_ok) { + (None, _) => "never fired".to_string(), + (Some(_), None) => "fired, never succeeded".to_string(), + (Some(_), Some(ok)) => { + let age = relative_age(now, ok); + format!("last ok {age}") + } + } + } +} + +// --------------------------------------------------------------------------- +// hook-errors.log +// --------------------------------------------------------------------------- + +/// One parsed line of `hook-errors.log`. +#[derive(Debug, Clone, Serialize)] +pub struct HookErrorEntry { + /// RFC3339 timestamp, verbatim — the file's two writers do not agree on + /// precision, so it is not normalized. + pub at: String, + /// The verb, or `shadow-validate:` for git-shadow entries. + pub verb: String, + /// Everything after the verb, flattened to one line and truncated. + pub message: String, +} + +/// A bounded, aggregated read of `hook-errors.log`. +#[derive(Debug, Clone, Serialize)] +pub struct HookErrors { + /// Number of entries parsed in the window that was read. + pub total: usize, + /// Physical lines consumed, including continuation lines folded into an + /// entry's message. Greater than `total` whenever messages spanned lines. + pub lines: usize, + /// Timestamp of the oldest entry in the window, if any. + pub first_seen: Option, + /// Timestamp of the newest entry in the window, if any. + pub last_seen: Option, + /// Human-readable description of what was read, e.g. `all 636 entries + /// from 1784 lines` — the distinction matters, because a bounded read + /// cannot claim the file is only as bad as the window. + pub window: String, + /// Error classes with counts, most frequent first. + pub classes: Vec, + /// The most recent entries, newest first. + pub recent: Vec, +} + +/// A group of similar errors, counted. +#[derive(Debug, Clone, Serialize)] +pub struct ErrorClass { + /// Stable-ish signature used to group entries. + pub signature: String, + /// A representative message. + pub example: String, + pub count: usize, +} + +/// Read a bounded tail of `hook-errors.log` and aggregate it. +/// +/// Returns `None` when the file is absent. A file that exists but cannot be +/// read or parsed yields an empty report rather than an error, because the +/// point of the report is to explain the situation. +pub fn read_errors(repo_root: &Path) -> Option { + let path = repo_root.join(".atomic").join("hook-errors.log"); + let meta = std::fs::metadata(&path).ok()?; + let file_len = meta.len(); + let mut file = std::fs::File::open(&path).ok()?; + + // Read only the tail; the file is unbounded and never rotated. + let start = file_len.saturating_sub(ERROR_LOG_TAIL_BYTES); + if start > 0 { + file.seek(SeekFrom::Start(start)).ok()?; + } + let mut buf = Vec::new(); + std::io::Read::read_to_end(&mut file, &mut buf).ok()?; + + let text = String::from_utf8_lossy(&buf); + // A tail seek can land mid-line, in which case the first line read is a + // fragment and must be dropped. When the whole file was read (start == + // 0) nothing is partial, so the first line is real and must be kept — + // dropping it would silently lose the oldest entry on every small log. + let body = if start > 0 { + match text.find('\n') { + Some(i) => &text[i + 1..], + None => "", + } + } else { + &text + }; + + let (entries, lines) = parse_error_log(body); + + Some(aggregate(&entries, lines, start)) +} + +/// Parse a log into entries, folding continuation lines into their entry. +/// +/// The plugin writer appends `message` unescaped, so a stack trace or a +/// multi-paragraph error spans several physical lines. A line that does not +/// start with a timestamp is therefore a continuation of the entry above it, +/// not a separate entry — and dropping it would lose most of the message. +/// +/// Returns the entries plus the number of physical lines consumed, so the +/// report can be honest about what fraction of the file it understood. +fn parse_error_log(body: &str) -> (Vec, usize) { + let mut entries: Vec = Vec::new(); + let mut lines = 0usize; + + for line in body.lines() { + if line.trim().is_empty() { + continue; + } + lines += 1; + + match parse_error_line(line) { + Some(entry) => entries.push(entry), + None => { + if let Some(last) = entries.last_mut() { + last.message = truncate_detail(&format!("{} {}", last.message, line.trim())); + } + } + } + } + + (entries, lines) +} + +/// Parse one log line into an entry, or `None` if it is not shaped like one. +/// +/// Two writers append to this file with different grammars: +/// +/// ```text +/// {rfc3339} {session_id} {verb} {message} # agent plugins +/// {rfc3339} shadow-validate:{rule} view={view} {message} # git-shadow validator +/// ``` +/// +/// The plugin grammar leads with the session id, so the verb is the third +/// field; the shadow grammar puts its own tag there. They are told apart by +/// the session-id prefix, which is the one thing both grammars can be +/// relied on to disagree about. +/// +/// A line whose first token is not a timestamp is not an entry of its own; +/// [`parse_error_log`] folds it into the previous one. +fn parse_error_line(line: &str) -> Option { + let mut fields = line.split_whitespace(); + let at = fields.next()?; + if chrono::DateTime::parse_from_rfc3339(at).is_err() { + return None; + } + + let second = fields.next().unwrap_or(""); + + // Everything after the verb, rebuilt so a multi-word message survives. + let (verb, rest): (&str, String) = if second.starts_with(SESSION_ID_PREFIX) { + let verb = fields.next().unwrap_or("unknown"); + (verb, fields.collect::>().join(" ")) + } else { + (second, fields.collect::>().join(" ")) + }; + + Some(HookErrorEntry { + at: at.to_string(), + verb: if verb.is_empty() { + "unknown".to_string() + } else { + verb.to_string() + }, + message: truncate_detail(&rest), + }) +} + +/// Group entries into error classes and build the report. +fn aggregate(entries: &[HookErrorEntry], lines: usize, start: u64) -> HookErrors { + // Group on a coarse signature: the error text with digits, hex and + // paths collapsed, so that "turn 78 is not running" and "turn 92 is not + // running" are one class rather than hundreds of singletons. + let mut order: Vec = Vec::new(); + let mut buckets: BTreeMap = BTreeMap::new(); + + for entry in entries { + let signature = error_signature(&entry.message); + match buckets.get_mut(&signature) { + Some((count, _)) => *count += 1, + None => { + order.push(signature.clone()); + buckets.insert(signature, (1, entry.message.clone())); + } + } + } + + let mut classes: Vec = order + .into_iter() + .filter_map(|signature| { + buckets.get(&signature).map(|(count, example)| ErrorClass { + signature, + example: example.clone(), + count: *count, + }) + }) + .collect(); + classes.sort_by(|a, b| b.count.cmp(&a.count).then(a.signature.cmp(&b.signature))); + + let mut recent: Vec = entries.to_vec(); + recent.reverse(); + + let window = if start > 0 { + format!( + "last {lines} lines, {k} entries (file is {kb} KiB; older entries not read)", + k = entries.len(), + kb = (start + 64 * 1024) / 1024, + ) + } else { + format!("all {lines} lines, {k} entries", k = entries.len()) + }; + + HookErrors { + total: entries.len(), + lines, + first_seen: entries.first().map(|e| e.at.clone()), + last_seen: entries.last().map(|e| e.at.clone()), + window, + classes: classes.into_iter().take(10).collect(), + recent: recent.into_iter().take(10).collect(), + } +} + +/// Collapse a message to a grouping key. +/// +/// Session ids, turn numbers, hashes and paths all vary between occurrences +/// of the same underlying failure. Normalizing them is what turns 546 +/// separate lines into one actionable row. +fn error_signature(message: &str) -> String { + let chars: Vec = message.chars().collect(); + let mut out = String::with_capacity(chars.len()); + let mut i = 0; + + while i < chars.len() { + let c = chars[i]; + if c.is_ascii_digit() { + while i < chars.len() && chars[i].is_ascii_digit() { + i += 1; + } + out.push('N'); + } else if c.is_ascii_alphanumeric() || c == '_' { + let start = i; + while i < chars.len() && (chars[i].is_ascii_alphanumeric() || chars[i] == '_') { + i += 1; + } + let word: String = chars[start..i].iter().collect(); + out.push_str(&normalize_token(&word)); + } else { + out.push(c); + i += 1; + } + } + + out.split_whitespace().collect::>().join(" ") +} + +/// Map a bare word to a placeholder when it looks like an identifier. +/// +/// Only length and shape are used — a token is not identified by its +/// spelling, so nothing sensitive is retained in the grouping key. +fn normalize_token(word: &str) -> String { + let has_digit = word.chars().any(|c| c.is_ascii_digit()); + let long = word.len() >= 16; + if has_digit && long { + "ID".to_string() + } else if has_digit && word.chars().all(|c| c.is_ascii_alphanumeric()) { + "N".to_string() + } else { + word.to_ascii_lowercase() + } +} + +fn truncate_detail(s: &str) -> String { + let flat = s.split_whitespace().collect::>().join(" "); + if flat.chars().count() <= MAX_DETAIL_LEN { + return flat; + } + let head: String = flat.chars().take(MAX_DETAIL_LEN).collect(); + format!("{head}...") +} + +/// Describe how long ago `then` was, relative to `now`, in coarse units. +fn relative_age(now: &str, then: &str) -> String { + let (Ok(now), Ok(then)) = ( + chrono::DateTime::parse_from_rfc3339(now), + chrono::DateTime::parse_from_rfc3339(then), + ) else { + return "recently".to_string(); + }; + let secs = (now - then).num_seconds().max(0); + match secs { + 0..=59 => format!("{secs}s ago"), + 60..=3599 => format!("{}m ago", secs / 60), + 3600..=86_399 => format!("{}h ago", secs / 3600), + _ => format!("{}d ago", secs / 86_400), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn tmp_repo() -> tempfile::TempDir { + tempfile::tempdir().unwrap() + } + + #[test] + fn record_then_summarize_reports_liveness() { + let dir = tmp_repo(); + let root = dir.path(); + std::fs::create_dir_all(root.join(".atomic")).unwrap(); + + let now = "2026-09-25T18:00:00+00:00"; + let mut health = HookHealth::empty(now); + health.record( + "opencode", + "OpenCode", + "session-start", + FireOutcome::Ok, + None, + now, + ); + + let rows = summarize(&health); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].name, "opencode"); + assert_eq!(rows[0].verbs_recorded, 1); + assert!(rows[0].failing_verbs.is_empty()); + assert_eq!(rows[0].liveness_at(now), "last ok 0s ago"); + } + + #[test] + fn a_fresh_failure_keeps_the_earlier_success_visible() { + let mut health = HookHealth::empty("2026-09-25T17:00:00+00:00"); + health.record( + "opencode", + "OpenCode", + "before-tool", + FireOutcome::Ok, + None, + "2026-09-25T17:00:00+00:00", + ); + health.record( + "opencode", + "OpenCode", + "before-tool", + FireOutcome::Error, + Some("boom"), + "2026-09-25T18:00:00+00:00", + ); + + let rows = summarize(&health); + assert_eq!(rows[0].failing_verbs, vec!["before-tool"]); + // last_fired moved forward, but last_ok stayed behind it — that + // difference is the whole reason both fields exist. + assert_eq!( + rows[0].last_error.as_deref(), + Some("2026-09-25T18:00:00+00:00") + ); + assert_eq!( + rows[0].last_ok.as_deref(), + Some("2026-09-25T17:00:00+00:00") + ); + assert_eq!( + rows[0].liveness_at("2026-09-25T18:00:00+00:00"), + "last ok 1h ago" + ); + } + + #[test] + fn a_stale_success_ages_even_when_no_new_hook_fires() { + let mut health = HookHealth::empty("2026-09-25T17:00:00+00:00"); + health.record( + "opencode", + "OpenCode", + "before-tool", + FireOutcome::Ok, + None, + "2026-09-25T17:00:00+00:00", + ); + let rows = summarize(&health); + assert_eq!(rows[0].last_fired, rows[0].last_ok); + assert_eq!( + rows[0].liveness_at("2026-09-27T17:00:00+00:00"), + "last ok 2d ago" + ); + } + + #[test] + fn a_hook_that_never_succeeded_is_distinguishable() { + let mut health = HookHealth::empty("2026-09-25T17:00:00+00:00"); + health.record( + "opencode", + "OpenCode", + "stop", + FireOutcome::Error, + Some("no such command"), + "2026-09-25T17:00:00+00:00", + ); + let rows = summarize(&health); + assert_eq!(rows[0].liveness(), "fired, never succeeded"); + } + + #[test] + fn a_child_success_does_not_heal_a_parent_failure() { + let event = |session: &str, parent: Option<&str>, outcome: &str, at: &str| { + crate::hook_log::SessionLogEvent { + event_id: session.into(), + session_id: Some(session.into()), + parent_session_id: parent.map(str::to_owned), + agent: "opencode".into(), + display_name: "OpenCode".into(), + verb: "before-tool".into(), + outcome: outcome.into(), + detail: None, + at: at.into(), + } + }; + let health = HookHealth::from_events(vec![ + event("child", Some("parent"), "ok", "2026-09-25T18:00:00Z"), + event("parent", None, "error", "2026-09-25T17:00:00Z"), + ]) + .unwrap(); + assert_eq!(summarize(&health)[0].failing_verbs, ["before-tool"]); + let child = health + .sessions + .values() + .find(|row| row.session_id.as_deref() == Some("child")) + .unwrap(); + assert_eq!(child.parent_session_id.as_deref(), Some("parent")); + assert_eq!(child.health.verbs["before-tool"].outcome, FireOutcome::Ok); + assert_eq!( + summarize(&health)[0].last_error.as_deref(), + Some("2026-09-25T17:00:00Z") + ); + } + + #[test] + fn replayed_older_observations_preserve_latest_outcome_and_last_success() { + let mut health = HookHealth::empty("2026-10-12T10:00:00Z"); + // Fractional precision and timezone offsets do not sort as strings. + health.record( + "opencode", + "OpenCode", + "before-tool", + FireOutcome::Error, + Some("new failure"), + "2026-10-12T10:00:00.500Z", + ); + health.record( + "opencode", + "OpenCode", + "before-tool", + FireOutcome::Ok, + None, + "2026-10-12T11:00:00+01:00", + ); + let verb = &health.agents["opencode"].verbs["before-tool"]; + assert_eq!(verb.outcome, FireOutcome::Error); + assert_eq!(verb.last_ok.as_deref(), Some("2026-10-12T11:00:00+01:00")); + assert_eq!(verb.detail.as_deref(), Some("new failure")); + health.record( + "opencode", + "OpenCode", + "after-tool", + FireOutcome::Ok, + None, + "2026-10-12T10:00:00Z", + ); + assert_eq!( + summarize(&health)[0].last_fired.as_deref(), + Some("2026-10-12T10:00:00.500Z") + ); + } + + #[test] + fn parses_the_plugin_grammar() { + let entry = parse_error_line( + "2026-09-21T23:35:21.635Z ses_f4ee01e5dffeKYy29CIT5h2Lst session.idle Error: boom", + ) + .expect("parses"); + assert_eq!(entry.at, "2026-09-21T23:35:21.635Z"); + assert_eq!(entry.verb, "session.idle"); + assert_eq!(entry.message, "Error: boom"); + } + + #[test] + fn parses_the_shadow_grammar() { + let entry = parse_error_line( + "2026-09-20T10:00:00Z shadow-validate:drift view=dev 2 files disagree", + ) + .expect("parses"); + assert_eq!(entry.verb, "shadow-validate:drift"); + assert_eq!(entry.message, "view=dev 2 files disagree"); + } + + #[test] + fn rejects_lines_that_are_not_entries() { + assert!(parse_error_line("just some prose").is_none()); + assert!(parse_error_line("").is_none()); + } + + #[test] + fn continuation_lines_fold_into_their_entry() { + // The real file: the plugin appends `message` unescaped, so a stack + // trace spans lines and the report must not read each as its own + // entry (or drop the useful tail). + let body = "\ +2026-09-15T23:07:37Z ses_abc before-tool exit 128: boom + at dispatch (hooks.rs:1) + at record (provenance.rs:2) +2026-09-15T23:07:38Z ses_abc before-tool exit 128: boom + at dispatch (hooks.rs:1) + at record (provenance.rs:2) +"; + let (entries, lines) = parse_error_log(body); + assert_eq!(lines, 6); + assert_eq!(entries.len(), 2, "continuations are not separate entries"); + assert!(entries[0].message.contains("at record (provenance.rs:2)")); + // Two identical failures differing only in the timestamp still group. + let report = aggregate(&entries, lines, 0); + assert_eq!(report.classes.len(), 1); + assert_eq!(report.classes[0].count, 2); + } + + #[test] + fn a_continuation_before_any_entry_is_dropped() { + let (entries, lines) = parse_error_log("orphan continuation\nno timestamp here\n"); + assert!(entries.is_empty()); + assert_eq!(lines, 2); + } + + #[test] + fn groups_turn_numbers_into_one_class() { + // The real shape from the atomic repo: 546 lines that differ only in + // the turn number and the session id. + let entries: Vec = (78..90) + .map(|t| { + parse_error_line(&format!( + "2026-09-15T23:07:37Z ses_abc before-tool exit 128: Provenance turn {t} is not running" + )) + .unwrap() + }) + .collect(); + + let report = aggregate(&entries, entries.len(), 0); + assert_eq!(report.total, 12); + assert_eq!( + report.classes.len(), + 1, + "turn numbers must not fragment the class" + ); + assert_eq!(report.classes[0].count, 12); + } + + #[test] + fn distinct_failures_stay_distinct() { + let entries = vec![ + parse_error_line("2026-09-15T23:00:00Z ses_abc before-tool turn 78 is not running") + .unwrap(), + parse_error_line( + "2026-09-15T23:00:01Z ses_abc session.idle view changed during a tool", + ) + .unwrap(), + ]; + let report = aggregate(&entries, entries.len(), 0); + assert_eq!(report.classes.len(), 2); + } + + #[test] + fn recent_is_newest_first() { + let entries = vec![ + parse_error_line("2026-09-15T23:00:00Z ses_abc before-tool the older error").unwrap(), + parse_error_line("2026-09-15T23:00:09Z ses_abc before-tool the newer error").unwrap(), + ]; + let report = aggregate(&entries, entries.len(), 0); + assert_eq!(report.recent[0].message, "the newer error"); + assert_eq!(report.recent[1].message, "the older error"); + } + + #[test] + fn a_bounded_read_says_so_rather_than_claiming_the_whole_file() { + let entries = + vec![parse_error_line("2026-09-15T23:00:00Z ses_abc before-tool boom").unwrap()]; + let report = aggregate(&entries, entries.len(), 644_000); + assert!(report.window.contains("older entries not read")); + } + + #[test] + fn error_signature_keeps_words_and_collapses_numbers() { + assert_eq!(error_signature("turn 78 failed"), "turn N failed"); + assert_eq!( + error_signature("session ses_f4ee01e5dffeKYy29CIT5h2Lst ended"), + "session ID ended" + ); + } + + #[test] + fn error_signature_terminates_on_punctuation_and_spaces() { + // Regression: an earlier version peeked without advancing, which + // looped forever on the first alphabetic run. + for msg in [ + "exit 128: Internal error: Failed to dispatch hook event", + "a.b c_d 12 e-f", + " leading and trailing ", + "", + ] { + let _ = error_signature(msg); + } + } + + #[test] + fn a_whole_file_read_keeps_its_first_entry() { + // Regression: the tail-trim that drops a partial first line after a + // seek was also applied when the whole file had been read, silently + // losing the oldest entry on every small log. + let dir = tmp_repo(); + let root = dir.path(); + std::fs::create_dir_all(root.join(".atomic")).unwrap(); + std::fs::write( + root.join(".atomic/hook-errors.log"), + b"2026-09-25T20:00:00Z ses_aaa before-tool first entry\n\ + 2026-09-25T20:00:01Z ses_aaa before-tool second entry\n", + ) + .unwrap(); + + let report = read_errors(root).expect("log is read"); + assert_eq!(report.total, 2); + assert_eq!(report.first_seen.as_deref(), Some("2026-09-25T20:00:00Z")); + } + + #[test] + fn long_messages_are_truncated() { + let long = "x".repeat(500); + let out = truncate_detail(&long); + assert!(out.chars().count() <= MAX_DETAIL_LEN + 3); + assert!(out.ends_with("...")); + } + + #[test] + fn relative_age_uses_coarse_units() { + let now = "2026-09-25T18:00:00+00:00"; + assert_eq!(relative_age(now, "2026-09-25T17:59:30+00:00"), "30s ago"); + assert_eq!(relative_age(now, "2026-09-25T17:45:00+00:00"), "15m ago"); + assert_eq!(relative_age(now, "2026-09-25T14:00:00+00:00"), "4h ago"); + assert_eq!(relative_age(now, "2026-09-20T18:00:00+00:00"), "5d ago"); + } +} diff --git a/atomic-agent/src/hook_log.rs b/atomic-agent/src/hook_log.rs new file mode 100644 index 00000000..2a99ea81 --- /dev/null +++ b/atomic-agent/src/hook_log.rs @@ -0,0 +1,333 @@ +//! Session correlation and a tracing layer for durable operational diagnostics. + +use std::collections::BTreeMap; +use std::sync::{Arc, Mutex}; + +pub use atomic_repository::redb_change_store::SessionLogEvent; +use tracing::field::{Field, Visit}; +use tracing::{Event, Subscriber}; +use tracing_subscriber::layer::{Context, SubscriberExt}; +use tracing_subscriber::Layer; + +/// Correlation survives process boundaries through payload fields or explicit +/// `ATOMIC_SESSION_ID` / `ATOMIC_PARENT_SESSION_ID` environment variables. +#[derive(Clone, Debug, Default)] +pub struct HookContext { + pub session_id: Option, + pub parent_session_id: Option, +} + +fn usable(value: Option<&str>) -> Option { + value + .filter(|value| !value.is_empty() && *value != "unknown") + .map(str::to_owned) +} + +impl HookContext { + /// Best-effort extraction keeps invalid JSON attributable when the caller + /// supplies correlation via the environment; otherwise it stays unassigned. + pub fn from_input(input: &[u8]) -> Self { + let value = serde_json::from_slice::(input).ok(); + Self::from_json(value.as_ref()).with_environment() + } + + pub fn from_event(event: &crate::event::TurnEvent) -> Self { + Self::from_event_payload(event).with_environment() + } + + /// Service-side extraction must use caller fields, never daemon env vars. + pub fn from_event_payload(event: &crate::event::TurnEvent) -> Self { + let mut context = Self::from_json(event.raw_json.as_ref()); + context.session_id = usable(Some(&event.session_id)).or(context.session_id); + context + } + + fn with_environment(mut self) -> Self { + self.session_id = self + .session_id + .or_else(|| usable(std::env::var("ATOMIC_SESSION_ID").ok().as_deref())); + self.parent_session_id = self + .parent_session_id + .or_else(|| usable(std::env::var("ATOMIC_PARENT_SESSION_ID").ok().as_deref())); + self + } + + fn from_json(value: Option<&serde_json::Value>) -> Self { + fn field(value: Option<&serde_json::Value>, names: &[&str]) -> Option { + names + .iter() + .find_map(|name| usable(value?.get(name)?.as_str())) + } + let nested = value.and_then(|value| value.get("raw_json")); + Self { + session_id: field( + value, + &["session_id", "sessionID", "thread-id", "conversation_id"], + ), + parent_session_id: field(value, &["parent_session_id", "parentSessionID", "parentID"]) + .or_else(|| { + field( + nested, + &["parent_session_id", "parentSessionID", "parentID"], + ) + }), + } + } +} + +#[derive(Default)] +struct Fields(BTreeMap); +impl Visit for Fields { + fn record_str(&mut self, field: &Field, value: &str) { + self.0.insert(field.name().into(), value.into()); + } + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + self.0.insert(field.name().into(), format!("{value:?}")); + } +} + +// Capture synchronously, then persist outside the subscriber callback. This +// avoids blocking/re-entering the async service runtime from tracing's on_event. +struct HookLogLayer(Arc>>); +impl Layer for HookLogLayer { + fn on_event(&self, event: &Event<'_>, _: Context<'_, S>) { + if event.metadata().target() != "atomic::hook" { + return; + } + let mut fields = Fields::default(); + event.record(&mut fields); + let mut get = |name: &str| fields.0.remove(name).unwrap_or_default(); + let event = SessionLogEvent { + event_id: get("event_id"), + session_id: usable(Some(&get("session_id"))), + parent_session_id: usable(Some(&get("parent_session_id"))), + agent: get("agent"), + display_name: get("display_name"), + verb: get("verb"), + outcome: get("outcome"), + detail: Some(get("detail")).filter(|detail| !detail.is_empty()), + at: get("at"), + }; + *self.0.lock().unwrap() = Some(event); + } +} + +/// Emit structured tracing fields and synchronously commit the captured event +/// through the caller's service-owned sink. Logging failure preserves dispatch. +/// +/// # Examples +/// +/// ``` +/// use atomic_agent::hook_log::{record_result, HookContext}; +/// let context = HookContext { +/// session_id: Some("child-session".into()), +/// parent_session_id: Some("parent-session".into()), +/// }; +/// let result = record_result(&context, "opencode", "OpenCode", "receive", +/// Ok::<_, String>(42), |event| { +/// assert_eq!(event.parent_session_id.as_deref(), Some("parent-session")); +/// // Forward this event to the service-owned append operation. +/// Ok(()) +/// }); +/// assert_eq!(result.unwrap(), 42); +/// ``` +pub fn record_result( + context: &HookContext, + agent: &str, + display_name: &str, + verb: &str, + result: Result, + sink: impl FnOnce(&SessionLogEvent) -> Result<(), String>, +) -> Result { + let captured = Arc::new(Mutex::new(None)); + let subscriber = tracing_subscriber::registry().with(HookLogLayer(captured.clone())); + let event_id = uuid::Uuid::new_v4().to_string(); + let at = chrono::Utc::now().to_rfc3339(); + let detail = result + .as_ref() + .err() + .map(ToString::to_string) + .unwrap_or_default(); + let outcome = if result.is_ok() { "ok" } else { "error" }; + tracing::subscriber::with_default(subscriber, || { + tracing::info!(target: "atomic::hook", + event_id = event_id.as_str(), at = at.as_str(), agent, display_name, verb, outcome, + session_id = context.session_id.as_deref().unwrap_or(""), + parent_session_id = context.parent_session_id.as_deref().unwrap_or(""), + detail = detail.as_str(), "hook dispatch completed"); + }); + let event = captured.lock().unwrap().take(); + if let Some(event) = event { + if let Err(error) = sink(&event) { + log::warn!("failed to persist hook log: {error}"); + } + } + result +} + +/// Durable transport-outage spool. Each event has its own immutable file, so +/// independent processes never share a read-modify-write critical section. +pub fn spool(root: &std::path::Path, event: &SessionLogEvent) -> Result<(), String> { + use std::io::Write; + let id = uuid::Uuid::parse_str(&event.event_id).map_err(|error| error.to_string())?; + let dir = root.join(".atomic/hook-log-pending"); + std::fs::create_dir_all(&dir).map_err(|error| error.to_string())?; + let temporary = dir.join(format!("{}.tmp", uuid::Uuid::new_v4())); + let result = (|| { + let mut file = std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&temporary)?; + file.write_all(&serde_json::to_vec(event)?)?; + file.sync_all()?; + std::fs::rename(&temporary, dir.join(format!("{id}.json"))) + })() + .map_err(|error: std::io::Error| error.to_string()); + if result.is_err() { + let _ = std::fs::remove_file(temporary); + } + result +} + +/// Pending transport failures remain inspectable before a service recovers. +pub fn pending( + root: &std::path::Path, +) -> Result, String> { + let dir = root.join(".atomic/hook-log-pending"); + let entries = match std::fs::read_dir(dir) { + Ok(entries) => entries, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), + Err(error) => return Err(error.to_string()), + }; + let mut events = Vec::new(); + for entry in entries { + let path = entry.map_err(|error| error.to_string())?.path(); + if path.extension().is_none_or(|extension| extension != "json") { + continue; + } + let bytes = match std::fs::read(&path) { + Ok(bytes) => bytes, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue, + Err(error) => return Err(error.to_string()), + }; + events.push(( + path, + serde_json::from_slice(&bytes).map_err(|error| error.to_string())?, + )); + } + Ok(events) +} + +/// Replay pending events idempotently while the service owns the database gate. +pub fn flush_pending( + root: &std::path::Path, + store: &atomic_repository::redb_change_store::RedbChangeStore, +) -> Result<(), String> { + for (path, event) in pending(root)? { + store + .append_session_log(&event) + .map_err(|error| error.to_string())?; + let _ = std::fs::remove_file(path); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn tracing_keeps_concurrent_contexts_and_dispatch_errors_intact() { + let events = Mutex::new(Vec::new()); + std::thread::scope(|scope| { + for index in 0..32 { + let events = &events; + scope.spawn(move || { + let context = HookContext { + session_id: Some(format!("child-{index}")), + parent_session_id: Some("parent".into()), + }; + let result = record_result( + &context, + "opencode", + "OpenCode", + "before-tool", + Err::<(), _>(format!("failure-{index}")), + |event| { + events.lock().unwrap().push(event.clone()); + Ok(()) + }, + ); + assert_eq!(result.unwrap_err(), format!("failure-{index}")); + }); + } + }); + let events = events.into_inner().unwrap(); + assert_eq!(events.len(), 32); + let ids: std::collections::HashSet<_> = + events.iter().map(|event| &event.event_id).collect(); + assert_eq!(ids.len(), 32); + for event in events { + let index = event + .session_id + .unwrap() + .strip_prefix("child-") + .unwrap() + .to_string(); + assert_eq!( + event.detail.as_deref(), + Some(format!("failure-{index}").as_str()) + ); + assert_eq!(event.parent_session_id.as_deref(), Some("parent")); + } + } + + #[test] + fn a_failed_sink_preserves_the_dispatch_result() { + let result = record_result( + &HookContext::default(), + "opencode", + "OpenCode", + "receive", + Ok::<_, String>(42), + |_| Err("offline".into()), + ); + assert_eq!(result.unwrap(), 42); + } + + #[test] + fn concurrent_outage_spool_replays_idempotently() { + let dir = tempfile::tempdir().unwrap(); + std::thread::scope(|scope| { + for index in 0..24 { + let root = dir.path(); + scope.spawn(move || { + let context = HookContext { + session_id: Some(format!("child-{index}")), + parent_session_id: Some("parent".into()), + }; + record_result( + &context, + "opencode", + "OpenCode", + "receive", + Err::<(), _>("transport failed"), + |event| spool(root, event), + ) + .unwrap_err(); + }); + } + }); + let events = pending(dir.path()).unwrap(); + assert_eq!(events.len(), 24); + let store = atomic_repository::redb_change_store::RedbChangeStore::open( + dir.path().join("log.redb"), + ) + .unwrap(); + // Simulate a lost ACK: one pending event was already committed. + store.append_session_log(&events[0].1).unwrap(); + flush_pending(dir.path(), &store).unwrap(); + assert!(pending(dir.path()).unwrap().is_empty()); + assert_eq!(store.load_session_log(None, 0, 512).unwrap().len(), 24); + } +} diff --git a/atomic-agent/src/lib.rs b/atomic-agent/src/lib.rs index 78c4b00a..dab26d13 100644 --- a/atomic-agent/src/lib.rs +++ b/atomic-agent/src/lib.rs @@ -87,6 +87,8 @@ pub mod envelope; pub mod error; pub mod event; pub mod export; +pub mod hook_health; +pub mod hook_log; pub mod hooks; pub mod identity; pub mod integrations; diff --git a/atomic-cli/src/commands/agent/hook_log.rs b/atomic-cli/src/commands/agent/hook_log.rs new file mode 100644 index 00000000..83b76379 --- /dev/null +++ b/atomic-cli/src/commands/agent/hook_log.rs @@ -0,0 +1,95 @@ +//! Hook diagnostics use the same local/reactor service seam as provenance. + +use std::path::Path; + +use atomic_agent::hook_health::HookHealth; +pub use atomic_agent::hook_log::HookContext; +use atomic_agent::hook_log::SessionLogEvent; +use libatomic::atomic as pb; + +use crate::error::CliResult; +use crate::service::Service; + +pub fn append(root: &Path, event: &SessionLogEvent) -> Result<(), String> { + if let Err(error) = append_service(root, event) { + log::debug!("hook log service unavailable, spooling event: {error}"); + atomic_agent::hook_log::spool(root, event)?; + } + Ok(()) +} + +fn append_service(root: &Path, event: &SessionLogEvent) -> Result<(), String> { + let service = Service::open_root(root) + .map_err(|error| error.to_string())? + .ok_or_else(|| "not an Atomic repository".to_string())?; + service + .append_hook_log(pb::AppendHookLogRequest { + repository: Some(service.reference.clone()), + meta: crate::commands::rpc::request_meta(), + event: Some(libatomic::daemon::services_provenance::hook_log_proto( + event.clone(), + )), + }) + .map_err(|error| error.to_string())?; + Ok(()) +} + +pub fn record_result( + root: &Path, + context: &HookContext, + agent: &str, + display: &str, + verb: &str, + result: Result, +) -> Result { + atomic_agent::hook_log::record_result(context, agent, display, verb, result, |event| { + append(root, event) + }) +} + +pub fn read(root: &Path) -> CliResult> { + let Some(service) = Service::open_root(root)? else { + return Ok(None); + }; + let mut health: Option = None; + let mut after_sequence = 0; + loop { + let page = service.load_hook_log(pb::LoadHookLogRequest { + repository: Some(service.reference.clone()), + meta: crate::commands::rpc::request_meta(), + session_id: None, + after_sequence, + limit: 512, + })?; + if page.entries.is_empty() { + break; + } + for entry in page.entries { + if entry.sequence <= after_sequence { + return Err(crate::error::CliError::Internal(anyhow::anyhow!( + "hook log cursor did not advance" + ))); + } + after_sequence = entry.sequence; + if let Some(event) = entry.event { + let event = libatomic::daemon::services_provenance::hook_log_domain(event); + health + .get_or_insert_with(|| HookHealth::empty(&event.at)) + .record_event(&event); + } + } + } + for (_, event) in atomic_agent::hook_log::pending(root) + .map_err(|error| crate::error::CliError::Internal(anyhow::anyhow!(error)))? + { + health + .get_or_insert_with(|| HookHealth::empty(&event.at)) + .record_event(&event); + } + Ok(health) +} + +pub fn pending_health(root: &Path) -> Option { + let events = atomic_agent::hook_log::pending(root).ok()?; + HookHealth::from_events(events.into_iter().map(|(_, event)| event).collect()) +} diff --git a/atomic-cli/src/commands/agent/hooks.rs b/atomic-cli/src/commands/agent/hooks.rs index 1d80536e..b2f1c08c 100644 --- a/atomic-cli/src/commands/agent/hooks.rs +++ b/atomic-cli/src/commands/agent/hooks.rs @@ -167,24 +167,42 @@ impl Command for Hooks { })?; // Parse the agent-specific JSON into a common TurnEvent - let event = agent.parse_event(hook_type, &input).map_err(|e| { - // Log to stderr but don't fail hard on parse errors for - // non-critical hooks (tool use events) - if hook_type.is_tool_use() { - eprintln!( - "[atomic] Warning: failed to parse {} {} input: {}", - self.agent_name, self.verb, e - ); - // Return a generic event so we can continue - return CliError::Internal(anyhow!("Hook parse failed: {}", e)); + let event = match agent.parse_event(hook_type, &input) { + Ok(event) => event, + Err(e) => { + // The hook ran but could not be understood — an agent + // upgrade that changed the payload shape lands here, and + // without a record it is indistinguishable from a hook that + // was never installed. Best-effort: the repo root is not + // resolved yet, so this is skipped outside a workspace. + if let Ok(root) = find_repository_root() { + let _ = super::hook_log::record_result::<(), _>( + &root, + &super::hook_log::HookContext::from_input(&input), + agent.name(), + agent.display_name(), + &self.verb, + Err(&e), + ); + } + // Log to stderr but don't fail hard on parse errors for + // non-critical hooks (tool use events) + if hook_type.is_tool_use() { + eprintln!( + "[atomic] Warning: failed to parse {} {} input: {}", + self.agent_name, self.verb, e + ); + // Return a generic event so we can continue + return Err(CliError::Internal(anyhow!("Hook parse failed: {}", e))); + } + return Err(CliError::Internal(anyhow!( + "Failed to parse hook input for {} {}: {}", + self.agent_name, + self.verb, + e + ))); } - CliError::Internal(anyhow!( - "Failed to parse hook input for {} {}: {}", - self.agent_name, - self.verb, - e - )) - })?; + }; // Find the repository root. Agents whose hooks run detached from the // workspace (e.g., Antigravity plugin hooks, which execute with the @@ -239,6 +257,7 @@ impl Command for Hooks { let agent_name = agent.name().to_string(); let agent_display = agent.display_name().to_string(); + let hook_context = super::hook_log::HookContext::from_event(&event); // Keep hooks quiet on stdout/stderr — Claude Code surfaces both, so // route diagnostics through `log` instead. @@ -249,7 +268,11 @@ impl Command for Hooks { self.verb ); - let result = rt.block_on(async { + // Record liveness before propagating, so a failing dispatch still + // leaves evidence that the hook ran. Without this the shell guard's + // `|| true` makes a dead hook indistinguishable from one that was + // never installed. + let dispatch = rt.block_on(async { // Create the orchestrator let mut orchestrator = atomic_agent::turn::orchestrator::TurnOrchestrator::new(&repo_root) @@ -297,7 +320,16 @@ impl Command for Hooks { .map_err(|e| CliError::Internal(anyhow!("Failed to dispatch hook event: {}", e)))?; Ok::<_, CliError>(dispatch_result) - })?; + }); + + let result = super::hook_log::record_result( + &repo_root, + &hook_context, + &agent_name, + &agent_display, + &self.verb, + dispatch, + )?; // Log warnings via log crate (not stderr — that leaks into agent TUIs) for warning in &result.warnings { diff --git a/atomic-cli/src/commands/agent/mod.rs b/atomic-cli/src/commands/agent/mod.rs index 67fa026a..dfaf7a5d 100644 --- a/atomic-cli/src/commands/agent/mod.rs +++ b/atomic-cli/src/commands/agent/mod.rs @@ -48,6 +48,8 @@ mod provenance_rpc; pub(crate) mod receive; mod status; +pub(crate) use atomic_agent::hook_health as health; +pub(crate) mod hook_log; use clap::{Args, Subcommand}; use crate::commands::Command; diff --git a/atomic-cli/src/commands/agent/receive.rs b/atomic-cli/src/commands/agent/receive.rs index c96ce47d..f1e18a4d 100644 --- a/atomic-cli/src/commands/agent/receive.rs +++ b/atomic-cli/src/commands/agent/receive.rs @@ -22,14 +22,10 @@ //! //! # Routing //! -//! While the daemon is reachable this command is a thin RPC client: the -//! typed event travels to `ProvenanceService.DispatchTurnEvent`, where the -//! daemon hosts the turn orchestrator against the repository databases it -//! owns — this process never opens redb. When the daemon is down the -//! in-process orchestrator fallback runs (auto mode only; `ATOMIC_RPC=1` -//! is strict and never falls back); its journal sink starts the daemon -//! per D4, so even the fallback's provenance flows over the service -//! layer. +//! Both service modes call `ProvenanceService.DispatchTurnEvent`: local mode +//! runs the libatomic handler in-process, and reactor mode calls it over the +//! daemon socket. The receiver records parsing and transport outcomes under +//! `receive`; the shared handler records dispatch outcomes by event type. use std::io::Read; use std::sync::Arc; @@ -72,19 +68,6 @@ pub struct ReceiveResult { impl Command for Receive { fn run(&self) -> CliResult<()> { - // The typed event arrives on stdin — no agent hook parsing here. - let mut input = Vec::new(); - std::io::stdin().read_to_end(&mut input).map_err(|e| { - CliError::Io(std::io::Error::new( - e.kind(), - format!("Failed to read typed event from stdin: {}", e), - )) - })?; - let event: TurnEvent = - serde_json::from_slice(&input).map_err(|e| CliError::InvalidArgument { - message: format!("Invalid typed TurnEvent JSON: {}", e), - })?; - let repo_root = crate::commands::find_repository_root_from(&self.repository)?; let registry = AgentRegistry::with_defaults(); @@ -96,21 +79,50 @@ impl Command for Receive { let agent_name = agent.name().to_string(); let agent_display = agent.display_name().to_string(); - // The identity selection chain resolves client-side (env > global - // setting > server profile) and rides the request. - let agent_identity = - super::identity::resolve_effective_agent_identity().map(|(name, _)| name); - - // The service area dispatches the SAME handler both transports - // serve: in-process over libatomic (local mode, the default) or - // over the daemon socket (reactor mode, D4). There is no second - // local dispatch implementation to fall back to. - let result = crate::commands::rpc::dispatch_turn_event( + // Keep the entry-point outcome separate from the service's event-type + // outcome: malformed stdin and transport failures never reach the + // service, but a later successful receive must clear those failures. + let mut hook_context = super::hook_log::HookContext::from_input(&[]); + let dispatch = (|| { + let mut input = Vec::new(); + std::io::stdin().read_to_end(&mut input).map_err(|e| { + CliError::Io(std::io::Error::new( + e.kind(), + format!("Failed to read typed event from stdin: {}", e), + )) + })?; + hook_context = super::hook_log::HookContext::from_input(&input); + let event: TurnEvent = + serde_json::from_slice(&input).map_err(|e| CliError::InvalidArgument { + message: format!("Invalid typed TurnEvent JSON: {}", e), + })?; + let mut event_context = super::hook_log::HookContext::from_event(&event); + event_context.parent_session_id = hook_context + .parent_session_id + .clone() + .or(event_context.parent_session_id); + hook_context = event_context; + + let agent_identity = + super::identity::resolve_effective_agent_identity().map(|(name, _)| name); + + // Both local and daemon transports call the same libatomic handler. + crate::commands::rpc::dispatch_turn_event( + &repo_root, + &event, + &agent_name, + &agent_display, + agent_identity, + &hook_context, + ) + })(); + let result = super::hook_log::record_result( &repo_root, - &event, + &hook_context, &agent_name, &agent_display, - agent_identity, + "receive", + dispatch, )?; for warning in &result.warnings { diff --git a/atomic-cli/src/commands/agent/status.rs b/atomic-cli/src/commands/agent/status.rs index 498582b7..36d58850 100644 --- a/atomic-cli/src/commands/agent/status.rs +++ b/atomic-cli/src/commands/agent/status.rs @@ -107,6 +107,25 @@ struct StatusJson { /// broken session store into an indistinguishable "no sessions". #[serde(skip_serializing_if = "Option::is_none")] sessions_error: Option, + /// Hook liveness, per agent, for the agents that have fired at least + /// once. Absent when nothing has fired — which is a real and important + /// answer for a repository whose hooks are installed but dead, and is + /// why the human output words that case explicitly rather than omitting + /// it. + #[serde(skip_serializing_if = "Option::is_none")] + hook_health: Option>, + #[serde(skip_serializing_if = "Vec::is_empty")] + hook_sessions: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + hook_health_error: Option, + /// A bounded, aggregated read of `.atomic/hook-errors.log`. Absent when + /// the file does not exist. + #[serde(skip_serializing_if = "Option::is_none")] + hook_errors: Option, + /// Agents with hooks installed that have never recorded a successful + /// dispatch. This is the failure the installed-hooks check cannot see. + #[serde(skip_serializing_if = "Vec::is_empty")] + silent_agents: Vec, } /// The effective agent identity, as data. @@ -166,6 +185,44 @@ impl AgentStatus { files_touched: sessions.iter().map(|s| s.files_touched.len()).sum(), }; + let (health, hook_health_error) = match super::hook_log::read(repo_root) { + Ok(health) => (health, None), + Err(error) => ( + super::hook_log::pending_health(repo_root), + Some(error.to_string()), + ), + }; + let hook_sessions = health + .as_ref() + .map(|health| health.sessions.values().cloned().collect()) + .unwrap_or_default(); + let hook_health = health + .as_ref() + .map(super::health::summarize) + .filter(|v| !v.is_empty()); + let hook_errors = super::health::read_errors(repo_root); + + // An agent whose hooks are installed but which has never recorded a + // successful dispatch is silently losing every turn. Report it by + // name so a caller does not have to correlate the two lists. + // + // Only when observations exist at all: with no events there is no + // evidence either way, and calling a freshly-upgraded repository + // "never recorded" would be a false alarm on every machine. + let silent_agents: Vec = if health.is_none() || hook_health_error.is_some() { + Vec::new() + } else { + installed + .iter() + .filter(|name| { + !hook_health.as_ref().is_some_and(|rows| { + rows.iter().any(|r| r.name == **name && r.last_ok.is_some()) + }) + }) + .map(|name| name.to_string()) + .collect() + }; + StatusJson { agents, sessions: sessions @@ -192,6 +249,11 @@ impl AgentStatus { }, ), sessions_error, + hook_health, + hook_sessions, + hook_health_error, + hook_errors, + silent_agents, } } } @@ -275,6 +337,9 @@ impl Command for AgentStatus { println!(); + // Hook health and failures + print_hook_health(&repo_root, &installed, ®istry); + // Sessions let session_store = match SessionStore::for_repo(&repo_root) { @@ -430,6 +495,102 @@ impl Command for AgentStatus { } } +/// Print hook liveness and aggregated failures. +/// +/// The two facts are printed together because either alone misleads: a +/// healthy log proves nothing about whether the recorder is running, and a +/// live recorder proves nothing about what it failed to do earlier. +fn print_hook_health(repo_root: &std::path::Path, installed: &[&str], registry: &AgentRegistry) { + let mut log_unavailable = false; + let health = super::hook_log::read(repo_root).unwrap_or_else(|error| { + log_unavailable = true; + eprintln!(" ⚠ Hook log unavailable: {error}"); + super::hook_log::pending_health(repo_root) + }); + let rows = health + .as_ref() + .map(super::health::summarize) + .unwrap_or_default(); + let errors = super::health::read_errors(repo_root); + + // With no observations there is no evidence either way, so stay + // quiet rather than accusing a freshly-upgraded repository of never + // having recorded. + let silent: Vec<&str> = if health.is_none() || log_unavailable { + Vec::new() + } else { + installed + .iter() + .copied() + .filter(|name| !rows.iter().any(|r| r.name == *name && r.last_ok.is_some())) + .collect() + }; + + if rows.is_empty() && errors.is_none() && silent.is_empty() { + return; + } + + // Liveness, one line per agent that has ever fired. + for row in &rows { + let display = registry.get(&row.name).map_or_else( + || row.display_name.clone(), + |a| a.display_name().to_string(), + ); + let marker = if row.last_ok.is_some() { "✓" } else { "⚠" }; + let failing = if row.failing_verbs.is_empty() { + String::new() + } else { + format!(" ⚠ failing: {}", row.failing_verbs.join(", ")) + }; + println!(" {marker} {display} — {}", row.liveness()); + if !failing.is_empty() { + println!("{failing}"); + } + } + + // Attribute outstanding failures to their sessions. + if let Some(health) = &health { + for session in health.sessions.values() { + let failing: Vec<_> = session + .health + .verbs + .iter() + .filter(|(_, verb)| verb.outcome == super::health::FireOutcome::Error) + .map(|(name, _)| name.as_str()) + .collect(); + if !failing.is_empty() { + println!( + " session {} (parent {}) — failing: {}", + session.session_id.as_deref().unwrap_or("unattributed"), + session.parent_session_id.as_deref().unwrap_or("—"), + failing.join(", ") + ); + } + } + } + + // Hooks installed but never recorded — the failure an installed-hooks + // check cannot see, and the one that cost a demo. + for name in &silent { + let display = registry + .get(name) + .map_or_else(|| (*name).to_string(), |a| a.display_name().to_string()); + println!(" ⚠ {display} — hooks installed; no successful dispatch recorded"); + } + + if let Some(errors) = &errors { + if errors.total > 0 { + println!(); + println!(" Hook errors: {} ({})", errors.total, errors.window); + for class in errors.classes.iter().take(5) { + println!(" {:>5} {}", class.count, class.example); + } + } + } + + println!(); +} + // Tests #[cfg(test)] diff --git a/atomic-cli/src/commands/rpc.rs b/atomic-cli/src/commands/rpc.rs index 7cf2558d..7c8d81ea 100644 --- a/atomic-cli/src/commands/rpc.rs +++ b/atomic-cli/src/commands/rpc.rs @@ -172,7 +172,7 @@ fn kg_edge_domain(edge: &pb::KgEdge) -> atomic_core::pristine::vault::KgEdge { } } -fn request_meta() -> Option { +pub(crate) fn request_meta() -> Option { Some(pb::RequestMeta { request_id: uuid::Uuid::new_v4().to_string(), observed_at: None, @@ -6870,6 +6870,7 @@ pub fn dispatch_turn_event( agent_id: &str, agent_display: &str, agent_identity: Option, + hook_context: &atomic_agent::hook_log::HookContext, ) -> CliResult { let Some(session) = Service::open_root(root)? else { return Err(CliError::InvalidArgument { @@ -6882,7 +6883,7 @@ pub fn dispatch_turn_event( agent_id: agent_id.to_string(), agent_display_name: Some(agent_display.to_string()), agent_identity, - event: Some(turn_event_body(event)), + event: Some(turn_event_body(event, hook_context)), })?; Ok(crate::commands::agent::receive::ReceiveResult { session_id: response.session_id, @@ -6897,9 +6898,13 @@ pub fn dispatch_turn_event( }) } -fn turn_event_body(event: &atomic_agent::event::TurnEvent) -> pb::TurnEventBody { +fn turn_event_body( + event: &atomic_agent::event::TurnEvent, + hook_context: &atomic_agent::hook_log::HookContext, +) -> pb::TurnEventBody { pb::TurnEventBody { session_id: event.session_id.clone(), + parent_session_id: hook_context.parent_session_id.clone(), event_type: match event.event_type { atomic_agent::event::HookType::SessionStart => "session_start", atomic_agent::event::HookType::SessionEnd => "session_end", diff --git a/atomic-cli/src/service/mod.rs b/atomic-cli/src/service/mod.rs index 1a72148a..703544f3 100644 --- a/atomic-cli/src/service/mod.rs +++ b/atomic-cli/src/service/mod.rs @@ -1109,6 +1109,46 @@ impl Service { }) } + pub fn append_hook_log( + &self, + request: pb::AppendHookLogRequest, + ) -> CliResult { + self.call(move |backend| async move { + match backend { + Backend::Local(state) => services_agent::ProvenanceImpl { state } + .append_hook_log(Request::new(request)) + .await + .map(|response| response.into_inner()), + Backend::Reactor(channel) => { + pb::provenance_service_client::ProvenanceServiceClient::new(channel) + .append_hook_log(request) + .await + .map(|response| response.into_inner()) + } + } + }) + } + + pub fn load_hook_log( + &self, + request: pb::LoadHookLogRequest, + ) -> CliResult { + self.call(move |backend| async move { + match backend { + Backend::Local(state) => services_agent::ProvenanceImpl { state } + .load_hook_log(Request::new(request)) + .await + .map(|response| response.into_inner()), + Backend::Reactor(channel) => { + pb::provenance_service_client::ProvenanceServiceClient::new(channel) + .load_hook_log(request) + .await + .map(|response| response.into_inner()) + } + } + }) + } + pub fn list_sessions( &self, request: pb::ListSessionsRequest, diff --git a/atomic-cli/tests/hook_health_integration_test.rs b/atomic-cli/tests/hook_health_integration_test.rs new file mode 100644 index 00000000..8d7db3cb --- /dev/null +++ b/atomic-cli/tests/hook_health_integration_test.rs @@ -0,0 +1,281 @@ +use std::io::Write; +use std::process::{Command, Output, Stdio}; + +use atomic_agent::event::{HookType, TurnEvent}; +use atomic_agent::hook_health::{FireOutcome, HookHealth}; +use atomic_repository::Repository; +use tempfile::TempDir; + +struct Fixture { + repo: TempDir, + home: TempDir, +} + +impl Fixture { + fn new() -> Self { + let repo = TempDir::new().unwrap(); + drop(Repository::init(repo.path()).unwrap()); + Self { + repo, + home: TempDir::new().unwrap(), + } + } + + fn command(&self, args: &[&str]) -> Command { + let mut command = Command::new(env!("CARGO_BIN_EXE_atomic")); + command + .args(args) + .current_dir(self.repo.path()) + .env("HOME", self.home.path()) + .env("USERPROFILE", self.home.path()) + .env("XDG_CONFIG_HOME", self.home.path().join("config")) + .env("APPDATA", self.home.path().join("config")) + .env("LOCALAPPDATA", self.home.path().join("local")) + .env("ATOMIC_SERVICE", "local") + .env_remove("ATOMIC_AGENT_IDENTITY") + .env_remove("ATOMIC_SESSION_ID") + .env_remove("ATOMIC_PARENT_SESSION_ID") + .env( + "ATOMIC_DAEMON_SOCKET", + self.home.path().join("missing.sock"), + ) + .env( + "ATOMIC_DAEMON_BIN", + self.home.path().join("missing-atomicd"), + ); + command + } + + fn fire(&self, args: &[&str], input: &[u8], reactor: bool) -> Output { + let mut command = self.command(args); + if reactor { + command.env("ATOMIC_SERVICE", "reactor"); + } + let mut child = command + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + child.stdin.take().unwrap().write_all(input).unwrap(); + child.wait_with_output().unwrap() + } + + fn health(&self) -> HookHealth { + let store = atomic_repository::redb_change_store::RedbChangeStore::open_existing( + self.repo.path().join(".atomic/atomic.redb"), + ) + .unwrap(); + let mut events: Vec<_> = store + .load_session_log(None, 0, 512) + .unwrap() + .into_iter() + .map(|entry| entry.event) + .collect(); + events.extend( + atomic_agent::hook_log::pending(self.repo.path()) + .unwrap() + .into_iter() + .map(|(_, event)| event), + ); + HookHealth::from_events(events).expect("hook log was recorded") + } +} + +#[test] +fn native_hook_reports_parse_and_initialization_failures() { + let fixture = Fixture::new(); + let args = [ + "agent", + "hooks", + "claude-code", + "session-start", + "--foreground", + ]; + let malformed = fixture.fire(&args, b"{", false); + assert!(!malformed.status.success()); + let health = fixture.health(); + let verb = &health.agents["claude-code"].verbs["session-start"]; + assert_eq!(verb.outcome, FireOutcome::Error); + assert!(verb.last_ok.is_none()); + + // Valid parsing, but the orchestrator cannot create its session store. + let sessions = fixture.repo.path().join(".atomic/sessions"); + if sessions.exists() { + std::fs::remove_dir_all(&sessions).unwrap(); + } + std::fs::write(sessions, "blocked").unwrap(); + let output = fixture.fire(&args, br#"{"session_id":"native-health"}"#, false); + assert!(!output.status.success()); + let health = fixture.health(); + let verb = &health.agents["claude-code"].verbs["session-start"]; + assert_eq!(verb.outcome, FireOutcome::Error); + assert!(verb + .detail + .as_deref() + .unwrap() + .contains("Failed to create orchestrator")); +} + +#[test] +fn typed_receive_reports_parsing_transport_and_service_outcomes() { + let fixture = Fixture::new(); + let args = ["agent", "receive", "--agent", "opencode", "--json"]; + let malformed = fixture.fire(&args, b"{", false); + assert!(!malformed.status.success()); + let health = fixture.health(); + let verb = &health.agents["opencode"].verbs["receive"]; + assert_eq!(verb.outcome, FireOutcome::Error); + assert!(verb.last_ok.is_none()); + + let mut event = + serde_json::to_value(TurnEvent::new("typed-health", HookType::SessionStart)).unwrap(); + event["parent_session_id"] = serde_json::json!("typed-parent"); + let event = serde_json::to_vec(&event).unwrap(); + let success = fixture.fire(&args, &event, false); + assert!( + success.status.success(), + "{}", + String::from_utf8_lossy(&success.stderr) + ); + let health = fixture.health(); + let verbs = &health.agents["opencode"].verbs; + assert_eq!(verbs["receive"].outcome, FireOutcome::Ok); + // This row is written inside libatomic, not by the CLI receiver. + assert_eq!(verbs["session_start"].outcome, FireOutcome::Ok); + let last_ok = verbs["receive"].last_ok.clone(); + { + let store = atomic_repository::redb_change_store::RedbChangeStore::open_existing( + fixture.repo.path().join(".atomic/atomic.redb"), + ) + .unwrap(); + let events = store + .load_session_log(Some("typed-health"), 0, 512) + .unwrap(); + assert_eq!(events.len(), 2); + assert!(events + .iter() + .all(|entry| entry.event.parent_session_id.as_deref() == Some("typed-parent"))); + } + + let failure = fixture.fire(&args, &event, true); + assert!(!failure.status.success()); + let health = fixture.health(); + let verb = &health.agents["opencode"].verbs["receive"]; + assert_eq!(verb.outcome, FireOutcome::Error); + assert_eq!(verb.last_ok, last_ok); + assert!(verb.detail.as_deref().unwrap().contains("routing failed")); + + let offline_status = fixture + .command(&["agent", "status", "--json"]) + .env("ATOMIC_SERVICE", "reactor") + .output() + .unwrap(); + assert!( + offline_status.status.success(), + "{}", + String::from_utf8_lossy(&offline_status.stderr) + ); + let offline_status: serde_json::Value = serde_json::from_slice(&offline_status.stdout).unwrap(); + assert!(offline_status["hook_health_error"].is_string()); + assert!(offline_status.get("silent_agents").is_none()); + assert_eq!( + offline_status["hook_sessions"][0]["session_id"], + "typed-health" + ); + + let recovered = fixture.fire(&args, &event, false); + assert!( + recovered.status.success(), + "{}", + String::from_utf8_lossy(&recovered.stderr) + ); + let health = fixture.health(); + assert_eq!( + health.agents["opencode"].verbs["receive"].outcome, + FireOutcome::Ok + ); + + let status = fixture + .command(&["agent", "status", "--json"]) + .output() + .unwrap(); + assert!(status.status.success()); + let status: serde_json::Value = serde_json::from_slice(&status.stdout).unwrap(); + let row = status["hook_health"] + .as_array() + .unwrap() + .iter() + .find(|row| row["name"] == "opencode") + .unwrap(); + assert_eq!(row["verbs_recorded"], 2); + // A later attributable success must not hide an unassigned parse failure. + assert_eq!(row["failing_verbs"], serde_json::json!(["receive"])); + assert_eq!(status["hook_sessions"].as_array().unwrap().len(), 2); + assert!(!fixture + .repo + .path() + .join(".atomic/hook-health.json") + .exists()); +} + +#[test] +fn concurrent_hook_processes_preserve_parent_child_and_unassigned_failures() { + let fixture = Fixture::new(); + std::thread::scope(|scope| { + for index in 0..16 { + let fixture = &fixture; + scope.spawn(move || { + let input = if index == 0 { + b"{".to_vec() + } else { + serde_json::to_vec(&serde_json::json!({ + "session_id": format!("child-{index}"), "parent_session_id": "parent", + "event_type": "invalid-event", + })) + .unwrap() + }; + let output = fixture.fire( + &["agent", "receive", "--agent", "opencode", "--json"], + &input, + false, + ); + assert!(!output.status.success()); + }); + } + }); + let store = atomic_repository::redb_change_store::RedbChangeStore::open_existing( + fixture.repo.path().join(".atomic/atomic.redb"), + ) + .unwrap(); + let mut events: Vec<_> = store + .load_session_log(None, 0, 512) + .unwrap() + .into_iter() + .map(|entry| entry.event) + .collect(); + events.extend( + atomic_agent::hook_log::pending(fixture.repo.path()) + .unwrap() + .into_iter() + .map(|(_, event)| event), + ); + let ids: std::collections::HashSet<_> = events.iter().map(|event| &event.event_id).collect(); + assert_eq!(ids.len(), 16); + assert_eq!( + events + .iter() + .filter(|event| event.session_id.is_none()) + .count(), + 1 + ); + assert!(events + .iter() + .filter(|event| event.session_id.is_some()) + .all(|event| event.parent_session_id.as_deref() == Some("parent"))); + assert!(!fixture + .repo + .path() + .join(".atomic/hook-health.json") + .exists()); +} diff --git a/atomic-core/src/pristine/schema.rs b/atomic-core/src/pristine/schema.rs index d498ed27..de331db6 100644 --- a/atomic-core/src/pristine/schema.rs +++ b/atomic-core/src/pristine/schema.rs @@ -111,6 +111,9 @@ pub(crate) fn visit_tables(visitor: &mut impl TableVisitor) -> PristineResult<() visitor.table(PROVENANCE_FINAL_HASHES)?; // Sessions + visitor.table(SESSION_LOG)?; + visitor.table(SESSION_LOG_IDS)?; + visitor.table(SESSION_LOG_INDEX)?; visitor.table(SESSION_EVENTS)?; visitor.table(SESSION_TODOS)?; visitor.table(SESSION_PHASES)?; diff --git a/atomic-core/src/pristine/tables.rs b/atomic-core/src/pristine/tables.rs index 2d43077a..60bb44e7 100644 --- a/atomic-core/src/pristine/tables.rs +++ b/atomic-core/src/pristine/tables.rs @@ -528,6 +528,14 @@ pub const PROVENANCE_EVENT_INDEX: TableDefinition<&[u8; 32], &[u8; 16]> = pub const PROVENANCE_FINAL_HASHES: TableDefinition<&[u8; 32], u64> = TableDefinition::new("provenance_final_hashes"); +/// Operational diagnostics, independent of provenance turn lifecycle. +pub const SESSION_LOG: TableDefinition = TableDefinition::new("session_log"); +/// Retry deduplication: event UUID → committed sequence. +pub const SESSION_LOG_IDS: TableDefinition<&str, u64> = TableDefinition::new("session_log_ids"); +/// File-independent session lookup: (BLAKE3(session ID), sequence BE) → sequence. +pub const SESSION_LOG_INDEX: TableDefinition<&[u8; 40], u64> = + TableDefinition::new("session_log_index"); + // Session Tables (provenance-derived session data) // // Populated from every provenance graph, regardless of the producing agent. diff --git a/atomic-repository/src/redb_change_store/mod.rs b/atomic-repository/src/redb_change_store/mod.rs index ff075cb8..1269e8d3 100644 --- a/atomic-repository/src/redb_change_store/mod.rs +++ b/atomic-repository/src/redb_change_store/mod.rs @@ -16,6 +16,7 @@ mod batch; mod provenance; mod queries; +mod session_log; pub use batch::StoredSection; pub use provenance::{ @@ -25,6 +26,7 @@ pub use provenance::{ StoredProvenanceEvent, StoredProvenanceTurn, MAX_FROZEN_PAGE_FRAGMENTS, }; pub use queries::{StoreStats, StoredContentChunk}; +pub use session_log::{SessionLogEvent, StoredSessionLogEvent}; use atomic_core::change::format_v3::{self, ChangeReader, FormatError, SectionType}; use atomic_core::change::{Change, ChangeHeader}; @@ -316,6 +318,9 @@ impl RedbChangeStore { let _ = txn.open_table(tables::CHANGE_UNHASHED)?; let _ = txn.open_table(tables::CHANGE_SIGNATURES)?; Self::initialize_provenance_tables(&txn)?; + let _ = txn.open_table(tables::SESSION_LOG)?; + let _ = txn.open_table(tables::SESSION_LOG_IDS)?; + let _ = txn.open_table(tables::SESSION_LOG_INDEX)?; } txn.commit()?; } diff --git a/atomic-repository/src/redb_change_store/session_log.rs b/atomic-repository/src/redb_change_store/session_log.rs new file mode 100644 index 00000000..4d354cdb --- /dev/null +++ b/atomic-repository/src/redb_change_store/session_log.rs @@ -0,0 +1,213 @@ +//! Immutable operational diagnostics alongside the session/provenance ledger. + +use atomic_core::pristine::tables::{SESSION_LOG, SESSION_LOG_IDS, SESSION_LOG_INDEX}; +use redb::{ReadableDatabase, ReadableTable}; +use serde::{Deserialize, Serialize}; + +use super::{RedbChangeStore, RedbStoreError, RedbStoreResult}; + +/// A single hook observation. `session_id == None` retains pre-parse failures. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct SessionLogEvent { + pub event_id: String, + pub session_id: Option, + pub parent_session_id: Option, + pub agent: String, + pub display_name: String, + pub verb: String, + /// `ok` or `error`. + pub outcome: String, + pub detail: Option, + /// RFC3339 observation time; retries retain this timestamp and event ID. + pub at: String, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct StoredSessionLogEvent { + pub sequence: u64, + pub event: SessionLogEvent, +} + +fn session_key(session: &str, sequence: u64) -> [u8; 40] { + let mut key = [0; 40]; + key[..32].copy_from_slice(atomic_core::types::Hash::of(session.as_bytes()).as_bytes()); + key[32..].copy_from_slice(&sequence.to_be_bytes()); + key +} + +impl RedbChangeStore { + /// Append and index one immutable event atomically. Identical retries return + /// the original sequence; reusing an ID for different data is rejected. + /// No running turn or session ledger record is required. + pub fn append_session_log(&self, event: &SessionLogEvent) -> RedbStoreResult { + if event.event_id.is_empty() + || event.agent.is_empty() + || event.verb.is_empty() + || !matches!(event.outcome.as_str(), "ok" | "error") + || chrono::DateTime::parse_from_rfc3339(&event.at).is_err() + { + return Err(RedbStoreError::Serialization( + "invalid session log event".into(), + )); + } + let bytes = serde_json::to_vec(event)?; + let txn = self.db().begin_write()?; + let sequence; + { + let mut events = txn.open_table(SESSION_LOG)?; + let mut ids = txn.open_table(SESSION_LOG_IDS)?; + if let Some(existing) = ids.get(event.event_id.as_str())? { + let sequence = existing.value(); + let stored = events.get(sequence)?.ok_or_else(|| { + RedbStoreError::Corrupt("session log index points to missing event".into()) + })?; + if stored.value() != bytes { + return Err(RedbStoreError::Serialization( + "session log event ID conflict".into(), + )); + } + return Ok(sequence); + } + sequence = events + .last()? + .map_or(Some(1), |(key, _)| key.value().checked_add(1)) + .ok_or_else(|| RedbStoreError::Corrupt("session log sequence exhausted".into()))?; + events.insert(sequence, bytes.as_slice())?; + ids.insert(event.event_id.as_str(), sequence)?; + if let Some(session) = &event.session_id { + txn.open_table(SESSION_LOG_INDEX)? + .insert(&session_key(session, sequence), sequence)?; + } + } + txn.commit()?; + Ok(sequence) + } + + /// Read a bounded page in committed order. `after` is exclusive; pass the + /// last returned sequence for continuation. A session filter uses its index. + pub fn load_session_log( + &self, + session: Option<&str>, + after: u64, + limit: usize, + ) -> RedbStoreResult> { + let Some(start) = after.checked_add(1) else { + return Ok(Vec::new()); + }; + let txn = self.db().begin_read()?; + let events = txn.open_table(SESSION_LOG)?; + let mut out = Vec::new(); + let limit = limit.clamp(1, 512); + if let Some(session) = session { + let index = txn.open_table(SESSION_LOG_INDEX)?; + let lo = session_key(session, start); + let hi = session_key(session, u64::MAX); + for row in index.range::<&[u8; 40]>(&lo..=&hi)?.take(limit) { + let (_, sequence) = row?; + let sequence = sequence.value(); + let value = events.get(sequence)?.ok_or_else(|| { + RedbStoreError::Corrupt("session log index points to missing event".into()) + })?; + out.push(StoredSessionLogEvent { + sequence, + event: serde_json::from_slice(value.value())?, + }); + } + } else { + for row in events.range(start..)?.take(limit) { + let (key, value) = row?; + out.push(StoredSessionLogEvent { + sequence: key.value(), + event: serde_json::from_slice(value.value())?, + }); + } + } + Ok(out) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + + fn event(id: &str, session: Option<&str>) -> SessionLogEvent { + SessionLogEvent { + event_id: id.into(), + session_id: session.map(str::to_owned), + parent_session_id: session + .filter(|session| *session != "parent") + .map(|_| "parent".into()), + agent: "opencode".into(), + display_name: "OpenCode".into(), + verb: "before-tool".into(), + outcome: "ok".into(), + detail: None, + at: "2026-10-12T10:00:00Z".into(), + } + } + + #[test] + fn concurrent_parent_children_are_lossless_and_session_pages_cross_byte_boundaries() { + let dir = tempfile::tempdir().unwrap(); + let store = Arc::new(RedbChangeStore::open(dir.path().join("store.redb")).unwrap()); + std::thread::scope(|scope| { + for worker in 0..8 { + let store = store.clone(); + scope.spawn(move || { + let session = if worker == 0 { + "parent".to_string() + } else { + format!("child-{worker}") + }; + for index in 0..80 { + let event = event(&format!("{worker}-{index}"), Some(&session)); + let sequence = store.append_session_log(&event).unwrap(); + assert_eq!(store.append_session_log(&event).unwrap(), sequence); + } + }); + } + }); + let first = store.load_session_log(None, 0, 512).unwrap(); + assert_eq!(first.len(), 512); + let rest = store + .load_session_log(None, first.last().unwrap().sequence, 512) + .unwrap(); + assert_eq!(rest.len(), 128); + for worker in 0..8 { + let session = if worker == 0 { + "parent".to_string() + } else { + format!("child-{worker}") + }; + let page = store.load_session_log(Some(&session), 0, 41).unwrap(); + let next = store + .load_session_log(Some(&session), page.last().unwrap().sequence, 512) + .unwrap(); + assert_eq!(page.len() + next.len(), 80); + assert!(page + .iter() + .chain(&next) + .all(|entry| entry.event.session_id.as_deref() == Some(&session))); + assert!(page.last().unwrap().sequence < next[0].sequence); + } + } + + #[test] + fn idempotency_rejects_changed_payload_and_unassigned_events_survive_reopen() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("store.redb"); + let store = RedbChangeStore::open(&path).unwrap(); + let mut event = event("parse-error", None); + event.outcome = "error".into(); + assert_eq!(store.append_session_log(&event).unwrap(), 1); + event.detail = Some("changed".into()); + assert!(store.append_session_log(&event).is_err()); + drop(store); + let store = RedbChangeStore::open_existing(path).unwrap(); + let events = store.load_session_log(None, 0, 512).unwrap(); + assert_eq!(events.len(), 1); + assert!(events[0].event.session_id.is_none()); + assert!(events[0].event.detail.is_none()); + } +} diff --git a/docs/README.md b/docs/README.md index 2a0a2cd8..2df63d9b 100644 --- a/docs/README.md +++ b/docs/README.md @@ -7,6 +7,7 @@ Welcome to the Atomic VCS documentation. This directory contains comprehensive d | Document | Description | Audience | |----------|-------------|----------| | [CLI Reference](../atomic-cli/README.md) | Complete command reference for the `atomic` CLI | Users, Everyone | +| [Hook Health and Session Logs](./hook-health.md) | Diagnose hook failures and inspect per-session health | Users, Integration developers | | [Stack Walkthrough](./walkthrough-stacks.md) | Hands-on tutorial for stacks, apply, and stash | Users, Contributors | | [Attestation Design](./attestation-design.md) | Graph-level audit nodes for AI cost, tokens, and compliance | Developers, Contributors | | [Intent Identity](./intent-identity.md) | ULID + `PROJECT::author::seq` intent identity and reference resolution | Users, Contributors | @@ -21,6 +22,7 @@ Welcome to the Atomic VCS documentation. This directory contains comprehensive d - **New to Atomic?** Start with [COMPARISON.md](./COMPARISON.md) to understand what makes Atomic different - **Using the CLI?** See the [CLI Reference](../atomic-cli/README.md) for every command and option +- **Are agent hooks working?** See [Hook Health and Session Logs](./hook-health.md) for liveness, failures, and session correlation - **Learning stacks?** Walk through the [Stack Walkthrough](./walkthrough-stacks.md) step by step - **AI audit trail?** Read the [Attestation Design](./attestation-design.md) for graph-level cost and compliance tracking - **Working with intents?** See [Intent Identity](./intent-identity.md) for how they're named and referenced, and [Intent Knowledge Graph](./intent-graph.md) for querying them @@ -86,4 +88,4 @@ This creates HTML documentation from doc comments in the source code. --- -*This documentation is part of the Atomic VCS clean-room implementation.* \ No newline at end of file +*This documentation is part of the Atomic VCS clean-room implementation.* diff --git a/docs/adding-agents.md b/docs/adding-agents.md index 2ab4dd72..0eb4da3d 100644 --- a/docs/adding-agents.md +++ b/docs/adding-agents.md @@ -299,6 +299,22 @@ and never breaks the agent if something fails: test -d .atomic || test -f .atomic-sandbox && atomic agent hooks || true ``` +### Correlation and hook diagnostics + +Give each session a stable ID and preserve `parent_session_id` in the event's +`raw_json` when the agent exposes parent/child relationships. Bridges can also +pass `ATOMIC_SESSION_ID` and `ATOMIC_PARENT_SESSION_ID` in the hook process's +environment. The typed `agent receive` entry point resolves the caller's parent +ID and carries it explicitly on the service request. + +Native hooks and typed dispatch already emit shared operational diagnostics; +adapters use this reporting path automatically. Check `atomic agent status` +and `atomic agent status --json` after wiring an integration. Validate both +successful dispatch and malformed input, and check that concurrent children +keep their own session IDs and parent linkage. See +[Hook Health and Session Logs](./hook-health.md) for the report fields and +outage behavior. + --- ## Why each agent must be added explicitly diff --git a/docs/hook-health.md b/docs/hook-health.md new file mode 100644 index 00000000..f105b495 --- /dev/null +++ b/docs/hook-health.md @@ -0,0 +1,79 @@ +# Hook Health and Session Logs + +Use hook health to check whether agent callbacks are reaching Atomic and to +locate failures that would otherwise be hidden by the hook's shell guard. + +```bash +atomic agent status +atomic agent status --json +``` + +## Reading the report + +The human report shows each agent's most recent successful dispatch, its age +relative to the current time, and outstanding failing verbs. Session rows +attribute failures to a session and its parent. Existing plugin and git-shadow +errors are grouped from a bounded tail of `.atomic/hook-errors.log`. + +Health is tracked per **agent, session, and verb**. A successful child cannot +clear a parent's failure. A later successful dispatch of the same verb in the +same session clears that session's current failure while retaining the earlier +observations in the log. A successful dispatch may legitimately produce no new +change when the working copy is unchanged. + +### JSON fields + +| Field | Meaning | +|-------|---------| +| `hook_health` | Per-agent summary: `last_fired`, `last_ok`, `last_error`, `verbs_recorded`, and `failing_verbs` | +| `hook_sessions` | Session-level rows with `agent`, `session_id`, `parent_session_id`, and `health.verbs` | +| `hook_health_error` | Why the operational log could not be loaded; pending observations may still be reported | +| `hook_errors` | Grouped errors and recent entries from the bounded legacy error-log tail | +| `silent_agents` | Installed agents with no successful dispatch in the available operational log | + +Each entry in `health.verbs` contains `last_fired`, `outcome` (`ok` or `error`), +an optional failure `detail`, and an optional `last_ok`. Failure summaries flatten +whitespace and truncate long details. Missing session IDs appear as `null`. + +The hook-health fields are omitted when there are no observations. An empty +log gives no evidence that a freshly installed hook has failed. When the service +is unavailable and only pending observations are accessible, `silent_agents` +is omitted because the report is incomplete. + +With `jq` installed, inspect one session's current outcomes: + +```bash +atomic agent status --json \ + | jq '.hook_sessions[]? | select(.session_id == "ses_child")' +``` + +## Parent/child correlation + +Native adapters retain the session ID and agent context from the payload. +Integration bridges can supply `parent_session_id` in `raw_json`, or explicitly +set `ATOMIC_SESSION_ID` and `ATOMIC_PARENT_SESSION_ID` for the hook process. +Typed `agent receive` also accepts a top-level `parent_session_id` and forwards +the resolved parent on its service request. Correlation is resolved in the +caller so it survives process boundaries. + +If malformed input has no usable session ID, Atomic retains it as an unassigned +diagnostic. It remains visible without being attributed to a different session. + +## Storage and recovery + +Native hooks, typed receive, and the shared dispatch handler emit structured +Rust `tracing` observations. The existing local/reactor service layer appends +them transactionally to `.atomic/atomic.redb`, with event-ID deduplication and a +session index for paginated reads. Status projects health from these immutable +events. It no longer rewrites a shared `hook-health.json` snapshot. + +Operational observations are local diagnostic data, independent of the +running-turn provenance journal. Parsing failures and failures after a turn +closes can therefore be logged without requiring a running turn. + +During a service outage, each observation is spooled to its own immutable file +under `.atomic/hook-log-pending/`. Status can inspect these pending observations. +A successful append replays pending events into the database and removes their +spool files after commit. Retried events retain their original IDs and +observation times, so replay neither duplicates them nor makes an older failure +look newer than a later recovery. diff --git a/libatomic/Cargo.toml b/libatomic/Cargo.toml index 02d987dc..f55b64e1 100644 --- a/libatomic/Cargo.toml +++ b/libatomic/Cargo.toml @@ -40,6 +40,7 @@ ulid = { workspace = true } redb = { workspace = true } anyhow = { workspace = true } uuid = { workspace = true } +tracing = { workspace = true } [dev-dependencies] tempfile = { workspace = true } diff --git a/libatomic/proto/atomic/services/provenance/provenance_messages.proto b/libatomic/proto/atomic/services/provenance/provenance_messages.proto index d730cf57..fd91c901 100644 --- a/libatomic/proto/atomic/services/provenance/provenance_messages.proto +++ b/libatomic/proto/atomic/services/provenance/provenance_messages.proto @@ -55,6 +55,41 @@ import "google/protobuf/timestamp.proto"; import "atomic/common/common.proto"; import "atomic/common/pagination.proto"; +// Operational diagnostics survive outside the running-turn provenance journal. +message HookLogEvent { + string event_id = 1; + optional string session_id = 2; + optional string parent_session_id = 3; + string agent = 4; + string display_name = 5; + string verb = 6; + string outcome = 7; // ok | error + optional string detail = 8; + string at = 9; // RFC3339, retained across retries +} +message AppendHookLogRequest { + RepositoryRef repository = 1; + RequestMeta meta = 2; + HookLogEvent event = 3; +} +message AppendHookLogResponse { + uint64 sequence = 1; + ResponseMeta meta = 100; +} +message LoadHookLogRequest { + RepositoryRef repository = 1; + RequestMeta meta = 2; + optional string session_id = 3; + uint64 after_sequence = 4; + uint32 limit = 5; // bounded to 512 +} +message HookLogEntry { + uint64 sequence = 1; + HookLogEvent event = 2; +} +message LoadHookLogResponse { + repeated HookLogEntry entries = 1; +} // ————— turn & checkpoint state (source: redb_change_store/provenance.rs) ————— enum StopCause { @@ -404,6 +439,7 @@ message TurnEventBody { optional string tool_use_id = 5; google.protobuf.Timestamp timestamp = 6; optional bytes raw_json = 7; // opaque agent context (never parsed here) + optional string parent_session_id = 8; // client-resolved correlation across processes } message DispatchTurnEventResponse { string session_id = 1; diff --git a/libatomic/proto/atomic/services/provenance/provenance_service.proto b/libatomic/proto/atomic/services/provenance/provenance_service.proto index ab3e2985..47197415 100644 --- a/libatomic/proto/atomic/services/provenance/provenance_service.proto +++ b/libatomic/proto/atomic/services/provenance/provenance_service.proto @@ -10,6 +10,19 @@ import "atomic/common/options.proto"; import "atomic/services/provenance/provenance_messages.proto"; service ProvenanceService { + // Session-scoped operational events. Local diagnostics, no turn reservation. + rpc AppendHookLog(AppendHookLogRequest) returns (AppendHookLogResponse) { + option (atomic.execution_scope) = EXECUTION_SCOPE_LOCAL; + option (atomic.required_capability) = "provenance.write"; + option (atomic.allowed_caller) = CALLER_CLASS_LOCAL; + option (atomic.effect) = RPC_EFFECT_REPOSITORY_WRITE; + } + rpc LoadHookLog(LoadHookLogRequest) returns (LoadHookLogResponse) { + option (atomic.execution_scope) = EXECUTION_SCOPE_LOCAL; + option (atomic.required_capability) = "provenance.read"; + option (atomic.allowed_caller) = CALLER_CLASS_LOCAL; + option (atomic.effect) = RPC_EFFECT_REPOSITORY_READ; + } // Grants the generation for this session/turn. The first reserve claims // the session for the caller's view (authorization matrix). rpc ReserveTurn(ReserveTurnRequest) returns (ReserveTurnResponse) { diff --git a/libatomic/src/daemon/services.rs b/libatomic/src/daemon/services.rs index d50d6193..dcee39a8 100644 --- a/libatomic/src/daemon/services.rs +++ b/libatomic/src/daemon/services.rs @@ -67,6 +67,8 @@ const METHOD_MATRIX: &[(&str, &str)] = &[ ("AttestationService", "VerifyAttestation"), ("KnowledgeService", "QueryGraph"), ("ProvenanceService", "DispatchTurnEvent"), + ("ProvenanceService", "AppendHookLog"), + ("ProvenanceService", "LoadHookLog"), ("ProvenanceService", "ReserveTurn"), ("ProvenanceService", "AppendEnvelopes"), ("ProvenanceService", "PrepareCheckpoint"), @@ -120,6 +122,27 @@ impl daemon_service_server::DaemonService for DaemonImpl { allowed_callers: vec![CallerClass::Local as i32], effects: vec![RpcEffect::RepositoryWrite as i32], } + } else if *service == "ProvenanceService" + && matches!(*method, "AppendHookLog" | "LoadHookLog") + { + let write = *method == "AppendHookLog"; + MethodDescriptorInfo { + service: service.to_string(), + method: method.to_string(), + scope: ExecutionScope::Local as i32, + required_capabilities: vec![if write { + "provenance.write" + } else { + "provenance.read" + } + .to_string()], + allowed_callers: vec![CallerClass::Local as i32], + effects: vec![if write { + RpcEffect::RepositoryWrite + } else { + RpcEffect::RepositoryRead + } as i32], + } } else { MethodDescriptorInfo { service: service.to_string(), diff --git a/libatomic/src/daemon/services_agent.rs b/libatomic/src/daemon/services_agent.rs index 5e8fbef8..bc358bf0 100644 --- a/libatomic/src/daemon/services_agent.rs +++ b/libatomic/src/daemon/services_agent.rs @@ -3782,6 +3782,19 @@ pub struct ProvenanceImpl { #[tonic::async_trait] impl provenance_service_server::ProvenanceService for ProvenanceImpl { + async fn append_hook_log( + &self, + request: Request, + ) -> Result, Status> { + crate::daemon::services_provenance::append_hook_log_impl(&self.state, request).await + } + + async fn load_hook_log( + &self, + request: Request, + ) -> Result, Status> { + crate::daemon::services_provenance::load_hook_log_impl(&self.state, request).await + } async fn dispatch_turn_event( &self, request: Request, @@ -3798,65 +3811,119 @@ impl provenance_service_server::ProvenanceService for ProvenanceImpl { .clone() .unwrap_or_else(|| agent_id.clone()); let identity = request.agent_identity.clone(); - let body = request + let verb = request .event - .ok_or_else(|| domain_status(ErrorCode::InvalidArgument, "typed event required"))?; - let hook_type = hook_type(&body.event_type).ok_or_else(|| { - domain_status( - ErrorCode::InvalidArgument, - format!("unknown event type '{}'", body.event_type), - ) - })?; - let raw_json = body - .raw_json .as_ref() - .filter(|bytes| !bytes.is_empty()) - .and_then(|bytes| serde_json::from_slice::(bytes).ok()); - let event = TurnEvent { - session_id: body.session_id.clone(), - event_type: hook_type, - transcript_path: None, - prompt: body.prompt.clone(), - tool_name: body.tool_name.clone(), - tool_use_id: body.tool_use_id.clone(), - timestamp: body - .timestamp + .map(|body| body.event_type.clone()) + .unwrap_or_else(|| "dispatch_turn_event".to_string()); + let mut hook_context = atomic_agent::hook_log::HookContext::default(); + if let Some(body) = &request.event { + let raw_json = body + .raw_json .as_ref() - .map(timestamp_domain) - .unwrap_or_else(chrono::Utc::now), - raw_json, - }; + .and_then(|bytes| serde_json::from_slice(bytes).ok()); + let event = TurnEvent { + session_id: body.session_id.clone(), + raw_json, + ..TurnEvent::new( + &body.session_id, + atomic_agent::event::HookType::SessionStart, + ) + }; + hook_context = atomic_agent::hook_log::HookContext::from_event_payload(&event); + if body + .parent_session_id + .as_deref() + .is_some_and(|parent| !parent.is_empty()) + { + hook_context.parent_session_id = body.parent_session_id.clone(); + } + } + let dispatch = async { + let body = request + .event + .ok_or_else(|| domain_status(ErrorCode::InvalidArgument, "typed event required"))?; + let hook_type = hook_type(&body.event_type).ok_or_else(|| { + domain_status( + ErrorCode::InvalidArgument, + format!("unknown event type '{}'", body.event_type), + ) + })?; + let raw_json = body + .raw_json + .as_ref() + .filter(|bytes| !bytes.is_empty()) + .and_then(|bytes| serde_json::from_slice::(bytes).ok()); + let event = TurnEvent { + session_id: body.session_id.clone(), + event_type: hook_type, + transcript_path: None, + prompt: body.prompt.clone(), + tool_name: body.tool_name.clone(), + tool_use_id: body.tool_use_id.clone(), + timestamp: body + .timestamp + .as_ref() + .map(timestamp_domain) + .unwrap_or_else(chrono::Utc::now), + raw_json, + }; - // The daemon hosts the turn orchestrator: it owns the repository - // databases, so the dispatch (session view, recording, journal - // checkpoint) runs in-process with a direct journal sink. - let sink = DirectJournalSink::open(&handle.root) - .map_err(|error| domain_status(ErrorCode::ProvenanceStore, error))?; - let mut orchestrator = - TurnOrchestrator::new(handle.root.clone()) - .await - .map_err(|error| { - domain_status(ErrorCode::Internal, format!("orchestrator: {error}")) - })?; - orchestrator.set_agent(agent_id, display); - orchestrator.set_agent_identity(identity); - orchestrator.set_journal_sink(Arc::new(sink)); - let result = orchestrator.dispatch(event).await.map_err(|error| { - domain_status(ErrorCode::Internal, format!("dispatch failed: {error}")) - })?; + // Local and RPC transports share this handler and its health report. + let sink = DirectJournalSink::open(&handle.root) + .map_err(|error| domain_status(ErrorCode::ProvenanceStore, error))?; + let mut orchestrator = + TurnOrchestrator::new(handle.root.clone()) + .await + .map_err(|error| { + domain_status(ErrorCode::Internal, format!("orchestrator: {error}")) + })?; + orchestrator.set_agent(agent_id.clone(), display.clone()); + orchestrator.set_agent_identity(identity); + orchestrator.set_journal_sink(Arc::new(sink)); + let result = orchestrator.dispatch(event).await.map_err(|error| { + domain_status(ErrorCode::Internal, format!("dispatch failed: {error}")) + })?; - let recorded = result.change_recorded.as_ref(); - Ok(Response::new(DispatchTurnEventResponse { - session_id: result.session_id.clone(), - recorded: recorded.is_some(), - change_hash: recorded.map(|outcome| super::convert::hash_proto(&outcome.hash)), - view: result.view.clone(), - files: recorded - .map(|outcome| outcome.recorded_file_list().to_vec()) - .unwrap_or_default(), - warnings: result.warnings.clone(), - meta: response_meta(&meta), - })) + let recorded = result.change_recorded.as_ref(); + Ok(Response::new(DispatchTurnEventResponse { + session_id: result.session_id.clone(), + recorded: recorded.is_some(), + change_hash: recorded.map(|outcome| super::convert::hash_proto(&outcome.hash)), + view: result.view.clone(), + files: recorded + .map(|outcome| outcome.recorded_file_list().to_vec()) + .unwrap_or_default(), + warnings: result.warnings.clone(), + meta: response_meta(&meta), + })) + } + .await; + atomic_agent::hook_log::record_result( + &hook_context, + &agent_id, + &display, + &verb, + dispatch, + |event| { + let append = (|| { + let store = handle.change_store().map_err(|error| error.to_string())?; + store + .append_session_log(event) + .map_err(|error| error.to_string())?; + if let Err(error) = atomic_agent::hook_log::flush_pending(&handle.root, &store) + { + tracing::warn!("hook log spool: {error}"); + } + Ok::<_, String>(()) + })(); + if let Err(error) = append { + tracing::debug!("hook log storage unavailable, spooling event: {error}"); + return atomic_agent::hook_log::spool(&handle.root, event); + } + Ok(()) + }, + ) } async fn reserve_turn( diff --git a/libatomic/src/daemon/services_provenance.rs b/libatomic/src/daemon/services_provenance.rs index b8af17a0..8f827430 100644 --- a/libatomic/src/daemon/services_provenance.rs +++ b/libatomic/src/daemon/services_provenance.rs @@ -40,6 +40,113 @@ use super::provenance_core as core; use super::services_agent::response_meta; use super::state::{domain_status, DaemonState}; +pub fn hook_log_domain( + event: crate::atomic::HookLogEvent, +) -> atomic_repository::redb_change_store::SessionLogEvent { + atomic_repository::redb_change_store::SessionLogEvent { + event_id: event.event_id, + session_id: event.session_id, + parent_session_id: event.parent_session_id, + agent: event.agent, + display_name: event.display_name, + verb: event.verb, + outcome: event.outcome, + detail: event.detail, + at: event.at, + } +} + +pub fn hook_log_proto( + event: atomic_repository::redb_change_store::SessionLogEvent, +) -> crate::atomic::HookLogEvent { + crate::atomic::HookLogEvent { + event_id: event.event_id, + session_id: event.session_id, + parent_session_id: event.parent_session_id, + agent: event.agent, + display_name: event.display_name, + verb: event.verb, + outcome: event.outcome, + detail: event.detail, + at: event.at, + } +} + +pub async fn append_hook_log_impl( + state: &Arc, + request: Request, +) -> Result, Status> { + let request = request.into_inner(); + let reference = request + .repository + .as_ref() + .ok_or_else(|| Status::invalid_argument("repository required"))?; + let handle = state.resolve(reference)?; + state.log_rpc("AppendHookLog", Some(&handle)); + let gate_handle = handle.clone(); + let _gate = gate_handle.exclusive().await; + let event = hook_log_domain( + request + .event + .ok_or_else(|| Status::invalid_argument("event required"))?, + ); + let response = tokio::task::spawn_blocking(move || { + let store = handle.change_store()?; + let sequence = store + .append_session_log(&event) + .map_err(|error| domain_status(ErrorCode::ProvenanceStore, error.to_string()))?; + if let Err(error) = atomic_agent::hook_log::flush_pending(&handle.root, &store) { + tracing::warn!("hook log spool: {error}"); + } + Ok::<_, Status>(crate::atomic::AppendHookLogResponse { + sequence, + meta: response_meta(&request.meta), + }) + }) + .await + .map_err(|error| Status::internal(error.to_string()))??; + Ok(Response::new(response)) +} + +pub async fn load_hook_log_impl( + state: &Arc, + request: Request, +) -> Result, Status> { + let request = request.into_inner(); + let reference = request + .repository + .as_ref() + .ok_or_else(|| Status::invalid_argument("repository required"))?; + let handle = state.resolve(reference)?; + state.log_rpc("LoadHookLog", Some(&handle)); + let gate_handle = handle.clone(); + let _gate = gate_handle.exclusive().await; + let response = tokio::task::spawn_blocking(move || { + let store = handle.change_store()?; + let entries = store + .load_session_log( + request.session_id.as_deref(), + request.after_sequence, + if request.limit == 0 { + 512 + } else { + request.limit as usize + }, + ) + .map_err(|error| domain_status(ErrorCode::ProvenanceStore, error.to_string()))? + .into_iter() + .map(|entry| crate::atomic::HookLogEntry { + sequence: entry.sequence, + event: Some(hook_log_proto(entry.event)), + }) + .collect(); + Ok::<_, Status>(crate::atomic::LoadHookLogResponse { entries }) + }) + .await + .map_err(|error| Status::internal(error.to_string()))??; + Ok(Response::new(response)) +} + // --------------------------------------------------------------------------- // domain ↔ protobuf mapping // --------------------------------------------------------------------------- diff --git a/libatomic/tests/hook_health_dispatch.rs b/libatomic/tests/hook_health_dispatch.rs new file mode 100644 index 00000000..c35ec47c --- /dev/null +++ b/libatomic/tests/hook_health_dispatch.rs @@ -0,0 +1,272 @@ +use std::sync::Arc; + +use atomic_agent::hook_health::{FireOutcome, HookHealth}; +use atomic_repository::Repository; +use libatomic::atomic::provenance_service_client::ProvenanceServiceClient; +use libatomic::atomic::provenance_service_server::{ProvenanceService, ProvenanceServiceServer}; +use libatomic::atomic::{DispatchTurnEventRequest, RepositoryRef, TurnEventBody}; +use libatomic::daemon::services_agent::ProvenanceImpl; +use libatomic::daemon::state::DaemonState; +use tempfile::TempDir; +use tonic::{Request, Status}; + +enum Dispatch { + Local(ProvenanceImpl), + Rpc(ProvenanceServiceClient), +} + +impl Dispatch { + fn duplicate(&self) -> Self { + match self { + Self::Local(service) => Self::Local(ProvenanceImpl { + state: service.state.clone(), + }), + Self::Rpc(client) => Self::Rpc(client.clone()), + } + } + + async fn append(&mut self, request: libatomic::atomic::AppendHookLogRequest) -> u64 { + match self { + Self::Local(service) => { + service + .append_hook_log(Request::new(request)) + .await + .unwrap() + .into_inner() + .sequence + } + Self::Rpc(client) => { + client + .append_hook_log(request) + .await + .unwrap() + .into_inner() + .sequence + } + } + } + + async fn load( + &mut self, + reference: &RepositoryRef, + session: Option<&str>, + after: u64, + limit: u32, + ) -> Vec { + let request = libatomic::atomic::LoadHookLogRequest { + repository: Some(reference.clone()), + session_id: session.map(str::to_owned), + after_sequence: after, + limit, + ..Default::default() + }; + match self { + Self::Local(service) => { + service + .load_hook_log(Request::new(request)) + .await + .unwrap() + .into_inner() + .entries + } + Self::Rpc(client) => { + client + .load_hook_log(request) + .await + .unwrap() + .into_inner() + .entries + } + } + } + async fn fire(&mut self, request: DispatchTurnEventRequest) -> Result<(), Box> { + let result = match self { + Self::Local(service) => service + .dispatch_turn_event(Request::new(request)) + .await + .map(|_| ()), + Self::Rpc(client) => client.dispatch_turn_event(request).await.map(|_| ()), + }; + result.map_err(Box::new) + } +} + +fn request(reference: &RepositoryRef, event_type: &str) -> DispatchTurnEventRequest { + DispatchTurnEventRequest { + repository: Some(reference.clone()), + agent_id: "opencode".to_string(), + agent_display_name: Some("OpenCode".to_string()), + event: Some(TurnEventBody { + session_id: "service-health".to_string(), + event_type: event_type.to_string(), + ..Default::default() + }), + ..Default::default() + } +} + +fn read_health(root: &std::path::Path) -> HookHealth { + let store = atomic_repository::redb_change_store::RedbChangeStore::open_existing( + root.join(".atomic/atomic.redb"), + ) + .unwrap(); + HookHealth::from_events( + store + .load_session_log(None, 0, 512) + .unwrap() + .into_iter() + .map(|entry| entry.event) + .collect(), + ) + .unwrap() +} + +#[tokio::test] +async fn local_and_rpc_dispatch_record_success_validation_failure_and_recovery() { + for rpc in [false, true] { + let repo = TempDir::new().unwrap(); + drop(Repository::init(repo.path()).unwrap()); + let state = Arc::new(DaemonState::new()); + let reference = state.register(repo.path().to_path_buf()).repository_ref(); + let service = ProvenanceImpl { state }; + let mut tasks = Vec::new(); + let mut dispatch = if rpc { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let (send, receive) = tokio::sync::mpsc::channel(8); + tasks.push(tokio::spawn(async move { + while let Ok((stream, _)) = listener.accept().await { + if send.send(Ok::<_, std::io::Error>(stream)).await.is_err() { + break; + } + } + })); + tasks.push(tokio::spawn(async move { + tonic::transport::Server::builder() + .add_service(ProvenanceServiceServer::new(service)) + .serve_with_incoming(tokio_stream::wrappers::ReceiverStream::new(receive)) + .await + .unwrap(); + })); + Dispatch::Rpc( + ProvenanceServiceClient::connect(format!("http://{address}")) + .await + .unwrap(), + ) + } else { + Dispatch::Local(service) + }; + + dispatch + .fire(request(&reference, "session_start")) + .await + .unwrap(); + let health = read_health(repo.path()); + let verb = &health.agents["opencode"].verbs["session_start"]; + assert_eq!(verb.outcome, FireOutcome::Ok); + let last_ok = verb.last_ok.clone(); + + let invalid = dispatch + .fire(request(&reference, "invalid-event")) + .await + .unwrap_err(); + assert_eq!(invalid.code(), tonic::Code::InvalidArgument); + let health = read_health(repo.path()); + let invalid = &health.agents["opencode"].verbs["invalid-event"]; + assert_eq!(invalid.outcome, FireOutcome::Error); + assert!(invalid + .detail + .as_deref() + .unwrap() + .contains("unknown event type")); + + let sessions = repo.path().join(".atomic/sessions"); + let backup = repo.path().join(".atomic/sessions-backup"); + std::fs::rename(&sessions, &backup).unwrap(); + std::fs::write(&sessions, "blocked").unwrap(); + let failure = dispatch + .fire(request(&reference, "session_start")) + .await + .unwrap_err(); + assert_eq!(failure.code(), tonic::Code::Internal); + let health = read_health(repo.path()); + let verb = &health.agents["opencode"].verbs["session_start"]; + assert_eq!(verb.outcome, FireOutcome::Error); + assert_eq!(verb.last_ok, last_ok); + assert!(verb.detail.as_deref().unwrap().contains("orchestrator")); + + std::fs::remove_file(&sessions).unwrap(); + std::fs::rename(&backup, &sessions).unwrap(); + dispatch + .fire(request(&reference, "session_start")) + .await + .unwrap(); + let health = read_health(repo.path()); + let verb = &health.agents["opencode"].verbs["session_start"]; + assert_eq!(verb.outcome, FireOutcome::Ok); + assert!(verb.detail.is_none()); + + // Overlapping callers use the same service gate and retain explicit + // parent lineage on the wire, including events that fail validation. + let before = dispatch.load(&reference, None, 0, 512).await.len(); + let mut callers = Vec::new(); + for index in 0..24 { + let mut caller = dispatch.duplicate(); + let mut event = request(&reference, "concurrent-invalid"); + let body = event.event.as_mut().unwrap(); + body.session_id = format!("child-{}", index % 3); + body.parent_session_id = Some("service-health".into()); + callers.push(tokio::spawn(async move { + caller.fire(event).await.unwrap_err() + })); + } + for caller in callers { + assert_eq!(caller.await.unwrap().code(), tonic::Code::InvalidArgument); + } + assert_eq!( + dispatch.load(&reference, None, 0, 512).await.len(), + before + 24 + ); + for index in 0..3 { + let session = format!("child-{index}"); + let page = dispatch.load(&reference, Some(&session), 0, 5).await; + let rest = dispatch + .load(&reference, Some(&session), page.last().unwrap().sequence, 5) + .await; + assert_eq!(page.len() + rest.len(), 8); + assert!(page.iter().chain(&rest).all(|entry| entry + .event + .as_ref() + .unwrap() + .parent_session_id + .as_deref() + == Some("service-health"))); + } + + let append = libatomic::atomic::AppendHookLogRequest { + repository: Some(reference.clone()), + event: Some(libatomic::atomic::HookLogEvent { + event_id: uuid::Uuid::new_v4().to_string(), + session_id: None, + parent_session_id: None, + agent: "opencode".into(), + display_name: "OpenCode".into(), + verb: "receive".into(), + outcome: "error".into(), + detail: Some("invalid JSON".into()), + at: chrono::Utc::now().to_rfc3339(), + }), + ..Default::default() + }; + let sequence = dispatch.append(append.clone()).await; + assert_eq!(dispatch.append(append).await, sequence); + assert_eq!( + dispatch.load(&reference, None, 0, 512).await.len(), + before + 25 + ); + assert!(!repo.path().join(".atomic/hook-health.json").exists()); + for task in tasks { + task.abort(); + } + } +}