io-harness 0.13.0

//! Watching a run while it happens.
//!
//! Until 0.12.0 the crate had no live surface at all: no observer, no callback,
//! no channel. Everything a run did was durable in the rusqlite trace and
//! readable *afterwards*, so an application that wanted to show progress had to
//! open the SQLite file with a second connection and poll it — against a schema
//! the crate never promised, having first configured the file itself because
//! [`Store::open`](crate::Store::open) set no pragmas. For a run designed to
//! last 24 hours that is not a detail; it is the difference between an agent an
//! operator can watch and one they cannot.
//!
//! Register an [`Observer`] and the run calls it as things happen. The events
//! report the same facts the trace records — that is asserted, not assumed —
//! so the two surfaces cannot drift into disagreeing.
//!
//! ```no_run
//! use io_harness::observe::{EventKind, Flow, Observer, RunEvent};
//!
//! struct Printer;
//!
//! impl Observer for Printer {
//!     fn event(&self, event: &RunEvent) -> Flow {
//!         if let EventKind::Step { decision, tokens, .. } = &event.kind {
//!             println!("step {} — {decision} ({tokens} tokens)", event.step);
//!         }
//!         Flow::Continue
//!     }
//! }
//! ```
//!
//! # Which surface is authoritative
//!
//! The trace is. An event is a notification that something happened, not the
//! record of it: the durable row is what a resume, an audit and an evaluation
//! all read. If an event and the trace ever disagree, the trace is right and the
//! event is a bug.
//!
//! # Ordering and timing
//!
//! Events arrive in the order the run produces them, synchronously, on the task
//! driving the run. [`Observer::event`] is therefore on the run's critical path:
//! a slow observer slows the run down. Do the minimum — push to a queue, send on
//! a channel — and do the work elsewhere.
//!
//! One event does not mean one committed row. A step is committed inside a
//! transaction and its [`EventKind::Step`] is emitted after that transaction
//! succeeds, but a retry emits an [`EventKind::Retry`] having written a row of
//! its own under the *same* step number, and a sub-agent step that pauses
//! because one of its children deferred is deliberately left uncommitted so a
//! resume replays it. Count events if you want to show activity; read the store
//! if you need to know what is durable.
//!
//! # Serialisation
//!
//! Every event serialises, because the process driving a run is often not the
//! process showing it to a person. A host can forward an event as JSON to a user
//! interface written in another language without hand-writing a mapping. The
//! wire shape is flat and tagged:
//!
//! ```json
//! {"run_id": 1, "step": 3, "depth": 0, "event": "step",
//!  "decision": "wrote src/a.rs", "tool_call": "write_file:{…}",
//!  "tokens": 412, "changed": true}
//! ```
//!
//! # Failure
//!
//! [`Observer::event`] returns no `Result`. An observer is a spectator, and a
//! run must not fail because something watching it did. If your observer can
//! fail, absorb the failure and report it out of band.
//!
//! A *panic* is different: `event` is called on the run's own task, so a
//! panicking observer takes the run's future with it and leaves the run row
//! `running`. Do not panic in an observer.

use serde::{Deserialize, Serialize};

/// Whether the run should keep going.
///
/// Returned from every [`Observer::event`] call. This is the only way to stop a
/// run from outside it: before 0.12.0 a caller's only option was to drop the
/// run's future, which abandoned it mid-step and left `runs.status` as
/// `running` forever, so nothing could tell it from a process that had crashed.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Flow {
    /// Keep running. What a watcher returns.
    #[default]
    Continue,
    /// Stop the run.
    ///
    /// Honoured at the next step boundary rather than immediately, because the
    /// points in between are not safe to stop at: a tool call is mid-flight, a
    /// file may be half-written, a child may be running. The run finishes the
    /// step it is on, records `cancelled`, and returns
    /// [`RunOutcome::Cancelled`](crate::RunOutcome::Cancelled). It stays
    /// resumable — cancelling is not abandoning.
    Cancel,
}

impl Flow {
    /// Whether this flow asks the run to stop.
    pub fn is_cancel(self) -> bool {
        self == Flow::Cancel
    }
}

/// One thing that happened during a run.
///
/// The common fields are here rather than repeated on every variant, so a
/// consumer can route on `run_id`/`depth` without matching the payload first.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct RunEvent {
    /// The run this belongs to. In a tree, the *agent's* own run id, not the
    /// root's — a child has its own.
    pub run_id: i64,
    /// The step it happened on. `0` for anything that happens before the first
    /// step, such as authorizing network access to the provider.
    pub step: u32,
    /// Depth in the agent tree: `0` for a single run or a tree's root.
    pub depth: u32,
    /// What happened.
    #[serde(flatten)]
    pub kind: EventKind,
}

impl RunEvent {
    /// An event at depth 0, the common case.
    pub fn new(run_id: i64, step: u32, kind: EventKind) -> Self {
        Self {
            run_id,
            step,
            depth: 0,
            kind,
        }
    }

    /// An event from a specific depth in an agent tree.
    pub fn at_depth(run_id: i64, step: u32, depth: u32, kind: EventKind) -> Self {
        Self {
            run_id,
            step,
            depth,
            kind,
        }
    }
}

/// What happened, and the detail that goes with it.
///
/// One enum with one `Observer` method, rather than a method per kind: adding a
/// kind is then a new variant a consumer can ignore with a `_` arm, instead of a
/// new trait method every implementer inherits. It is also what makes a single
/// serialised wire shape possible.
///
/// Every variant reports something the rusqlite trace already records. This
/// release added no new facts about a run — it added a way to see them while the
/// run is still going.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "event", rename_all = "snake_case")]
pub enum EventKind {
    /// The run began. Emitted once, before the first step.
    Started {
        /// What the run was asked to do.
        goal: String,
        /// The provider that will serve it, by name.
        provider: String,
    },
    /// A step completed and was committed to the store.
    ///
    /// Emitted from one place for all three loops (single-file, workspace and
    /// sub-agent), which before 0.12.0 each had their own copy of the boundary
    /// and their own differently-named log line.
    Step {
        /// What the agent decided, as recorded in `steps.decision`.
        decision: String,
        /// The tool calls it made, as recorded in `steps.tool_call`.
        tool_call: String,
        /// Tokens this step spent, or `0` if the provider reported no usage.
        tokens: u64,
        /// Whether this step changed anything in the workspace. The signal stall
        /// detection rests on, added in 0.11.0.
        changed: bool,
    },
    /// A tool was invoked, before its result is known.
    ToolCall {
        /// The tool's name.
        name: String,
        /// What it was pointed at — a path, a host, a server.
        target: String,
    },
    /// The policy refused an action. It did not happen.
    Refused {
        /// `"read"`, `"write"`, `"exec"` or `"net"`.
        act: String,
        /// What was refused.
        target: String,
        /// The rule that refused it, when a rule did.
        rule: Option<String>,
        /// The policy layer that rule came from.
        layer: Option<String>,
    },
    /// A sensitive action stopped to ask a human. The run is waiting.
    ApprovalRequested {
        /// The act being asked about.
        act: String,
        /// What it would affect.
        target: String,
    },
    /// A human answered.
    ApprovalDecided {
        /// The act that was asked about.
        act: String,
        /// What it would affect.
        target: String,
        /// `"approve"`, `"deny"` or `"defer"`.
        decision: String,
    },
    /// A step's tokens were drawn against a tree's shared ceiling.
    SpendDraw {
        /// Tokens drawn.
        tokens: u64,
        /// What the tree has left afterwards.
        remaining: Option<u64>,
    },
    /// A provider call failed in a way worth trying again, and will be.
    Retry {
        /// The failure's kind, as classified in 0.11.0.
        kind: String,
        /// Which attempt this is.
        attempt: u32,
        /// How long the run will wait first.
        delay_ms: u64,
    },
    /// A `Fallback` provider fell over, and this is who answered instead.
    FellBackTo {
        /// The provider that served the step.
        provider: String,
    },
    /// The agent has changed nothing for a while and has been told once to try
    /// something else. The run continues.
    Replan {
        /// How many steps without progress triggered it.
        window: u32,
    },
    /// The agent had already been told and is still going in circles. Terminal.
    Stalled,
    /// A sub-agent was started.
    Spawned {
        /// The child's own run id.
        child_run_id: i64,
        /// What the child was asked to do.
        goal: String,
    },
    /// A spawn was refused by containment rather than performed.
    SpawnRefused {
        /// Which cap refused it — agents, depth or concurrency.
        cap: String,
    },
    /// The agent wrote something to durable cross-run memory.
    MemoryWrote {
        /// The note's key.
        key: String,
    },
    /// A sandbox was created, ran something, hit a cap, or was destroyed.
    Sandbox {
        /// `"create"`, `"exec"`, `"cap_hit"`, `"destroy"` or
        /// `"gate_phase_failed"`.
        kind: String,
        /// The backend that isolated it, when one is known.
        backend: Option<String>,
    },
    /// An MCP server was reached, or one of its tools was called.
    Mcp {
        /// The server, by the name it was configured under.
        server: String,
        /// The tool called, for a call.
        tool: Option<String>,
        /// Whether the call succeeded.
        ok: Option<bool>,
        /// How long it took.
        millis: Option<u64>,
    },
    /// The run ended. Emitted once, last.
    Finished {
        /// The outcome string as written to `runs.outcome`.
        outcome: String,
        /// Steps completed.
        steps: u32,
        /// Tokens spent across the run.
        tokens: u64,
    },
}

/// Watches a run as it happens.
///
/// Shaped after [`Approver`](crate::Approver), the crate's other
/// inversion-of-control point: `Send + Sync` with `&self` methods, held as
/// `&dyn Observer`. `&self` rather than `&mut self` is not a style choice — a
/// tree runs up to `max_concurrent` children as concurrent futures on one task,
/// and a `&mut self` observer could not be shared between them. Keep whatever
/// state you need behind a `Mutex`, an atomic, or a channel.
pub trait Observer: Send + Sync {
    /// Called once per event, in order, on the run's own task.
    ///
    /// Return [`Flow::Cancel`] to stop the run at its next step boundary.
    /// Watchers return [`Flow::Continue`].
    ///
    /// This returns no `Result` on purpose: see the [module docs](self#failure).
    fn event(&self, event: &RunEvent) -> Flow;
}

/// Watches nothing. The default when a caller registers no observer.
///
/// Exists so the run has one code path rather than `Option<&dyn Observer>`
/// threaded through every call site, and so "no observer" costs a call to an
/// empty function that optimises away rather than a branch per event.
#[derive(Debug, Clone, Copy, Default)]
pub struct Ignore;

impl Observer for Ignore {
    fn event(&self, _event: &RunEvent) -> Flow {
        Flow::Continue
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn the_wire_shape_is_flat_and_tagged() {
        let e = RunEvent::new(
            7,
            3,
            EventKind::Step {
                decision: "wrote src/a.rs".into(),
                tool_call: "write_file:{}".into(),
                tokens: 412,
                changed: true,
            },
        );
        let v: serde_json::Value = serde_json::to_value(&e).unwrap();
        // Flat: a consumer in another language reads one object, not a nested
        // one. Pinned because this is a wire contract, not an internal detail.
        assert_eq!(v["run_id"], 7);
        assert_eq!(v["step"], 3);
        assert_eq!(v["depth"], 0);
        assert_eq!(v["event"], "step");
        assert_eq!(v["decision"], "wrote src/a.rs");
        assert_eq!(v["tokens"], 412);
        assert_eq!(v["changed"], true);
        assert!(v.get("kind").is_none(), "the payload must not be nested");
    }

    #[test]
    fn a_unit_variant_still_carries_its_tag() {
        let v = serde_json::to_value(RunEvent::new(1, 9, EventKind::Stalled)).unwrap();
        assert_eq!(v["event"], "stalled");
        assert_eq!(v["step"], 9);
    }

    #[test]
    fn every_variant_round_trips() {
        for kind in every_kind() {
            let e = RunEvent::at_depth(1, 2, 3, kind);
            let json = serde_json::to_string(&e).unwrap();
            let back: RunEvent = serde_json::from_str(&json).unwrap_or_else(|err| {
                panic!("{json} did not round-trip: {err}");
            });
            assert_eq!(e, back, "round-trip changed the event: {json}");
        }
    }

    #[test]
    fn each_variant_has_a_distinct_tag() {
        let mut tags = Vec::new();
        for kind in every_kind() {
            let v = serde_json::to_value(RunEvent::new(1, 1, kind)).unwrap();
            tags.push(v["event"].as_str().unwrap().to_string());
        }
        let mut unique = tags.clone();
        unique.sort();
        unique.dedup();
        assert_eq!(
            tags.len(),
            unique.len(),
            "two variants share a wire tag, so a consumer cannot tell them apart: {tags:?}"
        );
    }

    #[test]
    fn ignore_never_cancels() {
        assert_eq!(
            Ignore.event(&RunEvent::new(1, 1, EventKind::Stalled)),
            Flow::Continue
        );
        assert!(!Flow::Continue.is_cancel());
        assert!(Flow::Cancel.is_cancel());
        assert_eq!(Flow::default(), Flow::Continue);
    }

    /// One of every variant. Kept exhaustive by the `match` below: adding a
    /// variant without adding it here fails to compile, so the round-trip and
    /// distinct-tag tests cannot silently stop covering the enum.
    fn every_kind() -> Vec<EventKind> {
        let all = vec![
            EventKind::Started {
                goal: "g".into(),
                provider: "p".into(),
            },
            EventKind::Step {
                decision: "d".into(),
                tool_call: "t".into(),
                tokens: 1,
                changed: true,
            },
            EventKind::ToolCall {
                name: "n".into(),
                target: "t".into(),
            },
            EventKind::Refused {
                act: "read".into(),
                target: "t".into(),
                rule: Some("r".into()),
                layer: None,
            },
            EventKind::ApprovalRequested {
                act: "write".into(),
                target: "t".into(),
            },
            EventKind::ApprovalDecided {
                act: "write".into(),
                target: "t".into(),
                decision: "approve".into(),
            },
            EventKind::SpendDraw {
                tokens: 1,
                remaining: Some(2),
            },
            EventKind::Retry {
                kind: "rate_limited".into(),
                attempt: 1,
                delay_ms: 40,
            },
            EventKind::FellBackTo {
                provider: "p".into(),
            },
            EventKind::Replan { window: 3 },
            EventKind::Stalled,
            EventKind::Spawned {
                child_run_id: 2,
                goal: "g".into(),
            },
            EventKind::SpawnRefused {
                cap: "depth".into(),
            },
            EventKind::MemoryWrote { key: "k".into() },
            EventKind::Sandbox {
                kind: "create".into(),
                backend: Some("b".into()),
            },
            EventKind::Mcp {
                server: "s".into(),
                tool: Some("t".into()),
                ok: Some(true),
                millis: Some(5),
            },
            EventKind::Finished {
                outcome: "success".into(),
                steps: 4,
                tokens: 9,
            },
        ];
        // Exhaustiveness guard. Never executed for its result; it exists so the
        // compiler refuses a new variant that `all` does not mention.
        for k in &all {
            match k {
                EventKind::Started { .. }
                | EventKind::Step { .. }
                | EventKind::ToolCall { .. }
                | EventKind::Refused { .. }
                | EventKind::ApprovalRequested { .. }
                | EventKind::ApprovalDecided { .. }
                | EventKind::SpendDraw { .. }
                | EventKind::Retry { .. }
                | EventKind::FellBackTo { .. }
                | EventKind::Replan { .. }
                | EventKind::Stalled
                | EventKind::Spawned { .. }
                | EventKind::SpawnRefused { .. }
                | EventKind::MemoryWrote { .. }
                | EventKind::Sandbox { .. }
                | EventKind::Mcp { .. }
                | EventKind::Finished { .. } => {}
            }
        }
        all
    }
}