Skip to main content

io_harness/
state.rs

1//! Run state in rusqlite: the full trace of a run — prompts, decisions, tool
2//! calls, token usage, and outcome — readable back afterwards for audit, and
3//! enough to resume an interrupted run under the same run id.
4//!
5//! The 0.2.0 schema adds `prompt`, `tool_call`, and `tokens` columns to `steps`.
6//! An existing 0.1.0 database is migrated in place with `ALTER TABLE ADD COLUMN`
7//! (additive — a 0.1.0 binary still reads a migrated database).
8
9use rusqlite::Connection;
10
11use crate::context::{ObsKind, Observation};
12use crate::error::{Error, Result};
13use crate::policy::Policy;
14
15/// An observation kind as it is stored: the serde rendering, not
16/// [`ObsKind::label`], which is English for a prompt reader and renders `Write`
17/// as "wrote". Going through serde rather than a hand-written match is what
18/// keeps the mapping total — a new variant cannot be added without a rendering,
19/// and the pair below cannot drift apart.
20fn kind_wire(kind: ObsKind) -> String {
21    match serde_json::to_value(kind) {
22        Ok(serde_json::Value::String(s)) => s,
23        // Unreachable for a unit-variant enum with `rename_all`; falling back to
24        // the debug form keeps the write infallible without inventing a kind that
25        // would read back as a different observation.
26        other => format!("{other:?}"),
27    }
28}
29
30/// The inverse of [`kind_wire`]. A kind that does not parse is an error and not
31/// a skipped row: a ledger that came back silently shorter than it was would
32/// assemble a context nobody can account for, which is worse than refusing to
33/// restore it at all.
34fn kind_from_wire(kind: &str, run_id: i64) -> Result<ObsKind> {
35    serde_json::from_value(serde_json::Value::String(kind.to_string())).map_err(|e| Error::Resume {
36        reason: format!("run {run_id} has a ledger observation of unknown kind {kind:?}: {e}"),
37    })
38}
39
40/// The checkpoint layout version stamped into `PRAGMA user_version`. Bump when
41/// the on-disk checkpoint format changes incompatibly. A store whose version is
42/// higher than this is from a newer binary and is refused on resume.
43///
44/// The reason a caller ever reads it: a resume against a store a *newer*
45/// io-harness wrote fails, typed, before anything is replayed — and this constant
46/// is the version to name when reporting that.
47///
48/// ```
49/// use io_harness::{Error, Store, CHECKPOINT_FORMAT};
50///
51/// # fn main() -> io_harness::Result<()> {
52/// let path = std::env::temp_dir().join("io-harness-doc-newer-checkpoint.sqlite3");
53/// let _ = std::fs::remove_file(&path);
54/// {
55///     // Stand in for a store written by a future release.
56///     let conn = rusqlite::Connection::open(&path).unwrap();
57///     conn.pragma_update(None, "user_version", CHECKPOINT_FORMAT + 1).unwrap();
58/// }
59///
60/// let store = Store::open(&path)?;
61/// match store.check_resumable(1) {
62///     Err(Error::Resume { reason }) => {
63///         // Not a panic, and not a half-resume that reads a layout this binary
64///         // does not understand. Tell the operator what to do about it.
65///         assert!(reason.contains("newer"), "{reason}");
66///         eprintln!("this binary understands checkpoint format {CHECKPOINT_FORMAT}: {reason}");
67///     }
68///     other => panic!("a newer store must refuse to resume, got {other:?}"),
69/// }
70/// # let _ = std::fs::remove_file(&path);
71/// # Ok(())
72/// # }
73/// ```
74pub const CHECKPOINT_FORMAT: i64 = 7;
75
76/// The one outcome string that means the run did what it was asked.
77///
78/// Named rather than inlined so that [`RunSummary::success`] and any future
79/// reader agree by construction. Eleven outcome strings exist; exactly one of
80/// them is the task being done.
81///
82/// ```
83/// use io_harness::{Store, SUCCESS_OUTCOME};
84///
85/// # fn main() -> io_harness::Result<()> {
86/// let store = Store::memory()?;
87/// let worked = store.start_run("add a hello function", "src/hello.rs")?;
88/// let gave_up = store.start_run("add a goodbye function", "src/bye.rs")?;
89/// store.finish_run(worked, SUCCESS_OUTCOME)?;
90/// store.finish_run(gave_up, "step_cap_reached")?;
91///
92/// // Scoring a batch of runs: compare against this rather than against the
93/// // literal `"success"`, so a reader and the crate cannot drift apart about
94/// // which of the eleven endings counts.
95/// let succeeded = |id| -> io_harness::Result<bool> {
96///     Ok(store.outcome(id)?.as_deref() == Some(SUCCESS_OUTCOME))
97/// };
98/// assert!(succeeded(worked)?);
99/// assert!(!succeeded(gave_up)?);
100///
101/// // Which is exactly what `RunSummary::success` already carries, computed the
102/// // same way at the moment the run ended.
103/// assert_eq!(store.run_summary(worked)?.map(|s| s.success), Some(true));
104/// # Ok(())
105/// # }
106/// ```
107pub const SUCCESS_OUTCOME: &str = "success";
108
109/// The durable lifecycle status of a run, so a caller can tell a crashed run
110/// (still `Running`) from one paused for a human (`Paused`) or finished
111/// (`Completed`). OS- and rusqlite-free, so it is safe in the public API.
112///
113/// ```
114/// use io_harness::{RunStatus, Store};
115///
116/// # fn main() -> io_harness::Result<()> {
117/// let store = Store::memory()?;
118/// let crashed = store.start_run("summarise the README", "NOTES.md")?;
119/// let waiting = store.start_run("rewrite the changelog", "CHANGELOG.md")?;
120/// let done = store.start_run("add a hello function", "src/hello.rs")?;
121/// store.finish_run(waiting, "awaiting_approval")?;
122/// store.finish_run(done, "success")?;
123///
124/// // The triage a supervisor does on startup. A run left `Running` by a store
125/// // nobody is driving is one whose process died mid-loop: resume it. `Paused` is
126/// // waiting on a human and resumes with their decision, not without it.
127/// let resumable: Vec<i64> = store
128///     .runs()?
129///     .into_iter()
130///     .filter(|id| store.run_status(*id).ok().flatten() == Some(RunStatus::Running))
131///     .collect();
132/// assert_eq!(resumable, [crashed]);
133/// assert_eq!(store.run_status(waiting)?, Some(RunStatus::Paused));
134/// assert_eq!(store.run_status(done)?, Some(RunStatus::Completed));
135/// # Ok(())
136/// # }
137/// ```
138#[derive(Debug, Clone, Copy, PartialEq, Eq)]
139pub enum RunStatus {
140    /// The run is in progress — or was, until the process died mid-loop. A
141    /// `Running` run found in a store is the resume target.
142    Running,
143    /// The run paused for a human decision and can be resumed once it arrives.
144    Paused,
145    /// The run finished (with success or a terminal budget/deny outcome).
146    Completed,
147    /// The run ended in an error.
148    Failed,
149}
150
151impl RunStatus {
152    fn from_str(s: &str) -> Self {
153        match s {
154            "paused" => RunStatus::Paused,
155            "completed" => RunStatus::Completed,
156            "failed" => RunStatus::Failed,
157            _ => RunStatus::Running,
158        }
159    }
160}
161
162/// A persisted spawned-child contract, enough to rebuild and resume that exact
163/// child on a tree resume rather than spawning a duplicate.
164///
165/// ```
166/// use io_harness::Store;
167///
168/// # fn main() -> io_harness::Result<()> {
169/// let store = Store::memory()?;
170/// let parent = store.start_run("summarise the repo", "NOTES.md")?;
171/// let child = store.start_child_run("summarise src/", "NOTES.md", parent, 1)?;
172/// store.record_spawn(parent, 4, child, "summarise src/", "NOTES.md", "#", Some(8), "[]")?;
173///
174/// // What a tree resume does with it: the parent replays step 4, looks the spawn
175/// // up by (parent, step, goal), and adopts the child it already made. Without
176/// // this row the replay would spawn a second child and spend the tree's ledger
177/// // twice for one piece of work.
178/// let row = store.find_spawn(parent, 4, "summarise src/")?.expect("recorded above");
179/// assert_eq!(row.child_run_id, child);
180/// assert_eq!(row.max_steps, Some(8));
181/// // The narrowing the parent applied is stored too, so the adopted child resumes
182/// // under the policy it was contained by rather than the parent's wider one.
183/// assert_eq!(row.deny_write, "[]");
184///
185/// // A step that never spawned has no row, which is how a replay tells "already
186/// // done" from "not done yet".
187/// assert!(store.find_spawn(parent, 5, "summarise src/")?.is_none());
188/// # Ok(())
189/// # }
190/// ```
191#[derive(Debug, Clone, PartialEq)]
192pub struct SpawnRow {
193    /// The child run id already allocated for this spawn.
194    pub child_run_id: i64,
195    /// The child's goal.
196    pub goal: String,
197    /// The workspace-relative file the child's verification reads.
198    pub verify_file: String,
199    /// The substring the child's verification requires.
200    pub needle: String,
201    /// The child's step cap, if the parent set one.
202    pub max_steps: Option<u32>,
203    /// JSON array of `deny_write` globs the parent narrowed the child with.
204    pub deny_write: String,
205}
206
207/// What one finished run cost and whether it worked.
208///
209/// Written once by [`Store::finish_run`] and read back with
210/// [`Store::run_summary`]. Before 0.12.0 a consumer had to assemble this itself
211/// from three different queries plus knowledge of which of eleven outcome
212/// strings count as success — and could not get `duration_ms` at all, because
213/// nothing recorded when a run ended.
214///
215/// Serialisable, so a scoring tool can store or ship it without restating the
216/// shape.
217///
218/// ```
219/// use io_harness::{Store, StepRecord};
220///
221/// # fn main() -> io_harness::Result<()> {
222/// let store = Store::memory()?;
223/// let run_id = store.start_run("add a hello function", "src/hello.rs")?;
224/// store.record(run_id, &StepRecord::new(1, "wrote src/hello.rs", "ok").with_trace("", "", 1_280))?;
225/// store.finish_run(run_id, "step_cap_reached")?;
226///
227/// // One row instead of three queries plus knowledge of which of eleven outcome
228/// // strings mean success. This is what a scoring tool reads per run.
229/// let summary = store.run_summary(run_id)?.expect("the run has finished");
230/// assert!(!summary.success);
231/// // `outcome` and `success` are both kept on purpose: the flag says whether it
232/// // worked, the string says *how* it ended, and a step cap, a stall and a
233/// // human's refusal are three different things to act on.
234/// assert_eq!(summary.outcome, "step_cap_reached");
235/// assert_eq!((summary.steps, summary.tokens), (1, 1_280));
236///
237/// // `None` while a run is unfinished or paused for a human — absent rather than
238/// // a row of zeroes, which would read like a run that did nothing.
239/// let paused = store.start_run("rewrite the changelog", "CHANGELOG.md")?;
240/// store.finish_run(paused, "awaiting_approval")?;
241/// assert!(store.run_summary(paused)?.is_none());
242/// # Ok(())
243/// # }
244/// ```
245#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
246pub struct RunSummary {
247    /// The run this describes.
248    pub run_id: i64,
249    /// The raw outcome string, as written to `runs.outcome`.
250    ///
251    /// Kept alongside [`Self::success`] rather than replaced by it: the string
252    /// says *which* ending, the flag says whether it was the good one, and
253    /// collapsing them would throw away the distinction between a step cap, a
254    /// stall and a human's refusal.
255    pub outcome: String,
256    /// Whether the run achieved what it was asked to do.
257    ///
258    /// True for exactly one outcome — `success`. Every other ending, including
259    /// the ones that are nobody's fault like a rate-limited provider, is not the
260    /// task being done.
261    pub success: bool,
262    /// Steps completed, as `MAX(step)`.
263    ///
264    /// Not `COUNT(*)` over `steps`: a retry writes its own row under the same
265    /// step number, so counting rows counts trace entries rather than agent
266    /// steps. For a tree this is the ROOT agent's step count, not the tree's
267    /// total — each agent has its own summary.
268    pub steps: u32,
269    /// Tokens spent by this run, summed from its committed steps.
270    ///
271    /// Tokens, not money. A provider reports usage and never a price, so the
272    /// crate has nothing to convert with; see
273    /// [`Containment::max_total_cost`](crate::Containment::max_total_cost).
274    pub tokens: u64,
275    /// Wall-clock milliseconds from the run's start to its end.
276    ///
277    /// `None` for a run started before 0.7.0, which has no `started_at` to
278    /// measure from. Includes time the process was not running — a run that
279    /// crashed at midnight and resumed at nine counts the nine hours, because
280    /// that is how long the run took even though it was not working.
281    pub duration_ms: Option<u64>,
282    /// When the run ended, from the database clock.
283    pub finished_at: String,
284}
285
286/// A persisted run store. Use [`Store::open`] for a file, or [`Store::memory`]
287/// for an ephemeral in-memory database.
288///
289/// The store is not a log. It is what makes a run resumable after a crash and
290/// auditable afterwards, so which constructor you choose is a decision about
291/// whether either matters for this run.
292///
293/// ```no_run
294/// use io_harness::{run, OpenRouter, Store, TaskContract, Verification};
295///
296/// # async fn demo() -> io_harness::Result<()> {
297/// // A file store outlives the process: a run that dies mid-loop is resumable
298/// // from its last committed step, and a second process may read the trace while
299/// // this one is still writing it (WAL, plus `BUSY_TIMEOUT`).
300/// let store = Store::open("runs.sqlite3")?;
301///
302/// let contract = TaskContract::new(
303///     "add a hello function returning 42",
304///     "src/hello.rs",
305///     Verification::FileContains("fn hello".into()),
306/// );
307/// let result = run(&contract, &OpenRouter::from_env()?, &store).await?;
308///
309/// // Everything the run did, read back by id: the trace, the budget draws, the
310/// // policy refusals, and what it cost.
311/// for step in store.steps(result.run_id)? {
312///     println!("{}: {} ({} tokens)", step.step, step.decision, step.tokens);
313/// }
314/// println!("{:?}", store.run_summary(result.run_id)?);
315/// # Ok(())
316/// # }
317/// ```
318///
319/// [`Store::memory`] is the same API with no file: a throwaway run, or a test.
320/// Nothing survives the process, so nothing is resumable — which is the right
321/// trade only when a failed run is cheaper to restart than to continue.
322pub struct Store {
323    conn: Connection,
324}
325
326/// One durable checkpoint-lifecycle event: a step was checkpointed, a run was
327/// resumed, or an already-committed step was skipped on resume. Together they
328/// make a crashed-and-resumed run's history reconstructable from the store.
329///
330/// ```
331/// use io_harness::{CheckpointEvent, Store};
332///
333/// # fn main() -> io_harness::Result<()> {
334/// # let store = Store::memory()?;
335/// # let run_id = store.start_run("summarise the repo", "NOTES.md")?;
336/// # store.record_checkpoint_event(&CheckpointEvent::checkpoint(run_id, 1))?;
337/// # store.record_checkpoint_event(&CheckpointEvent::checkpoint(run_id, 2))?;
338/// # store.record_checkpoint_event(&CheckpointEvent::resume(run_id, 3, "restarted after a crash"))?;
339/// # store.record_checkpoint_event(&CheckpointEvent::skipped(run_id, 1))?;
340/// # store.record_checkpoint_event(&CheckpointEvent::skipped(run_id, 2))?;
341/// // Answers the question a crashed run leaves behind: did it restart, and did
342/// // the restart re-do work that was already committed?
343/// let events = store.checkpoint_events(run_id)?;
344/// let resumes = events.iter().filter(|e| e.kind == "resume").count();
345/// let replayed: Vec<u32> = events.iter().filter(|e| e.kind == "skipped").map(|e| e.step).collect();
346///
347/// assert_eq!(resumes, 1, "this run died once and came back");
348/// // Two steps were replayed and recognised as already done, so they cost
349/// // nothing the second time — that is what makes a resume idempotent rather
350/// // than a second charge for the same steps.
351/// assert_eq!(replayed, [1, 2]);
352/// # Ok(())
353/// # }
354/// ```
355#[derive(Debug, Clone, PartialEq)]
356pub struct CheckpointEvent {
357    /// The run this event belongs to.
358    pub run_id: i64,
359    /// The step it concerns.
360    pub step: u32,
361    /// `"checkpoint"`, `"resume"`, or `"skipped"`.
362    pub kind: String,
363    /// Optional human-readable detail (never file contents or secrets).
364    pub detail: Option<String>,
365}
366
367impl CheckpointEvent {
368    /// A step was durably checkpointed.
369    pub fn checkpoint(run_id: i64, step: u32) -> Self {
370        Self {
371            run_id,
372            step,
373            kind: "checkpoint".into(),
374            detail: None,
375        }
376    }
377    /// A run was resumed, re-driving from `step`.
378    pub fn resume(run_id: i64, step: u32, detail: impl Into<String>) -> Self {
379        Self {
380            run_id,
381            step,
382            kind: "resume".into(),
383            detail: Some(detail.into()),
384        }
385    }
386    /// An already-committed step was skipped on resume.
387    pub fn skipped(run_id: i64, step: u32) -> Self {
388        Self {
389            run_id,
390            step,
391            kind: "skipped".into(),
392            detail: None,
393        }
394    }
395}
396
397/// One recorded loop step — the full trace entry, as written and read back.
398///
399/// ```
400/// use io_harness::{StepRecord, Store};
401///
402/// # fn main() -> io_harness::Result<()> {
403/// let store = Store::memory()?;
404/// let run_id = store.start_run("add a hello function", "src/hello.rs")?;
405///
406/// // A transient provider failure, and then the step that worked. Both are step
407/// // 1: a retry writes its own row under the step number it retried.
408/// store.record(run_id, &StepRecord::new(1, "retry 1 after a 503", ""))?;
409///
410/// // `new` alone records a decision and its result; `with_trace` adds the audit
411/// // half — the exact prompt sent, the call the model made, and what it cost.
412/// // Without it the trace says what happened and not why.
413/// store.record(
414///     run_id,
415///     &StepRecord::new(1, "wrote src/hello.rs", "ok").with_trace(
416///         "<the assembled prompt>",
417///         r#"{"name":"write_file","arguments":{"path":"src/hello.rs"}}"#,
418///         1_280,
419///     ),
420/// )?;
421///
422/// // Read back for audit — and the reason `RunSummary::steps` is `MAX(step)`
423/// // rather than a row count: two rows here, one agent step.
424/// let steps = store.steps(run_id)?;
425/// assert_eq!(steps.len(), 2);
426/// assert_eq!(store.last_step(run_id)?, 1);
427/// assert!(steps[1].tool_call.contains("write_file"));
428/// assert_eq!(steps.iter().map(|s| s.tokens).sum::<u64>(), 1_280);
429/// # Ok(())
430/// # }
431/// ```
432#[derive(Debug, Clone, PartialEq)]
433pub struct StepRecord {
434    /// 1-based step number within the run.
435    pub step: u32,
436    /// What the agent decided this step (e.g. "wrote file", "retry 1 after error").
437    pub decision: String,
438    /// Intermediate result / model text for the step.
439    pub result: String,
440    /// The prompt sent to the model this step.
441    pub prompt: String,
442    /// The tool call the model made, as JSON, or "" if none.
443    pub tool_call: String,
444    /// Total tokens used this step, 0 if the provider reported none.
445    pub tokens: u64,
446}
447
448impl StepRecord {
449    /// A trace entry with the audit fields empty — for callers that only record
450    /// a decision and result.
451    pub fn new(step: u32, decision: impl Into<String>, result: impl Into<String>) -> Self {
452        Self {
453            step,
454            decision: decision.into(),
455            result: result.into(),
456            prompt: String::new(),
457            tool_call: String::new(),
458            tokens: 0,
459        }
460    }
461
462    /// Attach the prompt, tool call, and token count for the full trace.
463    pub fn with_trace(
464        mut self,
465        prompt: impl Into<String>,
466        tool_call: impl Into<String>,
467        tokens: u64,
468    ) -> Self {
469        self.prompt = prompt.into();
470        self.tool_call = tool_call.into();
471        self.tokens = tokens;
472        self
473    }
474}
475
476/// One policy event in the trace: an action refused, or a human decision.
477///
478/// Records the path, command, rule, layer, and decision — never file contents
479/// or credentials. (The write payload of a *deferred* action is held separately
480/// in the pending-approval row, because resuming it requires replaying exactly
481/// what was approved.)
482///
483/// ```
484/// use io_harness::{PolicyEvent, Store};
485///
486/// # fn main() -> io_harness::Result<()> {
487/// # let store = Store::memory()?;
488/// # let run_id = store.start_run("tidy the repo", "src/lib.rs")?;
489/// # store.record_event(run_id, &PolicyEvent::refusal(3, "write", "secrets/id_rsa")
490/// #     .with_rule("secrets/*", "app"))?;
491/// # store.record_event(run_id, &PolicyEvent::decision(4, "exec", "git push", "approve", "stdin")
492/// #     .with_performed("git push --dry-run"))?;
493/// // The audit an operator actually asks for: what did the agent try that it was
494/// // not allowed to do, and which rule stopped it?
495/// let events = store.events(run_id)?;
496/// let refused = events.iter().find(|e| e.kind == "refusal").expect("one refusal");
497/// assert_eq!(refused.target, "secrets/id_rsa");
498/// // Attributed to the rule and the layer, so "why was this denied" is answered
499/// // from the store rather than by re-deriving the policy stack by hand.
500/// assert_eq!((refused.rule.as_deref(), refused.layer.as_deref()), (Some("secrets/*"), Some("app")));
501///
502/// // And the case that is easy to miss: an approval that changed the action. The
503/// // agent asked for one command and a human let a different one through.
504/// let approved = events.iter().find(|e| e.kind == "decision").expect("one decision");
505/// assert_eq!(approved.decision.as_deref(), Some("approve"));
506/// assert_eq!(approved.performed.as_deref(), Some("git push --dry-run"));
507/// # Ok(())
508/// # }
509/// ```
510#[derive(Debug, Clone, PartialEq, Eq)]
511pub struct PolicyEvent {
512    /// 1-based step the event occurred on.
513    pub step: u32,
514    /// `"refusal"` or `"decision"`.
515    pub kind: String,
516    /// `"read"`, `"write"`, or `"exec"`.
517    pub act: String,
518    /// The path, or the binary name plus argv for an exec.
519    pub target: String,
520    /// The glob that decided, when a rule rather than a tier default did.
521    pub rule: Option<String>,
522    /// The layer the deciding rule came from.
523    pub layer: Option<String>,
524    /// `"approve"`, `"deny"`, or `"defer"` for a decision.
525    pub decision: Option<String>,
526    /// Which approver decided, or `"remembered"` when a remembered rule did.
527    pub source: Option<String>,
528    /// The action actually performed, when approve-with-changes altered it.
529    pub performed: Option<String>,
530}
531
532impl PolicyEvent {
533    /// An action refused by the policy.
534    pub fn refusal(step: u32, act: impl Into<String>, target: impl Into<String>) -> Self {
535        Self {
536            step,
537            kind: "refusal".into(),
538            act: act.into(),
539            target: target.into(),
540            rule: None,
541            layer: None,
542            decision: None,
543            source: None,
544            performed: None,
545        }
546    }
547
548    /// A human (or built-in approver) decision on a sensitive action.
549    pub fn decision(
550        step: u32,
551        act: impl Into<String>,
552        target: impl Into<String>,
553        decision: impl Into<String>,
554        source: impl Into<String>,
555    ) -> Self {
556        Self {
557            kind: "decision".into(),
558            decision: Some(decision.into()),
559            source: Some(source.into()),
560            ..Self::refusal(step, act, target)
561        }
562    }
563
564    /// Attribute the event to the rule and layer that produced it.
565    pub fn with_rule(mut self, rule: impl Into<String>, layer: impl Into<String>) -> Self {
566        self.rule = Some(rule.into());
567        self.layer = Some(layer.into());
568        self
569    }
570
571    /// Record that the action performed differed from the one requested.
572    pub fn with_performed(mut self, performed: impl Into<String>) -> Self {
573        self.performed = Some(performed.into());
574        self
575    }
576}
577
578/// An action paused awaiting a human decision, persisted so it outlives the
579/// process that requested it.
580///
581/// ```
582/// use io_harness::Store;
583///
584/// # fn main() -> io_harness::Result<()> {
585/// let store = Store::memory()?;
586/// let run_id = store.start_run("update the deploy config", "deploy/prod.yaml")?;
587///
588/// // The agent asked to write a production file, the policy said `Ask`, and the
589/// // approver deferred. The payload is stored with it, because resuming has to
590/// // replay exactly what was approved and not whatever the model would say now.
591/// let request_id = store.put_pending(run_id, 2, "write", "deploy/prod.yaml", Some("replicas: 4"))?;
592///
593/// // This process may now exit. A reviewer — a web UI, a CLI, a person the next
594/// // morning — reads the request back and decides.
595/// let pending = store.pending(request_id)?.expect("just written");
596/// assert_eq!((pending.act.as_str(), pending.target.as_str()), ("write", "deploy/prod.yaml"));
597/// assert_eq!(pending.content.as_deref(), Some("replicas: 4"));
598/// assert!(pending.resolved.is_none(), "nobody has decided yet");
599///
600/// store.resolve_pending(request_id, "approve")?;
601/// assert_eq!(store.pending(request_id)?.unwrap().resolved.as_deref(), Some("approve"));
602/// // `resume_with_decision` then continues the run from here.
603/// # Ok(())
604/// # }
605/// ```
606#[derive(Debug, Clone, PartialEq, Eq)]
607pub struct Pending {
608    /// The request id, as returned by [`Store::put_pending`].
609    pub id: i64,
610    /// The run this action belongs to.
611    pub run_id: i64,
612    /// The step it paused on.
613    pub step: u32,
614    /// `"read"`, `"write"`, or `"exec"`.
615    pub act: String,
616    /// The target path or binary.
617    pub target: String,
618    /// The write payload, needed to replay exactly what was approved.
619    pub content: Option<String>,
620    /// `None` while pending; otherwise `"approve"` or `"deny"`.
621    pub resolved: Option<String>,
622}
623
624/// One event in a tree of agents: a parent spawning a child, a spawn refused by
625/// the containment boundary, or a draw against the tree's shared spend ceiling.
626///
627/// Together with each run's `parent_run_id` these make the tree a reconstructable
628/// graph — who spawned whom, what was refused, and what the tree spent — long
629/// after the process that ran it has exited.
630///
631/// ```
632/// use io_harness::{AgentEvent, Store};
633///
634/// # fn main() -> io_harness::Result<()> {
635/// # let store = Store::memory()?;
636/// # let run_id = store.start_run("summarise the repo", "NOTES.md")?;
637/// # store.record_agent_event(&AgentEvent::budget_draw(run_id, 1, 1_280, 8_720))?;
638/// # store.record_agent_event(&AgentEvent::spawn(run_id, 2, 2, "summarise src/"))?;
639/// # store.record_agent_event(&AgentEvent::budget_draw(run_id, 2, 4_100, 4_620))?;
640/// # store.record_agent_event(&AgentEvent::spawn_refused(run_id, 3, "agents"))?;
641/// # store.record_agent_event(&AgentEvent::budget_draw(run_id, 3, 3_900, 720))?;
642/// // The only audit of what each step drew against the tree's *shared* ceiling.
643/// // A run's own token total does not show this: the ceiling is tree-wide, so
644/// // what matters is how fast `remaining` is falling for everyone.
645/// let events = store.agent_events(run_id)?;
646///
647/// let drawn: u64 = events.iter().filter_map(|e| e.tokens).sum();
648/// let left = events.iter().filter_map(|e| e.remaining).last();
649/// assert_eq!(drawn, 9_280);
650/// // 720 left after three steps that averaged over 3,000: this tree ends in
651/// // `BudgetCeilingReached` next step, and the row says so before it happens.
652/// assert_eq!(left, Some(720));
653///
654/// // The same table carries the shape of the tree and what it was denied.
655/// let children: Vec<i64> = events.iter().filter_map(|e| e.child_run_id).collect();
656/// let refusals: Vec<&str> = events
657///     .iter()
658///     .filter(|e| e.kind == "spawn_refused")
659///     .filter_map(|e| e.detail.as_deref())
660///     .collect();
661/// assert_eq!(children, [2]);
662/// // A parent that wanted a second child and hit the agent cap — a run doing
663/// // less work than it planned to, which is invisible in its outcome.
664/// assert_eq!(refusals, ["agents"]);
665/// # Ok(())
666/// # }
667/// ```
668#[derive(Debug, Clone, PartialEq, Eq)]
669pub struct AgentEvent {
670    /// The agent this event belongs to (the parent, for a spawn; the drawing
671    /// agent, for a budget draw).
672    pub run_id: i64,
673    /// The step it occurred on.
674    pub step: u32,
675    /// `"spawn"`, `"spawn_refused"`, or `"budget_draw"`.
676    pub kind: String,
677    /// The spawned child's run id, for a `"spawn"`.
678    pub child_run_id: Option<i64>,
679    /// Free-form detail: the child's goal for a spawn, the breached cap for a
680    /// refusal.
681    pub detail: Option<String>,
682    /// Tokens drawn, for a `"budget_draw"`.
683    pub tokens: Option<u64>,
684    /// The tree's remaining tokens after the draw.
685    pub remaining: Option<u64>,
686}
687
688impl AgentEvent {
689    /// A parent spawned a child.
690    pub fn spawn(run_id: i64, step: u32, child_run_id: i64, goal: impl Into<String>) -> Self {
691        Self {
692            run_id,
693            step,
694            kind: "spawn".into(),
695            child_run_id: Some(child_run_id),
696            detail: Some(goal.into()),
697            tokens: None,
698            remaining: None,
699        }
700    }
701
702    /// A spawn was refused by the containment boundary.
703    pub fn spawn_refused(run_id: i64, step: u32, cap: &str) -> Self {
704        Self {
705            run_id,
706            step,
707            kind: "spawn_refused".into(),
708            child_run_id: None,
709            detail: Some(cap.into()),
710            tokens: None,
711            remaining: None,
712        }
713    }
714
715    /// An agent drew `tokens` against the tree, leaving `remaining`.
716    pub fn budget_draw(run_id: i64, step: u32, tokens: u64, remaining: u64) -> Self {
717        Self {
718            run_id,
719            step,
720            kind: "budget_draw".into(),
721            child_run_id: None,
722            detail: None,
723            tokens: Some(tokens),
724            remaining: Some(remaining),
725        }
726    }
727}
728
729/// One event in the life of a sandboxed execution: the sandbox created for a
730/// run, a command run in it (with the backend that isolated it), a resource cap
731/// that killed it, a denied network attempt, or the sandbox torn down.
732///
733/// Together these let an operator audit not just *what* code ran but *where* and
734/// *how* it was isolated, reconstructable from the store alone after the run.
735///
736/// ```
737/// use io_harness::{SandboxEvent, Store};
738///
739/// # fn main() -> io_harness::Result<()> {
740/// # let store = Store::memory()?;
741/// # let run_id = store.start_run("make the tests pass", "src/lib.rs")?;
742/// # store.record_sandbox_event(&SandboxEvent::create(run_id, 4, "macos-sandbox-exec"))?;
743/// # store.record_sandbox_event(&SandboxEvent::exec(run_id, 4, "macos-sandbox-exec", "rustc --test subject.rs"))?;
744/// # store.record_sandbox_event(&SandboxEvent::gate_phase_failed(run_id, 4, "criterion-compile"))?;
745/// # store.record_sandbox_event(&SandboxEvent::destroy(run_id, 4))?;
746/// let events = store.sandbox_events(run_id)?;
747///
748/// // Which backend actually isolated the run. The crate picks a native one per
749/// // platform and falls back to a portable floor, so "was this really contained"
750/// // is a question about this field and not about the configuration.
751/// let backend = events.iter().find_map(|e| e.backend.as_deref());
752/// assert_eq!(backend, Some("macos-sandbox-exec"));
753///
754/// // And the phase to look for when a run that used to pass stops passing:
755/// // `criterion-compile` means the criterion no longer compiles *against* the
756/// // subject, which before 0.8.1 could be reported as a pass.
757/// let phases: Vec<&str> = events
758///     .iter()
759///     .filter(|e| e.kind == "gate_phase_failed")
760///     .filter_map(|e| e.detail.as_deref())
761///     .collect();
762/// assert_eq!(phases, ["criterion-compile"]);
763/// # Ok(())
764/// # }
765/// ```
766#[derive(Debug, Clone, PartialEq, Eq)]
767pub struct SandboxEvent {
768    /// The run this execution belongs to.
769    pub run_id: i64,
770    /// The step it occurred on.
771    pub step: u32,
772    /// `"create"`, `"exec"`, `"cap_hit"`, `"destroy"`, `"gate_phase_failed"`
773    /// (whose `detail` names the phase), or `"gate_output"` (whose `detail` is
774    /// what a failing gate command printed).
775    ///
776    /// A `"net_deny"` kind was documented here from 0.6.0 to 0.11.0 and never
777    /// existed: nothing constructed it and nothing emitted it. It was removed in
778    /// 0.12.0 rather than implemented, because a sandbox denies egress
779    /// *structurally* — the backend gives the child no route out, so there is no
780    /// attempt to observe and nothing to count. Network decisions the harness
781    /// actually makes are in `policy_events` with `act = "net"`.
782    pub kind: String,
783    /// The backend that isolated the run (e.g. `"macos-sandbox-exec"`).
784    pub backend: Option<String>,
785    /// The argv for an `"exec"`, the breached cap for a `"cap_hit"`, or the
786    /// bounded output of a failing gate command for a `"gate_output"`. Never the
787    /// agent's file contents and never credentials — the command line, or what
788    /// the caller's own criterion printed.
789    pub detail: Option<String>,
790}
791
792impl SandboxEvent {
793    /// A sandbox was created for a run, isolated by `backend`.
794    pub fn create(run_id: i64, step: u32, backend: &str) -> Self {
795        Self {
796            run_id,
797            step,
798            kind: "create".into(),
799            backend: Some(backend.into()),
800            detail: None,
801        }
802    }
803
804    /// A command ran in the sandbox under `backend`.
805    pub fn exec(run_id: i64, step: u32, backend: &str, argv: &str) -> Self {
806        Self {
807            run_id,
808            step,
809            kind: "exec".into(),
810            backend: Some(backend.into()),
811            detail: Some(argv.into()),
812        }
813    }
814
815    /// A resource cap killed the run.
816    pub fn cap_hit(run_id: i64, step: u32, cap: &str) -> Self {
817        Self {
818            run_id,
819            step,
820            kind: "cap_hit".into(),
821            backend: None,
822            detail: Some(cap.into()),
823        }
824    }
825
826    /// The sandbox was torn down (workdir removed, processes reaped).
827    pub fn destroy(run_id: i64, step: u32) -> Self {
828        Self {
829            run_id,
830            step,
831            kind: "destroy".into(),
832            backend: None,
833            detail: None,
834        }
835    }
836
837    /// Which phase of an execution gate failed: `"subject-compile"` (the file
838    /// under verification does not compile), `"criterion-compile"` (the
839    /// criterion does not compile *against* it), `"test-run"` (it compiled and
840    /// the test failed), or `"subject-emptied"` (the file compiled but a
841    /// crate-level attribute stripped its items, so nothing was type-checked —
842    /// the compile-only gates).
843    ///
844    /// 0.8.1 added this because the release deliberately makes some previously
845    /// passing runs fail. `criterion-compile` is the one to look for: before
846    /// 0.8.1 the subject and the criterion were one crate, so a subject could
847    /// shadow the names the criterion used — or delete it outright — and be
848    /// reported as passing. An operator whose run stopped passing on upgrade can
849    /// tell that case from an ordinary failed criterion without reading the
850    /// harness's source.
851    ///
852    /// A new `kind` value, not a new table or column: a 0.8.0 store takes it
853    /// with no migration.
854    pub fn gate_phase_failed(run_id: i64, step: u32, phase: &str) -> Self {
855        Self {
856            run_id,
857            step,
858            kind: "gate_phase_failed".into(),
859            backend: None,
860            detail: Some(phase.into()),
861        }
862    }
863
864    /// What a failing gate command printed, already bounded by the caller
865    /// (0.17.0).
866    ///
867    /// [`Verification::Command`](crate::Verification::Command) can run any
868    /// command in any language, so "the criterion did not pass" stops being a
869    /// self-explaining outcome: `cargo test` failing because the agent's change
870    /// is wrong and `npm test` failing because the machine has no `node_modules`
871    /// are the same discriminant and need opposite responses. The command's own
872    /// output is the only thing that tells them apart.
873    ///
874    /// This is the one `detail` that is not a command line. It is a *gate*
875    /// command's output — a caller-supplied criterion, never the agent's file
876    /// contents — and it is what makes a language-agnostic gate diagnosable at
877    /// all. Bounded by the caller before it arrives here.
878    ///
879    /// ```
880    /// use io_harness::{SandboxEvent, Store};
881    ///
882    /// # fn main() -> io_harness::Result<()> {
883    /// let store = Store::memory()?;
884    /// let run_id = store.start_run("make the suite pass", "/repo")?;
885    /// store.record_sandbox_event(&SandboxEvent::gate_output(
886    ///     run_id, 3, "FAIL test/parse.test.js\n  expected 2 received 1",
887    /// ))?;
888    ///
889    /// let why = store.sandbox_events(run_id)?;
890    /// assert!(why.iter().any(|e| e.kind == "gate_output"
891    ///     && e.detail.as_deref().is_some_and(|d| d.contains("expected 2"))));
892    /// # Ok(()) }
893    /// ```
894    ///
895    /// A new `kind` value, not a new table or column: a 0.6.0 store takes it with
896    /// no migration.
897    pub fn gate_output(run_id: i64, step: u32, output: &str) -> Self {
898        Self {
899            run_id,
900            step,
901            kind: "gate_output".into(),
902            backend: None,
903            detail: Some(output.into()),
904        }
905    }
906}
907
908/// One event in the life of an MCP connection: a server connected, a tool it
909/// offered, a tool called (with how long it took and whether it worked), or a
910/// server disconnected.
911///
912/// The `net` half of a run's egress history lives in [`PolicyEvent`] — an MCP
913/// server's host is checked by the same policy as any other outbound call — so
914/// this table is about the MCP conversation itself, not about permission.
915///
916/// ```
917/// use io_harness::{McpEvent, Store};
918///
919/// # fn main() -> io_harness::Result<()> {
920/// # let store = Store::memory()?;
921/// # let run_id = store.start_run("summarise the repo", "NOTES.md")?;
922/// # store.record_mcp(run_id, &McpEvent::connected("files", "stdio").with_millis(42))?;
923/// # store.record_mcp(run_id, &McpEvent::discovered("files", "mcp__files__read"))?;
924/// # store.record_mcp(run_id, &McpEvent::discovered("files", "mcp__files__list"))?;
925/// # store.record_mcp(run_id, &McpEvent::called("files", "mcp__files__read", true).at_step(2).with_millis(31))?;
926/// # store.record_mcp(run_id, &McpEvent::called("files", "mcp__files__read", false).at_step(3).with_millis(30_000).with_detail("timeout"))?;
927/// let events = store.mcp_events(run_id)?;
928///
929/// // What a server actually offered this run — the answer to "why did the model
930/// // not use the tool I configured", which is usually that it was never
931/// // discovered. Namespaced, so a server can never shadow `write_file`.
932/// let offered: Vec<&str> = events
933///     .iter()
934///     .filter(|e| e.kind == "discovered")
935///     .filter_map(|e| e.tool.as_deref())
936///     .collect();
937/// assert_eq!(offered, ["mcp__files__read", "mcp__files__list"]);
938///
939/// // And the call that went wrong, with how long it took before it did. Detail
940/// // carries a short note only — never arguments or results, which can carry
941/// // secrets.
942/// let failed = events.iter().find(|e| e.ok == Some(false)).expect("one failure");
943/// assert_eq!((failed.step, failed.millis), (3, Some(30_000)));
944/// assert_eq!(failed.detail.as_deref(), Some("timeout"));
945/// # Ok(())
946/// # }
947/// ```
948#[derive(Debug, Clone, PartialEq, Eq)]
949pub struct McpEvent {
950    /// The step it occurred on. `0` for connect/discover, which happen before
951    /// the run's first step.
952    pub step: u32,
953    /// `"connected"`, `"discovered"`, `"called"`, or `"disconnected"`.
954    pub kind: String,
955    /// The configured server's id.
956    pub server: String,
957    /// The namespaced tool name, for `"discovered"` and `"called"`.
958    pub tool: Option<String>,
959    /// Whether a `"called"` tool succeeded.
960    pub ok: Option<bool>,
961    /// How long a connect or call took, in milliseconds.
962    pub millis: Option<u64>,
963    /// Transport for a connect, or a note such as `"truncated"`. Never tool
964    /// arguments or results — those can carry secrets.
965    pub detail: Option<String>,
966}
967
968impl McpEvent {
969    fn new(kind: &str, server: &str) -> Self {
970        Self {
971            step: 0,
972            kind: kind.into(),
973            server: server.into(),
974            tool: None,
975            ok: None,
976            millis: None,
977            detail: None,
978        }
979    }
980
981    /// A server connected over `transport`.
982    pub fn connected(server: &str, transport: &str) -> Self {
983        Self::new("connected", server).with_detail(transport)
984    }
985
986    /// A server offered a tool, under its namespaced name.
987    pub fn discovered(server: &str, tool: &str) -> Self {
988        let mut e = Self::new("discovered", server);
989        e.tool = Some(tool.into());
990        e
991    }
992
993    /// A tool was called, and whether it worked.
994    pub fn called(server: &str, tool: &str, ok: bool) -> Self {
995        let mut e = Self::new("called", server);
996        e.tool = Some(tool.into());
997        e.ok = Some(ok);
998        e
999    }
1000
1001    /// A server was disconnected.
1002    pub fn disconnected(server: &str) -> Self {
1003        Self::new("disconnected", server)
1004    }
1005
1006    /// Attach the step this happened on.
1007    pub fn at_step(mut self, step: u32) -> Self {
1008        self.step = step;
1009        self
1010    }
1011
1012    /// Attach a duration in milliseconds.
1013    pub fn with_millis(mut self, millis: u64) -> Self {
1014        self.millis = Some(millis);
1015        self
1016    }
1017
1018    /// Attach a short note. An empty note is dropped rather than stored blank.
1019    pub fn with_detail(mut self, detail: impl Into<String>) -> Self {
1020        let detail = detail.into();
1021        self.detail = (!detail.is_empty()).then_some(detail);
1022        self
1023    }
1024}
1025
1026// ---- 0.10.0: durable cross-run memory ----
1027
1028/// One durable memory entry: a fact or decision an agent wrote deliberately,
1029/// keyed to a workspace rather than to a run.
1030///
1031/// ```
1032/// use io_harness::Store;
1033///
1034/// # fn main() -> io_harness::Result<()> {
1035/// let store = Store::memory()?;
1036/// let first = store.start_run("make the tests pass", "src/lib.rs")?;
1037///
1038/// // What an agent learned the expensive way, written once so the next run over
1039/// // this workspace does not spend three steps rediscovering it.
1040/// store.memory_put("/repo", "test-command", "cargo test --features documents", first, 6)?;
1041///
1042/// // A later run — a different process, days afterwards — recalls it by key.
1043/// let entry = store.memory_get("/repo", "test-command")?.expect("written above");
1044/// assert_eq!(entry.value, "cargo test --features documents");
1045/// // Attributed, which is what makes a stale fact traceable: this came from run
1046/// // 1, step 6, and you can go and read what that step actually did.
1047/// assert_eq!((entry.run_id, entry.step), (first, 6));
1048///
1049/// // Keyed to the workspace, never to the run: another workspace sees nothing.
1050/// assert!(store.memory_get("/other-repo", "test-command")?.is_none());
1051/// # Ok(())
1052/// # }
1053/// ```
1054#[derive(Debug, Clone, PartialEq)]
1055pub struct MemoryEntry {
1056    /// The name it is recalled by, unique within its workspace.
1057    pub key: String,
1058    /// The remembered text.
1059    pub value: String,
1060    /// The run that wrote it, so a later reader knows where a fact came from.
1061    pub run_id: i64,
1062    /// The step of that run which wrote it.
1063    pub step: u32,
1064    /// UTC write time, refreshed on every overwrite so ordering is by recency.
1065    pub created_at: String,
1066}
1067
1068/// Most entries one workspace may hold.
1069///
1070/// A write past the cap is not refused — the oldest entry is evicted to make
1071/// room, and the evicted keys come back so the caller can record the loss in the
1072/// trace rather than discovering it later as a fact that quietly stopped existing.
1073///
1074/// ```
1075/// use io_harness::{Store, MEMORY_MAX_ENTRIES};
1076///
1077/// # fn main() -> io_harness::Result<()> {
1078/// let store = Store::memory()?;
1079/// for i in 0..MEMORY_MAX_ENTRIES {
1080///     let evicted = store.memory_put("/repo", &format!("fact-{i}"), "small", 1, 1)?;
1081///     assert!(evicted.is_empty(), "everything fits until the cap");
1082/// }
1083///
1084/// // One more. The oldest goes, oldest first, and is named.
1085/// let evicted = store.memory_put("/repo", "fact-new", "small", 1, 2)?;
1086/// assert_eq!(evicted, ["fact-0"]);
1087/// assert_eq!(store.memory_list("/repo")?.len(), MEMORY_MAX_ENTRIES);
1088/// assert!(store.memory_get("/repo", "fact-0")?.is_none());
1089/// # Ok(())
1090/// # }
1091/// ```
1092pub const MEMORY_MAX_ENTRIES: usize = 64;
1093
1094/// Most characters one workspace's entries may total.
1095///
1096/// The second of the two caps, and the one that actually binds: sixty-four
1097/// entries of a paragraph each is a context section nobody budgeted for, so the
1098/// character total evicts even when the entry count is nowhere near its limit.
1099///
1100/// ```
1101/// use io_harness::{Store, MEMORY_MAX_CHARS, MEMORY_MAX_ENTRIES, MEMORY_MAX_ENTRY_CHARS};
1102///
1103/// # fn main() -> io_harness::Result<()> {
1104/// let store = Store::memory()?;
1105/// let long_note = "x".repeat(MEMORY_MAX_ENTRY_CHARS);
1106///
1107/// // Eight full-size notes fill the workspace exactly — well inside the entry
1108/// // count, and exactly on the character ceiling.
1109/// for i in 0..8 {
1110///     assert!(store.memory_put("/repo", &format!("note-{i}"), &long_note, 1, 1)?.is_empty());
1111/// }
1112/// assert!(8 < MEMORY_MAX_ENTRIES);
1113/// assert_eq!(long_note.chars().count() * 8, MEMORY_MAX_CHARS);
1114///
1115/// // The ninth evicts, on characters rather than on count.
1116/// assert_eq!(store.memory_put("/repo", "note-8", &long_note, 1, 2)?, ["note-0"]);
1117/// # Ok(())
1118/// # }
1119/// ```
1120pub const MEMORY_MAX_CHARS: usize = 16_000;
1121
1122/// Most characters one entry may hold. A longer value is cut down to this and
1123/// marked, never refused — a too-long fact is still worth remembering.
1124///
1125/// ```
1126/// use io_harness::{Store, MEMORY_MAX_ENTRY_CHARS};
1127///
1128/// # fn main() -> io_harness::Result<()> {
1129/// let store = Store::memory()?;
1130/// let rambling = "y".repeat(MEMORY_MAX_ENTRY_CHARS * 3);
1131/// store.memory_put("/repo", "build-notes", &rambling, 1, 4)?;
1132///
1133/// // Cut to the ceiling and *marked*, so a later reader can see the value is a
1134/// // fragment. A silent truncation reads like a complete fact that happens to
1135/// // end mid-sentence.
1136/// let stored = store.memory_get("/repo", "build-notes")?.expect("written above");
1137/// assert_eq!(stored.value.chars().count(), MEMORY_MAX_ENTRY_CHARS);
1138/// assert!(stored.value.ends_with("[truncated]"));
1139/// # Ok(())
1140/// # }
1141/// ```
1142pub const MEMORY_MAX_ENTRY_CHARS: usize = MEMORY_MAX_CHARS / 8;
1143
1144/// The visible marker appended to a value that was cut to the per-entry ceiling.
1145const MEMORY_TRUNCATED: &str = "…[truncated]";
1146
1147/// Cut `value` to [`MEMORY_MAX_ENTRY_CHARS`] on a char boundary, marking the
1148/// cut. Returned unchanged when it already fits.
1149fn truncate_memory_value(value: &str) -> String {
1150    if value.chars().count() <= MEMORY_MAX_ENTRY_CHARS {
1151        return value.to_string();
1152    }
1153    let keep = MEMORY_MAX_ENTRY_CHARS - MEMORY_TRUNCATED.chars().count();
1154    let mut out: String = value.chars().take(keep).collect();
1155    out.push_str(MEMORY_TRUNCATED);
1156    out
1157}
1158
1159// ---- 0.10.0: what the context assembler decided ----
1160
1161/// One decision the context assembler made: the section it built for a turn, or
1162/// a stale read it re-read (or was refused).
1163///
1164/// One row per turn plus one per re-read — never one per elided observation,
1165/// which would put a row explosion in the trace to say nothing new. Together
1166/// they answer "why did the model not see the thing it read at step 3", and
1167/// `est_tokens` beside `reported_tokens` is what records the estimator's drift
1168/// from the provider's own count (see [`crate::context::estimate_tokens`]).
1169///
1170/// ```
1171/// use io_harness::{ContextEvent, Store};
1172///
1173/// # fn main() -> io_harness::Result<()> {
1174/// # let store = Store::memory()?;
1175/// # let run_id = store.start_run("summarise the repo", "NOTES.md")?;
1176/// # store.record_context_event(run_id, &ContextEvent::assembled(3, "carried=3 stubbed=5 reread=1", 4_100))?;
1177/// # store.record_context_reported(run_id, 3, 4_690)?;
1178/// # store.record_context_event(run_id, &ContextEvent::reread_refused(3, "secrets/id_rsa: denied by policy"))?;
1179/// let events = store.context_events(run_id)?;
1180///
1181/// // Why the model did not see something it read earlier: five observations were
1182/// // stubbed to fit the budget, and one stale read could not be refreshed at all
1183/// // because the policy now refuses that path.
1184/// let refused: Vec<&str> = events
1185///     .iter()
1186///     .filter(|e| e.kind == "reread_refused")
1187///     .filter_map(|e| e.detail.as_deref())
1188///     .collect();
1189/// assert_eq!(refused, ["secrets/id_rsa: denied by policy"]);
1190///
1191/// // And the pair that makes the budget trustworthy: what the assembler
1192/// // estimated, beside what the provider actually charged for the same turn.
1193/// // Drift here is the estimator being wrong, not the budget being generous.
1194/// let assembled = events.iter().find(|e| e.kind == "assembled").expect("one per turn");
1195/// assert_eq!((assembled.est_tokens, assembled.reported_tokens), (Some(4_100), Some(4_690)));
1196/// # Ok(())
1197/// # }
1198/// ```
1199#[derive(Debug, Clone, PartialEq, Eq)]
1200pub struct ContextEvent {
1201    /// The step it belongs to.
1202    pub step: u32,
1203    /// `"assembled"`, `"reread"`, or `"reread_refused"`.
1204    pub kind: String,
1205    /// For `"assembled"`, the turn's summary (`carried=3 stubbed=5 reread=1`);
1206    /// for a re-read, the path and why it was or was not re-read.
1207    pub detail: Option<String>,
1208    /// The assembler's own estimate for the section it built.
1209    pub est_tokens: Option<u64>,
1210    /// What the provider said the request actually cost, when it says anything.
1211    pub reported_tokens: Option<u64>,
1212}
1213
1214impl ContextEvent {
1215    /// The section built for one turn.
1216    pub fn assembled(step: u32, detail: impl Into<String>, est_tokens: u64) -> Self {
1217        Self {
1218            step,
1219            kind: "assembled".into(),
1220            detail: Some(detail.into()),
1221            est_tokens: Some(est_tokens),
1222            reported_tokens: None,
1223        }
1224    }
1225
1226    /// A stale read was re-read at assembly time.
1227    pub fn reread(step: u32, detail: impl Into<String>) -> Self {
1228        Self {
1229            step,
1230            kind: "reread".into(),
1231            detail: Some(detail.into()),
1232            est_tokens: None,
1233            reported_tokens: None,
1234        }
1235    }
1236
1237    /// A stale read could not be re-read — the policy refused it, or it is gone.
1238    pub fn reread_refused(step: u32, detail: impl Into<String>) -> Self {
1239        Self {
1240            step,
1241            kind: "reread_refused".into(),
1242            detail: Some(detail.into()),
1243            est_tokens: None,
1244            reported_tokens: None,
1245        }
1246    }
1247
1248    /// The agent recorded a durable note for later runs over this workspace.
1249    pub fn memory_write(step: u32, detail: impl Into<String>) -> Self {
1250        Self::of("memory_write", step, detail)
1251    }
1252
1253    /// A note was dropped to hold the workspace's memory caps.
1254    pub fn memory_evict(step: u32, detail: impl Into<String>) -> Self {
1255        Self::of("memory_evict", step, detail)
1256    }
1257
1258    /// Notes from earlier runs were carried into this turn's context.
1259    pub fn memory_recall(step: u32, detail: impl Into<String>) -> Self {
1260        Self::of("memory_recall", step, detail)
1261    }
1262
1263    /// Which provider actually answered this step.
1264    ///
1265    /// Recorded only when the answer is not obvious from configuration — a
1266    /// [`Fallback`](crate::provider::Fallback) that fell over. `runs.provider` is one
1267    /// label for a whole run and stops being true the moment a run can use two.
1268    pub fn served(step: u32, provider: impl Into<String>) -> Self {
1269        Self::of("served", step, provider)
1270    }
1271
1272    /// The agent made no progress and was told once to change approach. The run
1273    /// continues.
1274    ///
1275    /// Split from [`Self::stalled`] in 0.12.0. Both were recorded under the one
1276    /// `"stalled"` kind, distinguishable only by prose in `detail` — so anything
1277    /// scoring a run could not tell "was nudged and carried on" from "gave up"
1278    /// without string-matching an English sentence the crate never promised.
1279    pub fn replan(step: u32, detail: impl Into<String>) -> Self {
1280        Self::of("replan", step, detail)
1281    }
1282
1283    /// The agent made no progress, had already been told once, and the run is
1284    /// ending here. Terminal.
1285    ///
1286    /// A nudge that did not work is [`Self::replan`]; see there for why the two
1287    /// are separate kinds since 0.12.0.
1288    pub fn stalled(step: u32, detail: impl Into<String>) -> Self {
1289        Self::of("stalled", step, detail)
1290    }
1291
1292    fn of(kind: &str, step: u32, detail: impl Into<String>) -> Self {
1293        Self {
1294            step,
1295            kind: kind.into(),
1296            detail: Some(detail.into()),
1297            est_tokens: None,
1298            reported_tokens: None,
1299        }
1300    }
1301}
1302
1303/// How long a contended statement waits for the writer before giving up, set on
1304/// every store opened from a file. Without it rusqlite's default is to fail
1305/// immediately with `SQLITE_BUSY`, which turns a moment of contention into an
1306/// error rather than a short wait.
1307///
1308/// ```no_run
1309/// use io_harness::{Store, BUSY_TIMEOUT};
1310///
1311/// # fn main() -> io_harness::Result<()> {
1312/// // A dashboard tailing a run another process is still writing. `Store::open`
1313/// // sets WAL and this timeout, so this read waits for the writer instead of
1314/// // failing — which is why watching a live run needs no coordination with it.
1315/// let store = Store::open("runs.sqlite3")?;
1316/// let run_id = store.last_run()?.expect("at least one run in the store");
1317/// println!("step {} so far", store.last_step(run_id)?);
1318///
1319/// // The value is public because it is the bound on how long such a read can
1320/// // block: a poller on a shorter interval than this can queue up behind a busy
1321/// // writer, and it should size its own deadline knowing that.
1322/// assert!(BUSY_TIMEOUT >= std::time::Duration::from_secs(1));
1323/// # Ok(())
1324/// # }
1325/// ```
1326pub const BUSY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
1327
1328impl Store {
1329    /// Open (creating if absent) a store at `path` and ensure the schema exists.
1330    ///
1331    /// Sets `journal_mode = WAL` and a [`BUSY_TIMEOUT`], so a second process may
1332    /// read the trace while a run is still writing it without either side
1333    /// blocking or aborting the other. Before 0.12.0 this was a bare
1334    /// `Connection::open`, which left every reader to configure the file itself
1335    /// — reaching around this API to do it, and having to do it before the
1336    /// harness opened the file at all.
1337    ///
1338    /// WAL is a persistent property of the database file, not of this
1339    /// connection: a store opened once by 0.12.0 stays in WAL mode afterwards.
1340    /// That is why it is documented as a migration.
1341    pub fn open(path: impl AsRef<std::path::Path>) -> Result<Self> {
1342        let conn = Connection::open(path)?;
1343        conn.busy_timeout(BUSY_TIMEOUT)?;
1344        // `query_row` rather than `execute`: this pragma returns the resulting
1345        // mode as a row, and rusqlite's `execute` rejects a statement that
1346        // yields rows. The returned mode is not asserted — a database on a
1347        // filesystem that cannot support WAL stays in its previous journal mode
1348        // and still works, just without concurrent readers.
1349        let _: String = conn.query_row("PRAGMA journal_mode = WAL", [], |r| r.get(0))?;
1350        Self::from_conn(conn)
1351    }
1352
1353    /// An in-memory store, for tests and throwaway runs.
1354    pub fn memory() -> Result<Self> {
1355        Self::from_conn(Connection::open_in_memory()?)
1356    }
1357
1358    fn from_conn(conn: Connection) -> Result<Self> {
1359        conn.execute_batch(
1360            "CREATE TABLE IF NOT EXISTS runs (
1361                 id       INTEGER PRIMARY KEY AUTOINCREMENT,
1362                 goal     TEXT NOT NULL,
1363                 file     TEXT NOT NULL,
1364                 outcome  TEXT,
1365                 provider TEXT
1366             );
1367             CREATE TABLE IF NOT EXISTS steps (
1368                 id       INTEGER PRIMARY KEY AUTOINCREMENT,
1369                 run_id   INTEGER NOT NULL REFERENCES runs(id),
1370                 step     INTEGER NOT NULL,
1371                 decision TEXT NOT NULL,
1372                 result   TEXT NOT NULL,
1373                 prompt    TEXT NOT NULL DEFAULT '',
1374                 tool_call TEXT NOT NULL DEFAULT '',
1375                 tokens    INTEGER NOT NULL DEFAULT 0
1376             );",
1377        )?;
1378
1379        // Migrate a 0.1.0 database whose `steps` table predates the trace
1380        // columns. ADD COLUMN errors on an already-present column; ignore it.
1381        for col in [
1382            "prompt TEXT NOT NULL DEFAULT ''",
1383            "tool_call TEXT NOT NULL DEFAULT ''",
1384            "tokens INTEGER NOT NULL DEFAULT 0",
1385        ] {
1386            let _ = conn.execute(&format!("ALTER TABLE steps ADD COLUMN {col}"), []);
1387        }
1388        // 0.3.0: record which provider ran. Additive — a 0.1/0.2 database gains
1389        // the column and a 0.2 binary still reads a migrated database.
1390        let _ = conn.execute("ALTER TABLE runs ADD COLUMN provider TEXT", []);
1391
1392        // 0.4.0: policy refusals/decisions, and actions paused awaiting a human.
1393        // New tables only — a 0.3.0 database gains them and a 0.3.0 binary,
1394        // which never queries them, still reads a migrated database.
1395        conn.execute_batch(
1396            "CREATE TABLE IF NOT EXISTS policy_events (
1397                 id        INTEGER PRIMARY KEY AUTOINCREMENT,
1398                 run_id    INTEGER NOT NULL,
1399                 step      INTEGER NOT NULL,
1400                 kind      TEXT NOT NULL,
1401                 act       TEXT NOT NULL,
1402                 target    TEXT NOT NULL,
1403                 rule      TEXT,
1404                 layer     TEXT,
1405                 decision  TEXT,
1406                 source    TEXT,
1407                 performed TEXT
1408             );
1409             CREATE TABLE IF NOT EXISTS pending_approvals (
1410                 id       INTEGER PRIMARY KEY AUTOINCREMENT,
1411                 run_id   INTEGER NOT NULL,
1412                 step     INTEGER NOT NULL,
1413                 act      TEXT NOT NULL,
1414                 target   TEXT NOT NULL,
1415                 content  TEXT,
1416                 resolved TEXT
1417             );",
1418        )?;
1419
1420        // 0.5.0: sub-agent trees. Runs gain a parent edge and a depth; a new
1421        // table records spawns, spawn refusals, and draws against the tree's
1422        // shared spend ceiling. All additive — a 0.4.0 database gains the column
1423        // and table and a 0.4.0 binary still reads a migrated database.
1424        let _ = conn.execute("ALTER TABLE runs ADD COLUMN parent_run_id INTEGER", []);
1425        let _ = conn.execute(
1426            "ALTER TABLE runs ADD COLUMN depth INTEGER NOT NULL DEFAULT 0",
1427            [],
1428        );
1429        conn.execute_batch(
1430            "CREATE TABLE IF NOT EXISTS agent_events (
1431                 id           INTEGER PRIMARY KEY AUTOINCREMENT,
1432                 run_id       INTEGER NOT NULL,
1433                 step         INTEGER NOT NULL,
1434                 kind         TEXT NOT NULL,
1435                 child_run_id INTEGER,
1436                 detail       TEXT,
1437                 tokens       INTEGER,
1438                 remaining    INTEGER
1439             );",
1440        )?;
1441
1442        // 0.6.0: sandbox lifecycle events (create, exec+backend, cap hit, net
1443        // deny, destroy). New table only — a 0.5.0 database gains it and a 0.5.0
1444        // binary, which never queries it, still reads a migrated database.
1445        conn.execute_batch(
1446            "CREATE TABLE IF NOT EXISTS sandbox_events (
1447                 id       INTEGER PRIMARY KEY AUTOINCREMENT,
1448                 run_id   INTEGER NOT NULL,
1449                 step     INTEGER NOT NULL,
1450                 kind     TEXT NOT NULL,
1451                 backend  TEXT,
1452                 detail   TEXT
1453             );",
1454        )?;
1455
1456        // 0.7.0: durable checkpoint + resume. `runs` gains a resumable status and
1457        // a start timestamp so wall-clock elapsed survives a restart; a new table
1458        // records checkpoint / resume / step-skipped events so a multi-crash run's
1459        // history is reconstructable from the store alone. All additive — a 0.6.0
1460        // database gains the columns/table and a 0.6.0 binary still reads it.
1461        let _ = conn.execute(
1462            "ALTER TABLE runs ADD COLUMN status TEXT NOT NULL DEFAULT 'running'",
1463            [],
1464        );
1465        let _ = conn.execute("ALTER TABLE runs ADD COLUMN started_at TEXT", []);
1466        conn.execute_batch(
1467            "CREATE TABLE IF NOT EXISTS checkpoint_events (
1468                 id     INTEGER PRIMARY KEY AUTOINCREMENT,
1469                 run_id INTEGER NOT NULL,
1470                 step   INTEGER NOT NULL,
1471                 kind   TEXT NOT NULL,
1472                 detail TEXT
1473             );
1474             CREATE TABLE IF NOT EXISTS spawns (
1475                 id            INTEGER PRIMARY KEY AUTOINCREMENT,
1476                 parent_run_id INTEGER NOT NULL,
1477                 step          INTEGER NOT NULL,
1478                 child_run_id  INTEGER NOT NULL,
1479                 goal          TEXT NOT NULL,
1480                 verify_file   TEXT NOT NULL,
1481                 needle        TEXT NOT NULL,
1482                 max_steps     INTEGER,
1483                 deny_write    TEXT NOT NULL DEFAULT '[]'
1484             );",
1485        )?;
1486
1487        // 0.8.0: the MCP conversation — connects, tool discovery, tool calls,
1488        // disconnects. New table only, so a 0.7.0 database gains it and a 0.7.0
1489        // binary, which never queries it, still reads a migrated database. The
1490        // network *verdicts* deliberately do not live here: they go to
1491        // policy_events beside every other permission decision, because an
1492        // operator auditing "what was this run allowed to do" should find them
1493        // in one place.
1494        conn.execute_batch(
1495            "CREATE TABLE IF NOT EXISTS mcp_events (
1496                 id     INTEGER PRIMARY KEY AUTOINCREMENT,
1497                 run_id INTEGER NOT NULL,
1498                 step   INTEGER NOT NULL,
1499                 kind   TEXT NOT NULL,
1500                 server TEXT NOT NULL,
1501                 tool   TEXT,
1502                 ok     INTEGER,
1503                 millis INTEGER,
1504                 detail TEXT
1505             );",
1506        )?;
1507
1508        // 0.10.0: durable cross-run memory — facts and decisions an agent wrote
1509        // deliberately, keyed to a *workspace* instead of a run, so a later run
1510        // recalls what an earlier one learned. New table only, so a 0.9.1
1511        // database gains it and a 0.9.1 binary, which never queries it, still
1512        // reads a migrated database. Deliberately NOT a CHECKPOINT_FORMAT bump:
1513        // no checkpoint layout changed, and bumping it would make
1514        // [`Store::check_resumable`] refuse every 0.9.1 checkpoint.
1515        conn.execute_batch(
1516            "CREATE TABLE IF NOT EXISTS memory (
1517                 id         INTEGER PRIMARY KEY,
1518                 workspace  TEXT NOT NULL,
1519                 key        TEXT NOT NULL,
1520                 value      TEXT NOT NULL,
1521                 run_id     INTEGER NOT NULL,
1522                 step       INTEGER NOT NULL,
1523                 created_at TEXT NOT NULL,
1524                 UNIQUE(workspace, key)
1525             );",
1526        )?;
1527
1528        // 0.10.0: what the context assembler decided each turn — one row per turn
1529        // plus one per re-read. New table only, so a 0.9.1 database gains it and a
1530        // 0.9.1 binary, which never queries it, still opens and resumes a migrated
1531        // database. Deliberately NOT a `CHECKPOINT_FORMAT` bump: nothing about a
1532        // checkpoint's layout changed, and bumping it would refuse every 0.9.1
1533        // store on resume for an additive audit table.
1534        conn.execute_batch(
1535            "CREATE TABLE IF NOT EXISTS context_events (
1536                 id              INTEGER PRIMARY KEY,
1537                 run_id          INTEGER NOT NULL,
1538                 step            INTEGER NOT NULL,
1539                 kind            TEXT NOT NULL,
1540                 detail          TEXT,
1541                 est_tokens      INTEGER,
1542                 reported_tokens INTEGER
1543             );",
1544        )?;
1545
1546        // 0.12.0: one row per finished run, so "did it work, how long did it take,
1547        // what did it cost" is one read rather than a reconstruction.
1548        //
1549        // Every field but the end stamp was already derivable, and derivable was
1550        // not good enough: a consumer had to know that success is one of eleven
1551        // free-text strings, that steps means MAX(step) and not COUNT(*) because
1552        // retry rows share a step number, and that spend is SUM(steps.tokens). That
1553        // is schema knowledge the crate never promised, so io-eval would have been
1554        // coupled to internals from its first line.
1555        //
1556        // `finished_at` is the genuinely new fact. Nothing in the schema recorded
1557        // when a run ENDED — only `runs.started_at` — and `Store::elapsed_secs`
1558        // measures against `julianday('now')`, so it keeps growing after the run is
1559        // over and cannot reconstruct a finished run's latency. Stamped from
1560        // SQLite's clock for the same reason `started_at` is: the pair must come
1561        // from one clock or the difference is meaningless.
1562        //
1563        // A separate table rather than columns on `runs`: additive, and `runs` is
1564        // read by resume on the hot path. New table only, so a 0.11.0 database gains
1565        // it and a 0.11.0 binary, which never queries it, still opens and resumes a
1566        // migrated database. Deliberately NOT a `CHECKPOINT_FORMAT` bump — no
1567        // checkpoint layout changed, and bumping it would make
1568        // [`Store::check_resumable`] refuse every 0.11.0 store for an additive table.
1569        conn.execute_batch(
1570            "CREATE TABLE IF NOT EXISTS run_outcomes (
1571                 run_id      INTEGER PRIMARY KEY,
1572                 outcome     TEXT NOT NULL,
1573                 success     INTEGER NOT NULL,
1574                 steps       INTEGER NOT NULL,
1575                 tokens      INTEGER NOT NULL,
1576                 duration_ms INTEGER,
1577                 finished_at TEXT NOT NULL
1578             );",
1579        )?;
1580
1581        // 0.13.0: the policy a run was started under, kept so a later resume can
1582        // tell what boundary the caller enforced instead of guessing. Nothing in
1583        // the schema recorded it: `policy_events` holds the decisions a policy
1584        // produced, which is the opposite direction — a run that was never asked
1585        // to do anything forbidden leaves no events at all, and a permissive run
1586        // leaves none either, so the two are indistinguishable after the fact.
1587        //
1588        // Stored as JSON in one column rather than shredded into rule rows: the
1589        // only reader wants the whole [`Policy`] back, and a serialised blob
1590        // cannot drift from the type the way a hand-written flattening would.
1591        //
1592        // New table only, so a 0.12.0 database gains it and a 0.12.0 binary, which
1593        // never queries it, still opens and resumes a migrated database.
1594        // Deliberately NOT a `CHECKPOINT_FORMAT` bump: no checkpoint layout
1595        // changed, and bumping it would make [`Store::check_resumable`] refuse
1596        // every 0.12.0 store for an additive table.
1597        conn.execute_batch(
1598            "CREATE TABLE IF NOT EXISTS run_policies (
1599                 run_id INTEGER PRIMARY KEY,
1600                 policy TEXT NOT NULL
1601             );",
1602        )?;
1603
1604        // 0.13.0: the observation ledger the context assembler builds, made
1605        // durable so a resumed run restores the context it had instead of
1606        // re-deriving one from the workspace.
1607        //
1608        // The text was already durable — `steps.result` holds one step's
1609        // observations concatenated — but concatenated is the problem: a step with
1610        // three observations stores one string, and the typed triple assembly
1611        // actually reasons about (`step`, `kind`, `target`) is not recoverable
1612        // from it at all. `ObsKind::target_is_the_subject` decides supersession
1613        // from `kind`, so a ledger rebuilt from `steps.result` would assemble
1614        // differently from the one it replaced, which is worse than the honest
1615        // re-derivation it would be replacing.
1616        //
1617        // One row per observation, ordered by `id` like every other event table
1618        // here, because the ledger is an ordered log and `step` alone does not
1619        // order the observations within a step.
1620        //
1621        // New table only, so a 0.12.0 database gains it and a 0.12.0 binary, which
1622        // never queries it, still opens and resumes a migrated database.
1623        // Deliberately NOT a `CHECKPOINT_FORMAT` bump: no checkpoint layout
1624        // changed, and bumping it would make [`Store::check_resumable`] refuse
1625        // every 0.12.0 store for an additive table.
1626        conn.execute_batch(
1627            "CREATE TABLE IF NOT EXISTS ledger_observations (
1628                 id     INTEGER PRIMARY KEY,
1629                 run_id INTEGER NOT NULL,
1630                 step   INTEGER NOT NULL,
1631                 kind   TEXT NOT NULL,
1632                 target TEXT,
1633                 text   TEXT NOT NULL
1634             );",
1635        )?;
1636
1637        // Stamp the checkpoint-format version. A fresh or pre-0.7.0 database reads
1638        // back 0; we bump it to the current format. A database written by a NEWER
1639        // format reads back a higher number and [`Store::check_resumable`] refuses
1640        // it with a typed [`Error::Resume`] rather than resuming a layout it does
1641        // not understand.
1642        let format: i64 = conn.query_row("PRAGMA user_version", [], |r| r.get(0))?;
1643        if format < CHECKPOINT_FORMAT {
1644            conn.execute_batch(&format!("PRAGMA user_version = {CHECKPOINT_FORMAT}"))?;
1645        }
1646
1647        Ok(Self { conn })
1648    }
1649
1650    /// Record a policy refusal or a human decision against a run.
1651    pub fn record_event(&self, run_id: i64, e: &PolicyEvent) -> Result<()> {
1652        self.conn.execute(
1653            "INSERT INTO policy_events
1654                 (run_id, step, kind, act, target, rule, layer, decision, source, performed)
1655             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
1656            (
1657                run_id,
1658                e.step,
1659                &e.kind,
1660                &e.act,
1661                &e.target,
1662                &e.rule,
1663                &e.layer,
1664                &e.decision,
1665                &e.source,
1666                &e.performed,
1667            ),
1668        )?;
1669        Ok(())
1670    }
1671
1672    /// Every policy event recorded for a run, in order.
1673    pub fn events(&self, run_id: i64) -> Result<Vec<PolicyEvent>> {
1674        let mut stmt = self.conn.prepare(
1675            "SELECT step, kind, act, target, rule, layer, decision, source, performed
1676             FROM policy_events WHERE run_id = ?1 ORDER BY id ASC",
1677        )?;
1678        let rows = stmt.query_map([run_id], |r| {
1679            Ok(PolicyEvent {
1680                step: r.get::<_, i64>(0)? as u32,
1681                kind: r.get(1)?,
1682                act: r.get(2)?,
1683                target: r.get(3)?,
1684                rule: r.get(4)?,
1685                layer: r.get(5)?,
1686                decision: r.get(6)?,
1687                source: r.get(7)?,
1688                performed: r.get(8)?,
1689            })
1690        })?;
1691        Ok(rows.collect::<std::result::Result<_, _>>()?)
1692    }
1693
1694    /// Persist an action awaiting a human decision; returns its request id.
1695    pub fn put_pending(
1696        &self,
1697        run_id: i64,
1698        step: u32,
1699        act: &str,
1700        target: &str,
1701        content: Option<&str>,
1702    ) -> Result<i64> {
1703        self.conn.execute(
1704            "INSERT INTO pending_approvals (run_id, step, act, target, content)
1705             VALUES (?1, ?2, ?3, ?4, ?5)",
1706            (run_id, step, act, target, content),
1707        )?;
1708        Ok(self.conn.last_insert_rowid())
1709    }
1710
1711    /// Read a pending action back by request id.
1712    pub fn pending(&self, request_id: i64) -> Result<Option<Pending>> {
1713        let mut stmt = self.conn.prepare(
1714            "SELECT id, run_id, step, act, target, content, resolved
1715             FROM pending_approvals WHERE id = ?1",
1716        )?;
1717        let mut rows = stmt.query_map([request_id], |r| {
1718            Ok(Pending {
1719                id: r.get(0)?,
1720                run_id: r.get(1)?,
1721                step: r.get::<_, i64>(2)? as u32,
1722                act: r.get(3)?,
1723                target: r.get(4)?,
1724                content: r.get(5)?,
1725                resolved: r.get(6)?,
1726            })
1727        })?;
1728        Ok(rows.next().transpose()?)
1729    }
1730
1731    /// Mark a pending action decided, so a resume knows what the human chose.
1732    pub fn resolve_pending(&self, request_id: i64, decision: &str) -> Result<()> {
1733        self.conn.execute(
1734            "UPDATE pending_approvals SET resolved = ?1 WHERE id = ?2",
1735            (decision, request_id),
1736        )?;
1737        Ok(())
1738    }
1739
1740    /// Start a run row; returns its id. Stamps `started_at` (UTC, from SQLite's
1741    /// clock) so a 24h wall-clock budget survives a restart, and marks the run
1742    /// `running`.
1743    pub fn start_run(&self, goal: &str, file: &str) -> Result<i64> {
1744        self.conn.execute(
1745            "INSERT INTO runs (goal, file, status, started_at)
1746             VALUES (?1, ?2, 'running', strftime('%Y-%m-%dT%H:%M:%fZ','now'))",
1747            (goal, file),
1748        )?;
1749        Ok(self.conn.last_insert_rowid())
1750    }
1751
1752    /// Start a child run under `parent_run_id` at `depth`, so the tree records
1753    /// who spawned whom. Returns the child's run id.
1754    pub fn start_child_run(
1755        &self,
1756        goal: &str,
1757        file: &str,
1758        parent_run_id: i64,
1759        depth: u32,
1760    ) -> Result<i64> {
1761        self.conn.execute(
1762            "INSERT INTO runs (goal, file, parent_run_id, depth, status, started_at)
1763             VALUES (?1, ?2, ?3, ?4, 'running', strftime('%Y-%m-%dT%H:%M:%fZ','now'))",
1764            (goal, file, parent_run_id, depth),
1765        )?;
1766        Ok(self.conn.last_insert_rowid())
1767    }
1768
1769    /// Record a spawn, a spawn refusal, or a budget draw against the tree.
1770    pub fn record_agent_event(&self, e: &AgentEvent) -> Result<()> {
1771        self.conn.execute(
1772            "INSERT INTO agent_events (run_id, step, kind, child_run_id, detail, tokens, remaining)
1773             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
1774            (
1775                e.run_id,
1776                e.step,
1777                &e.kind,
1778                e.child_run_id,
1779                &e.detail,
1780                e.tokens,
1781                e.remaining,
1782            ),
1783        )?;
1784        Ok(())
1785    }
1786
1787    /// Every agent event recorded for a run, in order.
1788    pub fn agent_events(&self, run_id: i64) -> Result<Vec<AgentEvent>> {
1789        let mut stmt = self.conn.prepare(
1790            "SELECT run_id, step, kind, child_run_id, detail, tokens, remaining
1791             FROM agent_events WHERE run_id = ?1 ORDER BY id ASC",
1792        )?;
1793        let rows = stmt.query_map([run_id], |r| {
1794            Ok(AgentEvent {
1795                run_id: r.get(0)?,
1796                step: r.get::<_, i64>(1)? as u32,
1797                kind: r.get(2)?,
1798                child_run_id: r.get(3)?,
1799                detail: r.get(4)?,
1800                tokens: r.get::<_, Option<i64>>(5)?.map(|n| n as u64),
1801                remaining: r.get::<_, Option<i64>>(6)?.map(|n| n as u64),
1802            })
1803        })?;
1804        Ok(rows.collect::<std::result::Result<_, _>>()?)
1805    }
1806
1807    /// Record one sandbox lifecycle event against a run.
1808    pub fn record_sandbox_event(&self, e: &SandboxEvent) -> Result<()> {
1809        self.conn.execute(
1810            "INSERT INTO sandbox_events (run_id, step, kind, backend, detail)
1811             VALUES (?1, ?2, ?3, ?4, ?5)",
1812            (e.run_id, e.step, &e.kind, &e.backend, &e.detail),
1813        )?;
1814        Ok(())
1815    }
1816
1817    /// Record one MCP event.
1818    pub fn record_mcp(&self, run_id: i64, e: &McpEvent) -> Result<()> {
1819        self.conn.execute(
1820            "INSERT INTO mcp_events (run_id, step, kind, server, tool, ok, millis, detail)
1821             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
1822            (
1823                run_id,
1824                e.step,
1825                &e.kind,
1826                &e.server,
1827                &e.tool,
1828                e.ok,
1829                e.millis.map(|m| m as i64),
1830                &e.detail,
1831            ),
1832        )?;
1833        Ok(())
1834    }
1835
1836    /// Every MCP event recorded for a run, in order.
1837    pub fn mcp_events(&self, run_id: i64) -> Result<Vec<McpEvent>> {
1838        let mut stmt = self.conn.prepare(
1839            "SELECT step, kind, server, tool, ok, millis, detail
1840             FROM mcp_events WHERE run_id = ?1 ORDER BY id ASC",
1841        )?;
1842        let rows = stmt.query_map([run_id], |r| {
1843            Ok(McpEvent {
1844                step: r.get::<_, i64>(0)? as u32,
1845                kind: r.get(1)?,
1846                server: r.get(2)?,
1847                tool: r.get(3)?,
1848                ok: r.get(4)?,
1849                millis: r.get::<_, Option<i64>>(5)?.map(|m| m as u64),
1850                detail: r.get(6)?,
1851            })
1852        })?;
1853        Ok(rows.collect::<std::result::Result<_, _>>()?)
1854    }
1855
1856    /// Record one context-assembly event against a run.
1857    pub fn record_context_event(&self, run_id: i64, e: &ContextEvent) -> Result<()> {
1858        self.conn.execute(
1859            "INSERT INTO context_events (run_id, step, kind, detail, est_tokens, reported_tokens)
1860             VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
1861            (
1862                run_id,
1863                e.step,
1864                &e.kind,
1865                &e.detail,
1866                e.est_tokens.map(|n| n as i64),
1867                e.reported_tokens.map(|n| n as i64),
1868            ),
1869        )?;
1870        Ok(())
1871    }
1872
1873    /// Fill in what the provider said one turn's request cost, once the
1874    /// completion has returned. The estimate is left as it was: the pair is the
1875    /// point — one row carries both numbers, so drift is readable.
1876    pub fn record_context_reported(&self, run_id: i64, step: u32, reported: u64) -> Result<()> {
1877        self.conn.execute(
1878            "UPDATE context_events SET reported_tokens = ?1
1879             WHERE run_id = ?2 AND step = ?3 AND kind = 'assembled'",
1880            (reported as i64, run_id, step),
1881        )?;
1882        Ok(())
1883    }
1884
1885    /// Every context-assembly event recorded for a run, in order.
1886    pub fn context_events(&self, run_id: i64) -> Result<Vec<ContextEvent>> {
1887        let mut stmt = self.conn.prepare(
1888            "SELECT step, kind, detail, est_tokens, reported_tokens
1889             FROM context_events WHERE run_id = ?1 ORDER BY id ASC",
1890        )?;
1891        let rows = stmt.query_map([run_id], |r| {
1892            Ok(ContextEvent {
1893                step: r.get::<_, i64>(0)? as u32,
1894                kind: r.get(1)?,
1895                detail: r.get(2)?,
1896                est_tokens: r.get::<_, Option<i64>>(3)?.map(|n| n as u64),
1897                reported_tokens: r.get::<_, Option<i64>>(4)?.map(|n| n as u64),
1898            })
1899        })?;
1900        Ok(rows.collect::<std::result::Result<_, _>>()?)
1901    }
1902
1903    /// Record the policy a run was started under.
1904    ///
1905    /// `INSERT OR REPLACE`, like every other per-run row, so recording twice for
1906    /// one run — a resume that re-states its boundary — replaces rather than
1907    /// duplicates or fails.
1908    pub fn record_run_policy(&self, run_id: i64, policy: &Policy) -> Result<()> {
1909        self.conn.execute(
1910            "INSERT OR REPLACE INTO run_policies (run_id, policy) VALUES (?1, ?2)",
1911            (
1912                run_id,
1913                serde_json::to_string(policy).expect("a Policy is always serialisable"),
1914            ),
1915        )?;
1916        Ok(())
1917    }
1918
1919    /// The policy a run was started under, or `None` if none was recorded.
1920    ///
1921    /// `None` is not [`Policy::permissive`] and must never be read as it: a run
1922    /// written by 0.12.0 has no row at all, so the honest answer is "nobody
1923    /// recorded what the boundary was", not "the caller chose to enforce
1924    /// nothing". A caller that needs a policy either way has to decide which to
1925    /// assume, and it should decide that knowingly.
1926    /// Unlike the other getters in this file, a failed read is an error rather
1927    /// than `None`. They can fold the two together because a missing memory
1928    /// entry and an unreadable one lead to the same recovery; here they do not.
1929    /// `None` is what tells [`crate::resume`] the run had no boundary and may be
1930    /// resumed permissively, so a disk error that read as `None` would hand a
1931    /// policy-bearing run an agent with no policy — silently, and by exactly the
1932    /// route this table exists to close.
1933    pub fn run_policy(&self, run_id: i64) -> Result<Option<Policy>> {
1934        let json: Option<String> = match self.conn.query_row(
1935            "SELECT policy FROM run_policies WHERE run_id = ?1",
1936            [run_id],
1937            |r| r.get(0),
1938        ) {
1939            Ok(json) => Some(json),
1940            Err(rusqlite::Error::QueryReturnedNoRows) => None,
1941            Err(e) => return Err(e.into()),
1942        };
1943        json.map(|j| {
1944            serde_json::from_str(&j).map_err(|e| Error::Resume {
1945                reason: format!("run {run_id} has an unreadable recorded policy: {e}"),
1946            })
1947        })
1948        .transpose()
1949    }
1950
1951    /// Append observations to a run's durable ledger, in one transaction.
1952    ///
1953    /// Called once at a committed step boundary rather than once per
1954    /// observation: the step is the unit the rest of the checkpoint works in, and
1955    /// an observation belonging to a step that never committed must not survive a
1956    /// crash the step itself did not survive.
1957    pub fn record_observations(&self, run_id: i64, entries: &[Observation]) -> Result<()> {
1958        if entries.is_empty() {
1959            return Ok(());
1960        }
1961        let tx = self.conn.unchecked_transaction()?;
1962        {
1963            let mut stmt = tx.prepare(
1964                "INSERT INTO ledger_observations (run_id, step, kind, target, text)
1965                 VALUES (?1, ?2, ?3, ?4, ?5)",
1966            )?;
1967            for e in entries {
1968                stmt.execute((run_id, e.step as i64, kind_wire(e.kind), &e.target, &e.text))?;
1969            }
1970        }
1971        tx.commit()?;
1972        Ok(())
1973    }
1974
1975    /// A run's durable ledger, in the order it was observed.
1976    ///
1977    /// Empty for a run that recorded nothing and for a run written before 0.13.0
1978    /// — the two are the same to a reader, and both mean "there is nothing to
1979    /// restore", which is 0.12.0's behaviour and not a lie about it.
1980    pub fn observations(&self, run_id: i64) -> Result<Vec<Observation>> {
1981        let mut stmt = self.conn.prepare(
1982            "SELECT step, kind, target, text
1983             FROM ledger_observations WHERE run_id = ?1 ORDER BY id ASC",
1984        )?;
1985        let rows = stmt.query_map([run_id], |r| {
1986            Ok((
1987                r.get::<_, i64>(0)? as u32,
1988                r.get::<_, String>(1)?,
1989                r.get::<_, Option<String>>(2)?,
1990                r.get::<_, String>(3)?,
1991            ))
1992        })?;
1993        let mut out = Vec::new();
1994        for row in rows {
1995            let (step, kind, target, text) = row?;
1996            out.push(Observation::new(
1997                step,
1998                kind_from_wire(&kind, run_id)?,
1999                target,
2000                text,
2001            ));
2002        }
2003        Ok(out)
2004    }
2005
2006    /// Every sandbox event recorded for a run, in order.
2007    pub fn sandbox_events(&self, run_id: i64) -> Result<Vec<SandboxEvent>> {
2008        let mut stmt = self.conn.prepare(
2009            "SELECT run_id, step, kind, backend, detail
2010             FROM sandbox_events WHERE run_id = ?1 ORDER BY id ASC",
2011        )?;
2012        let rows = stmt.query_map([run_id], |r| {
2013            Ok(SandboxEvent {
2014                run_id: r.get(0)?,
2015                step: r.get::<_, i64>(1)? as u32,
2016                kind: r.get(2)?,
2017                backend: r.get(3)?,
2018                detail: r.get(4)?,
2019            })
2020        })?;
2021        Ok(rows.collect::<std::result::Result<_, _>>()?)
2022    }
2023
2024    /// The run ids of the direct children of `run_id`, in spawn order.
2025    pub fn children(&self, run_id: i64) -> Result<Vec<i64>> {
2026        let mut stmt = self
2027            .conn
2028            .prepare("SELECT id FROM runs WHERE parent_run_id = ?1 ORDER BY id ASC")?;
2029        let rows = stmt.query_map([run_id], |r| r.get(0))?;
2030        Ok(rows.collect::<std::result::Result<_, _>>()?)
2031    }
2032
2033    /// The parent run id of `run_id`, or `None` for a root run.
2034    pub fn parent(&self, run_id: i64) -> Result<Option<i64>> {
2035        Ok(self.conn.query_row(
2036            "SELECT parent_run_id FROM runs WHERE id = ?1",
2037            [run_id],
2038            |r| r.get(0),
2039        )?)
2040    }
2041
2042    /// The nesting depth recorded for a run (0 at the root).
2043    pub fn depth(&self, run_id: i64) -> Result<u32> {
2044        let d: i64 =
2045            self.conn
2046                .query_row("SELECT depth FROM runs WHERE id = ?1", [run_id], |r| {
2047                    r.get(0)
2048                })?;
2049        Ok(d as u32)
2050    }
2051
2052    /// Record one step's full trace entry.
2053    pub fn record(&self, run_id: i64, step: &StepRecord) -> Result<()> {
2054        self.conn.execute(
2055            "INSERT INTO steps (run_id, step, decision, result, prompt, tool_call, tokens)
2056             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
2057            (
2058                run_id,
2059                step.step,
2060                &step.decision,
2061                &step.result,
2062                &step.prompt,
2063                &step.tool_call,
2064                step.tokens,
2065            ),
2066        )?;
2067        Ok(())
2068    }
2069
2070    /// Durably checkpoint one completed step: the step's trace row and its
2071    /// checkpoint event are written in a single transaction, so a crash leaves
2072    /// either both (the step is done) or neither (it replays) — never a torn
2073    /// half. The committed checkpoint is the step's completion marker.
2074    pub fn checkpoint_step(&self, run_id: i64, step: &StepRecord) -> Result<()> {
2075        let tx = self.conn.unchecked_transaction()?;
2076        tx.execute(
2077            "INSERT INTO steps (run_id, step, decision, result, prompt, tool_call, tokens)
2078             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
2079            (
2080                run_id,
2081                step.step,
2082                &step.decision,
2083                &step.result,
2084                &step.prompt,
2085                &step.tool_call,
2086                step.tokens,
2087            ),
2088        )?;
2089        tx.execute(
2090            "INSERT INTO checkpoint_events (run_id, step, kind, detail)
2091             VALUES (?1, ?2, 'checkpoint', NULL)",
2092            (run_id, step.step),
2093        )?;
2094        tx.commit()?;
2095        Ok(())
2096    }
2097
2098    /// Record a checkpoint/resume/skipped event on its own (not tied to a step
2099    /// commit) — used for resume and skip markers.
2100    pub fn record_checkpoint_event(&self, e: &CheckpointEvent) -> Result<()> {
2101        self.conn.execute(
2102            "INSERT INTO checkpoint_events (run_id, step, kind, detail) VALUES (?1, ?2, ?3, ?4)",
2103            (e.run_id, e.step, &e.kind, &e.detail),
2104        )?;
2105        Ok(())
2106    }
2107
2108    /// Every checkpoint-lifecycle event recorded for a run, in order.
2109    pub fn checkpoint_events(&self, run_id: i64) -> Result<Vec<CheckpointEvent>> {
2110        let mut stmt = self.conn.prepare(
2111            "SELECT run_id, step, kind, detail
2112             FROM checkpoint_events WHERE run_id = ?1 ORDER BY id ASC",
2113        )?;
2114        let rows = stmt.query_map([run_id], |r| {
2115            Ok(CheckpointEvent {
2116                run_id: r.get(0)?,
2117                step: r.get::<_, i64>(1)? as u32,
2118                kind: r.get(2)?,
2119                detail: r.get(3)?,
2120            })
2121        })?;
2122        Ok(rows.collect::<std::result::Result<_, _>>()?)
2123    }
2124
2125    /// Set the durable run status (`running`, `paused`, `completed`, `failed`).
2126    pub fn set_status(&self, run_id: i64, status: &str) -> Result<()> {
2127        self.conn.execute(
2128            "UPDATE runs SET status = ?1 WHERE id = ?2",
2129            (status, run_id),
2130        )?;
2131        Ok(())
2132    }
2133
2134    /// The durable run status, if the run exists.
2135    pub fn status(&self, run_id: i64) -> Result<Option<String>> {
2136        Ok(self
2137            .conn
2138            .query_row("SELECT status FROM runs WHERE id = ?1", [run_id], |r| {
2139                r.get(0)
2140            })
2141            .ok())
2142    }
2143
2144    /// Real wall-clock seconds elapsed since the run's `started_at`, from the
2145    /// database clock — so a budget over duration counts time that passed while
2146    /// the process was down, not just this process's uptime. Zero if the run has
2147    /// no start stamp (a pre-0.7.0 run).
2148    pub fn elapsed_secs(&self, run_id: i64) -> Result<f64> {
2149        let secs: Option<f64> = self.conn.query_row(
2150            "SELECT (julianday('now') - julianday(started_at)) * 86400.0
2151             FROM runs WHERE id = ?1",
2152            [run_id],
2153            |r| r.get(0),
2154        )?;
2155        Ok(secs.unwrap_or(0.0).max(0.0))
2156    }
2157
2158    /// Total tokens recorded across this run's steps — the durable spend, so a
2159    /// resume restores the token budget instead of restarting it at zero.
2160    pub fn spent_tokens(&self, run_id: i64) -> Result<u64> {
2161        let n: i64 = self.conn.query_row(
2162            "SELECT COALESCE(SUM(tokens), 0) FROM steps WHERE run_id = ?1",
2163            [run_id],
2164            |r| r.get(0),
2165        )?;
2166        Ok(n as u64)
2167    }
2168
2169    /// Every run id in the tree rooted at `root` (the root plus all descendants),
2170    /// via the `parent_run_id` edge — the set a tree-level resume re-drives.
2171    pub fn tree_run_ids(&self, root: i64) -> Result<Vec<i64>> {
2172        let mut stmt = self.conn.prepare(
2173            "WITH RECURSIVE tree(id) AS (
2174                 SELECT id FROM runs WHERE id = ?1
2175                 UNION ALL
2176                 SELECT r.id FROM runs r JOIN tree t ON r.parent_run_id = t.id
2177             )
2178             SELECT id FROM tree ORDER BY id ASC",
2179        )?;
2180        let rows = stmt.query_map([root], |r| r.get(0))?;
2181        Ok(rows.collect::<std::result::Result<_, _>>()?)
2182    }
2183
2184    /// Total tokens spent across the whole tree rooted at `root` — the durable
2185    /// aggregate-ledger spend restored on a tree resume.
2186    pub fn spent_tokens_tree(&self, root: i64) -> Result<u64> {
2187        let n: i64 = self.conn.query_row(
2188            "WITH RECURSIVE tree(id) AS (
2189                 SELECT id FROM runs WHERE id = ?1
2190                 UNION ALL
2191                 SELECT r.id FROM runs r JOIN tree t ON r.parent_run_id = t.id
2192             )
2193             SELECT COALESCE(SUM(s.tokens), 0)
2194             FROM steps s JOIN tree ON s.run_id = tree.id",
2195            [root],
2196            |r| r.get(0),
2197        )?;
2198        Ok(n as u64)
2199    }
2200
2201    /// Number of agents (run rows) in the tree rooted at `root` — the durable
2202    /// agent count restored on a tree resume.
2203    pub fn agent_count_tree(&self, root: i64) -> Result<u32> {
2204        let n: i64 = self.conn.query_row(
2205            "WITH RECURSIVE tree(id) AS (
2206                 SELECT id FROM runs WHERE id = ?1
2207                 UNION ALL
2208                 SELECT r.id FROM runs r JOIN tree t ON r.parent_run_id = t.id
2209             )
2210             SELECT COUNT(*) FROM tree",
2211            [root],
2212            |r| r.get(0),
2213        )?;
2214        Ok(n as u32)
2215    }
2216
2217    /// Persist a spawned child's contract so a crashed tree can rebuild and
2218    /// resume that exact child on resume instead of spawning a duplicate. Keyed
2219    /// by (parent, step, goal) so a replayed spawn step adopts the existing child.
2220    #[allow(clippy::too_many_arguments)]
2221    pub fn record_spawn(
2222        &self,
2223        parent_run_id: i64,
2224        step: u32,
2225        child_run_id: i64,
2226        goal: &str,
2227        verify_file: &str,
2228        needle: &str,
2229        max_steps: Option<u32>,
2230        deny_write_json: &str,
2231    ) -> Result<()> {
2232        self.conn.execute(
2233            "INSERT INTO spawns
2234                 (parent_run_id, step, child_run_id, goal, verify_file, needle, max_steps, deny_write)
2235             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
2236            (
2237                parent_run_id,
2238                step,
2239                child_run_id,
2240                goal,
2241                verify_file,
2242                needle,
2243                max_steps,
2244                deny_write_json,
2245            ),
2246        )?;
2247        Ok(())
2248    }
2249
2250    /// Find the child spawned by `parent_run_id` at `step` for `goal`, if any —
2251    /// the adopt-on-resume lookup that makes a replayed spawn step idempotent.
2252    pub fn find_spawn(
2253        &self,
2254        parent_run_id: i64,
2255        step: u32,
2256        goal: &str,
2257    ) -> Result<Option<SpawnRow>> {
2258        Ok(self
2259            .conn
2260            .query_row(
2261                "SELECT child_run_id, goal, verify_file, needle, max_steps, deny_write
2262                 FROM spawns WHERE parent_run_id = ?1 AND step = ?2 AND goal = ?3
2263                 ORDER BY id ASC LIMIT 1",
2264                (parent_run_id, step, goal),
2265                |r| {
2266                    Ok(SpawnRow {
2267                        child_run_id: r.get(0)?,
2268                        goal: r.get(1)?,
2269                        verify_file: r.get(2)?,
2270                        needle: r.get(3)?,
2271                        max_steps: r.get::<_, Option<i64>>(4)?.map(|n| n as u32),
2272                        deny_write: r.get(5)?,
2273                    })
2274                },
2275            )
2276            .ok())
2277    }
2278
2279    /// Check a run can be resumed from its checkpoint, or return a typed
2280    /// [`Error::Resume`]. Refuses a store written by a newer checkpoint format
2281    /// (rather than misreading a layout it does not understand) and a run id that
2282    /// does not exist. An already-`completed` run is resumable as a no-op, so it
2283    /// is not refused here.
2284    pub fn check_resumable(&self, run_id: i64) -> Result<()> {
2285        let format: i64 = self
2286            .conn
2287            .query_row("PRAGMA user_version", [], |r| r.get(0))?;
2288        if format > CHECKPOINT_FORMAT {
2289            return Err(Error::Resume {
2290                reason: format!(
2291                    "checkpoint format {format} is newer than supported {CHECKPOINT_FORMAT}; \
2292                     upgrade io-harness to resume this run"
2293                ),
2294            });
2295        }
2296        let exists: bool = self
2297            .conn
2298            .query_row("SELECT 1 FROM runs WHERE id = ?1", [run_id], |_| Ok(true))
2299            .unwrap_or(false);
2300        if !exists {
2301            return Err(Error::Resume {
2302                reason: format!("no run with id {run_id} in the store"),
2303            });
2304        }
2305        Ok(())
2306    }
2307
2308    /// Record which provider ran this run, for the audit trace.
2309    pub fn set_provider(&self, run_id: i64, provider: &str) -> Result<()> {
2310        self.conn.execute(
2311            "UPDATE runs SET provider = ?1 WHERE id = ?2",
2312            (provider, run_id),
2313        )?;
2314        Ok(())
2315    }
2316
2317    /// The provider recorded for a run, if any.
2318    pub fn provider(&self, run_id: i64) -> Result<Option<String>> {
2319        Ok(self
2320            .conn
2321            .query_row("SELECT provider FROM runs WHERE id = ?1", [run_id], |r| {
2322                r.get(0)
2323            })?)
2324    }
2325
2326    /// Record the run's final outcome, and derive the durable status from it:
2327    /// `success` completes the run, `awaiting_approval` pauses it, any other
2328    /// terminal outcome completes it (finished, just not with success). A run
2329    /// that crashed mid-loop never reaches here, so it stays `running` and is
2330    /// resumable.
2331    pub fn finish_run(&self, run_id: i64, outcome: &str) -> Result<()> {
2332        let status = match outcome {
2333            "awaiting_approval" => "paused",
2334            _ => "completed",
2335        };
2336        self.conn.execute(
2337            "UPDATE runs SET outcome = ?1, status = ?2 WHERE id = ?3",
2338            (outcome, status, run_id),
2339        )?;
2340        // A paused run has not finished — it is waiting for a human and will be
2341        // resumed — so it gets no summary yet. It gets one when it really ends.
2342        if status == "completed" {
2343            self.write_summary(run_id, outcome)?;
2344        }
2345        Ok(())
2346    }
2347
2348    /// Record the run's outcome summary. Called by [`Self::finish_run`].
2349    ///
2350    /// Written here rather than assembled by the caller because a run that
2351    /// escalates or is refused returns `Err` and never reaches a
2352    /// [`RunResult`](crate::RunResult) at all — so a summary built at the call
2353    /// site would be missing for exactly the endings a scoring tool most wants to
2354    /// count.
2355    fn write_summary(&self, run_id: i64, outcome: &str) -> Result<()> {
2356        // Both stamps from the database clock, like `started_at`. Mixing SQLite's
2357        // clock with the process's would make the difference meaningless.
2358        let (finished_at, duration_ms): (String, Option<f64>) = self.conn.query_row(
2359            "SELECT strftime('%Y-%m-%dT%H:%M:%fZ','now'),
2360                    (julianday('now') - julianday(started_at)) * 86400000.0
2361             FROM runs WHERE id = ?1",
2362            [run_id],
2363            |r| Ok((r.get(0)?, r.get(1)?)),
2364        )?;
2365        let duration_ms = duration_ms.map(|ms| ms.max(0.0) as u64);
2366        // `INSERT OR REPLACE`, because `finish_run` is reachable more than once for
2367        // one run: a paused run resumes and finishes, and a resume of an already
2368        // finished run is documented as idempotent. The last ending is the true one.
2369        self.conn.execute(
2370            "INSERT OR REPLACE INTO run_outcomes
2371                 (run_id, outcome, success, steps, tokens, duration_ms, finished_at)
2372             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
2373            (
2374                run_id,
2375                outcome,
2376                i64::from(outcome == SUCCESS_OUTCOME),
2377                self.last_step(run_id)?,
2378                self.spent_tokens(run_id)?,
2379                duration_ms,
2380                &finished_at,
2381            ),
2382        )?;
2383        Ok(())
2384    }
2385
2386    /// What a finished run cost and whether it worked.
2387    ///
2388    /// `None` if the run has not finished, is paused awaiting a human, or was
2389    /// finished by a pre-0.12.0 binary — a missing summary is reported as absent
2390    /// rather than as a row of zeroes, which would be indistinguishable from a run
2391    /// that did nothing.
2392    pub fn run_summary(&self, run_id: i64) -> Result<Option<RunSummary>> {
2393        let mut q = self.conn.prepare(
2394            "SELECT run_id, outcome, success, steps, tokens, duration_ms, finished_at
2395             FROM run_outcomes WHERE run_id = ?1",
2396        )?;
2397        let mut rows = q.query_map([run_id], |r| {
2398            Ok(RunSummary {
2399                run_id: r.get(0)?,
2400                outcome: r.get(1)?,
2401                success: r.get::<_, i64>(2)? != 0,
2402                steps: r.get(3)?,
2403                tokens: r.get(4)?,
2404                duration_ms: r.get(5)?,
2405                finished_at: r.get(6)?,
2406            })
2407        })?;
2408        match rows.next() {
2409            Some(row) => Ok(Some(row?)),
2410            None => Ok(None),
2411        }
2412    }
2413
2414    /// Every run in this store, newest first.
2415    ///
2416    /// Exists because an escalation returns `Err` rather than a
2417    /// [`RunResult`](crate::RunResult), so a caller whose run escalated has no
2418    /// `run_id` to resume with and therefore no way to reach
2419    /// [`RunOutcome::Escalated`](crate::RunOutcome::Escalated) — the outcome added
2420    /// for exactly that case. A caller who did not record the id before starting
2421    /// can find it here.
2422    pub fn runs(&self) -> Result<Vec<i64>> {
2423        let mut stmt = self.conn.prepare("SELECT id FROM runs ORDER BY id DESC")?;
2424        let rows = stmt.query_map([], |r| r.get(0))?;
2425        Ok(rows.collect::<std::result::Result<_, _>>()?)
2426    }
2427
2428    /// The most recently started run, if this store holds one.
2429    ///
2430    /// A convenience over [`Store::runs`] for the common single-run case. With
2431    /// concurrent runs in one store, "most recent" is by insertion order and a
2432    /// caller that cares should track its own ids.
2433    pub fn last_run(&self) -> Result<Option<i64>> {
2434        Ok(self.runs()?.into_iter().next())
2435    }
2436
2437    /// The recorded final outcome string of a run, if it has finished.
2438    pub fn outcome(&self, run_id: i64) -> Result<Option<String>> {
2439        Ok(self
2440            .conn
2441            .query_row("SELECT outcome FROM runs WHERE id = ?1", [run_id], |r| {
2442                r.get(0)
2443            })
2444            .ok()
2445            .flatten())
2446    }
2447
2448    /// The durable run status as a typed [`RunStatus`], if the run exists.
2449    pub fn run_status(&self, run_id: i64) -> Result<Option<RunStatus>> {
2450        Ok(self.status(run_id)?.map(|s| RunStatus::from_str(&s)))
2451    }
2452
2453    /// The highest step number recorded for a run, or 0 if none — the resume
2454    /// point for [`crate::resume`].
2455    pub fn last_step(&self, run_id: i64) -> Result<u32> {
2456        let n: i64 = self.conn.query_row(
2457            "SELECT COALESCE(MAX(step), 0) FROM steps WHERE run_id = ?1",
2458            [run_id],
2459            |r| r.get(0),
2460        )?;
2461        Ok(n as u32)
2462    }
2463
2464    /// Read every step of a run back, in order, as the full trace.
2465    /// The run's trace reduced to the part that two identical runs must match,
2466    /// as diffable text.
2467    ///
2468    /// This is the crate's definition of "the same run twice", and it exists
2469    /// because equality could not be row identity: `steps` has no
2470    /// `UNIQUE(run_id, step)` and a retry inserts its own row under the step
2471    /// number the eventual commit will reuse, so counting or comparing rows
2472    /// compares trace entries rather than agent behaviour.
2473    ///
2474    /// # What is compared
2475    ///
2476    /// Every `steps` row — step number, decision, result, prompt, tool call and
2477    /// tokens — and every `context_events` row's step, kind and detail. Between
2478    /// them these are what the agent was shown, what it decided, what it did, and
2479    /// what that cost.
2480    ///
2481    /// # What is excluded, and why
2482    ///
2483    /// Everything whose value is a fact about *this* execution rather than about
2484    /// the run:
2485    ///
2486    /// - **Wall-clock stamps** — `runs.started_at`, `memory.created_at`,
2487    ///   `run_outcomes.finished_at` and `duration_ms`. Two runs of the same case
2488    ///   take different amounts of time; that is not a divergence.
2489    /// - **`mcp_events.millis`** — a measured duration, for the same reason.
2490    /// - **`sandbox_events.detail`** — it carries the argv, and the argv carries
2491    ///   an ephemeral tempdir path that is different every run by design.
2492    /// - **Run and child ids** — `AUTOINCREMENT` values, meaningful only within
2493    ///   one store.
2494    ///
2495    /// Excluding a field is a decision that this crate cannot promise it, not a
2496    /// convenience. Anything added to this list should be added to this doc with
2497    /// its reason, because a comparison that quietly excludes what it cannot
2498    /// match is a comparison that asserts nothing.
2499    ///
2500    /// # What it assumes
2501    ///
2502    /// That each run being compared has its **own fresh store**. Run ids are
2503    /// excluded from the text, but a child agent's run id is embedded in the
2504    /// parent's composed observation (`[child 5 "goal" -> …]`), which is real
2505    /// content the model was shown. In a fresh store those ids start at 1 and are
2506    /// allocated in spawn order, so they match; in a shared store the second run's
2507    /// ids are higher and the traces differ for a reason that has nothing to do
2508    /// with the agent.
2509    ///
2510    /// Deterministic replay also requires the provider to answer identically —
2511    /// see [`Replay`](crate::provider::Replay) — and the same workspace state to
2512    /// start from.
2513    pub fn canonical_trace(&self, run_id: i64) -> Result<String> {
2514        let mut out = String::new();
2515        for s in self.steps(run_id)? {
2516            out.push_str(&format!(
2517                "step {} | tokens {} | decision {} | tool_call {} | prompt {} | result {}\n",
2518                s.step, s.tokens, s.decision, s.tool_call, s.prompt, s.result
2519            ));
2520        }
2521        for e in self.context_events(run_id)? {
2522            out.push_str(&format!(
2523                "context {} | {} | {}\n",
2524                e.step,
2525                e.kind,
2526                e.detail.as_deref().unwrap_or("")
2527            ));
2528        }
2529        Ok(out)
2530    }
2531
2532    pub fn steps(&self, run_id: i64) -> Result<Vec<StepRecord>> {
2533        let mut stmt = self.conn.prepare(
2534            "SELECT step, decision, result, prompt, tool_call, tokens
2535             FROM steps WHERE run_id = ?1 ORDER BY step ASC, id ASC",
2536        )?;
2537        let rows = stmt.query_map([run_id], |r| {
2538            Ok(StepRecord {
2539                step: r.get::<_, i64>(0)? as u32,
2540                decision: r.get(1)?,
2541                result: r.get(2)?,
2542                prompt: r.get(3)?,
2543                tool_call: r.get(4)?,
2544                tokens: r.get::<_, i64>(5)? as u64,
2545            })
2546        })?;
2547        Ok(rows.collect::<std::result::Result<_, _>>()?)
2548    }
2549
2550    // ---- 0.10.0: durable cross-run memory ----
2551
2552    /// Write or replace `key` for `workspace`, attributed to the run and step
2553    /// that wrote it. A value past [`MEMORY_MAX_ENTRY_CHARS`] is truncated with
2554    /// a visible marker rather than refused. Returns the keys evicted to stay
2555    /// inside the caps, oldest first — the caller records the eviction in the
2556    /// trace; this never writes a trace row itself.
2557    pub fn memory_put(
2558        &self,
2559        workspace: &str,
2560        key: &str,
2561        value: &str,
2562        run_id: i64,
2563        step: u32,
2564    ) -> Result<Vec<String>> {
2565        let value = truncate_memory_value(value);
2566        // An overwrite re-attributes the entry and refreshes `created_at`, so
2567        // recency ordering reflects the latest write rather than the first.
2568        self.conn.execute(
2569            "INSERT INTO memory (workspace, key, value, run_id, step, created_at)
2570             VALUES (?1, ?2, ?3, ?4, ?5, strftime('%Y-%m-%dT%H:%M:%fZ','now'))
2571             ON CONFLICT(workspace, key) DO UPDATE SET
2572                 value      = excluded.value,
2573                 run_id     = excluded.run_id,
2574                 step       = excluded.step,
2575                 created_at = excluded.created_at",
2576            (workspace, key, &value, run_id, step),
2577        )?;
2578        self.enforce_memory_caps(workspace, key)
2579    }
2580
2581    /// Evict this workspace's oldest entries until both caps hold, never the
2582    /// entry `keep` (the one just written — evicting it would make a write a
2583    /// silent no-op). Returns the evicted keys in eviction order.
2584    fn enforce_memory_caps(&self, workspace: &str, keep: &str) -> Result<Vec<String>> {
2585        // LENGTH() on TEXT counts characters, not bytes — the cap is in chars.
2586        let rows: Vec<(String, i64)> = {
2587            let mut stmt = self.conn.prepare(
2588                "SELECT key, LENGTH(value) FROM memory WHERE workspace = ?1
2589                 ORDER BY created_at ASC, id ASC",
2590            )?;
2591            let rows = stmt.query_map([workspace], |r| Ok((r.get(0)?, r.get(1)?)))?;
2592            rows.collect::<std::result::Result<_, _>>()?
2593        };
2594
2595        let mut count = rows.len();
2596        let mut chars: i64 = rows.iter().map(|(_, n)| *n).sum();
2597        let mut evicted = Vec::new();
2598        for (key, n) in &rows {
2599            if count <= MEMORY_MAX_ENTRIES && chars <= MEMORY_MAX_CHARS as i64 {
2600                break;
2601            }
2602            if key == keep {
2603                continue;
2604            }
2605            self.conn.execute(
2606                "DELETE FROM memory WHERE workspace = ?1 AND key = ?2",
2607                (workspace, key),
2608            )?;
2609            count -= 1;
2610            chars -= n;
2611            evicted.push(key.clone());
2612        }
2613        Ok(evicted)
2614    }
2615
2616    /// Every entry for `workspace`, oldest first. Never another workspace's.
2617    pub fn memory_list(&self, workspace: &str) -> Result<Vec<MemoryEntry>> {
2618        let mut stmt = self.conn.prepare(
2619            "SELECT key, value, run_id, step, created_at FROM memory
2620             WHERE workspace = ?1 ORDER BY created_at ASC, id ASC",
2621        )?;
2622        let rows = stmt.query_map([workspace], |r| {
2623            Ok(MemoryEntry {
2624                key: r.get(0)?,
2625                value: r.get(1)?,
2626                run_id: r.get(2)?,
2627                step: r.get::<_, i64>(3)? as u32,
2628                created_at: r.get(4)?,
2629            })
2630        })?;
2631        Ok(rows.collect::<std::result::Result<_, _>>()?)
2632    }
2633
2634    /// One entry of `workspace` by key, if it holds one.
2635    pub fn memory_get(&self, workspace: &str, key: &str) -> Result<Option<MemoryEntry>> {
2636        Ok(self
2637            .conn
2638            .query_row(
2639                "SELECT key, value, run_id, step, created_at FROM memory
2640                 WHERE workspace = ?1 AND key = ?2",
2641                (workspace, key),
2642                |r| {
2643                    Ok(MemoryEntry {
2644                        key: r.get(0)?,
2645                        value: r.get(1)?,
2646                        run_id: r.get(2)?,
2647                        step: r.get::<_, i64>(3)? as u32,
2648                        created_at: r.get(4)?,
2649                    })
2650                },
2651            )
2652            .ok())
2653    }
2654
2655    /// Forget one entry of `workspace`. True when an entry was removed.
2656    pub fn memory_delete(&self, workspace: &str, key: &str) -> Result<bool> {
2657        let n = self.conn.execute(
2658            "DELETE FROM memory WHERE workspace = ?1 AND key = ?2",
2659            (workspace, key),
2660        )?;
2661        Ok(n > 0)
2662    }
2663
2664    /// Removes every entry for `workspace`; returns how many. Other workspaces
2665    /// keep theirs.
2666    pub fn memory_clear(&self, workspace: &str) -> Result<usize> {
2667        Ok(self
2668            .conn
2669            .execute("DELETE FROM memory WHERE workspace = ?1", [workspace])?)
2670    }
2671}
2672
2673#[cfg(test)]
2674mod tests {
2675    use super::*;
2676
2677    #[test]
2678    fn refusals_record_action_target_rule_and_layer() {
2679        let store = Store::memory().unwrap();
2680        let run = store.start_run("goal", "root").unwrap();
2681        store
2682            .record_event(
2683                run,
2684                &PolicyEvent::refusal(2, "write", "secrets/key.txt").with_rule("secrets/*", "base"),
2685            )
2686            .unwrap();
2687
2688        let events = store.events(run).unwrap();
2689        assert_eq!(events.len(), 1);
2690        let e = &events[0];
2691        assert_eq!(e.kind, "refusal");
2692        assert_eq!(e.act, "write");
2693        assert_eq!(e.target, "secrets/key.txt");
2694        assert_eq!(e.rule.as_deref(), Some("secrets/*"));
2695        // Attributable to the layer that refused, so a base-layer deny is findable.
2696        assert_eq!(e.layer.as_deref(), Some("base"));
2697    }
2698
2699    #[test]
2700    fn decisions_record_their_value_source_and_any_altered_target() {
2701        let store = Store::memory().unwrap();
2702        let run = store.start_run("goal", "root").unwrap();
2703        store
2704            .record_event(
2705                run,
2706                &PolicyEvent::decision(1, "write", "src/a.rs", "approve", "stdin")
2707                    .with_performed("src/sandbox/a.rs"),
2708            )
2709            .unwrap();
2710        store
2711            .record_event(
2712                run,
2713                &PolicyEvent::decision(2, "write", "src/b.rs", "approve", "remembered"),
2714            )
2715            .unwrap();
2716
2717        let events = store.events(run).unwrap();
2718        assert_eq!(events.len(), 2);
2719        // Requested and performed forms are distinguishable.
2720        assert_eq!(events[0].decision.as_deref(), Some("approve"));
2721        assert_eq!(events[0].target, "src/a.rs");
2722        assert_eq!(events[0].performed.as_deref(), Some("src/sandbox/a.rs"));
2723        // An auto-approval by a remembered rule is not confusable with a fresh one.
2724        assert_eq!(events[1].source.as_deref(), Some("remembered"));
2725        assert_eq!(events[1].performed, None);
2726    }
2727
2728    #[test]
2729    fn a_pre_0_4_database_migrates_in_place_and_keeps_its_rows() {
2730        let dir = tempfile::tempdir().unwrap();
2731        let path = dir.path().join("runs.db");
2732
2733        // A 0.3.0-shaped database: runs + steps only, no policy tables.
2734        {
2735            let conn = rusqlite::Connection::open(&path).unwrap();
2736            conn.execute_batch(
2737                "CREATE TABLE runs (id INTEGER PRIMARY KEY AUTOINCREMENT, goal TEXT NOT NULL,
2738                     file TEXT NOT NULL, outcome TEXT, provider TEXT);
2739                 CREATE TABLE steps (id INTEGER PRIMARY KEY AUTOINCREMENT, run_id INTEGER NOT NULL,
2740                     step INTEGER NOT NULL, decision TEXT NOT NULL, result TEXT NOT NULL,
2741                     prompt TEXT NOT NULL DEFAULT '', tool_call TEXT NOT NULL DEFAULT '',
2742                     tokens INTEGER NOT NULL DEFAULT 0);
2743                 INSERT INTO runs (goal, file) VALUES ('old goal', 'old.txt');",
2744            )
2745            .unwrap();
2746        }
2747
2748        let store = Store::open(&path).unwrap();
2749        // The pre-existing row survives; the new tables are usable.
2750        assert_eq!(store.last_step(1).unwrap(), 0);
2751        store
2752            .record_event(1, &PolicyEvent::refusal(1, "read", ".env"))
2753            .unwrap();
2754        assert_eq!(store.events(1).unwrap().len(), 1);
2755    }
2756
2757    #[test]
2758    fn a_pending_approval_survives_the_store_being_reopened() {
2759        let dir = tempfile::tempdir().unwrap();
2760        let path = dir.path().join("runs.db");
2761
2762        let request_id = {
2763            let store = Store::open(&path).unwrap();
2764            let run = store.start_run("goal", "root").unwrap();
2765            store
2766                .put_pending(run, 3, "write", "src/a.rs", Some("fn a() {}"))
2767                .unwrap()
2768        };
2769
2770        // A different Store over the same file — the process that created it is gone.
2771        let store = Store::open(&path).unwrap();
2772        let p = store.pending(request_id).unwrap().expect("still pending");
2773        assert_eq!(p.step, 3);
2774        assert_eq!(p.act, "write");
2775        assert_eq!(p.target, "src/a.rs");
2776        assert_eq!(p.content.as_deref(), Some("fn a() {}"));
2777        assert_eq!(p.resolved, None);
2778
2779        store.resolve_pending(request_id, "approve").unwrap();
2780        let p = store.pending(request_id).unwrap().unwrap();
2781        assert_eq!(p.resolved.as_deref(), Some("approve"));
2782    }
2783
2784    #[test]
2785    fn the_tree_is_reconstructable_from_a_reopened_store() {
2786        let dir = tempfile::tempdir().unwrap();
2787        let path = dir.path().join("runs.db");
2788
2789        // A parent spawns two children (one nests a grandchild) and the tree
2790        // draws against its ceiling, then everything is dropped.
2791        let (root, c1, c2, gc) = {
2792            let store = Store::open(&path).unwrap();
2793            let root = store.start_run("root goal", "ws").unwrap();
2794            let c1 = store.start_child_run("child 1", "ws", root, 1).unwrap();
2795            let c2 = store.start_child_run("child 2", "ws", root, 1).unwrap();
2796            let gc = store.start_child_run("grandchild", "ws", c1, 2).unwrap();
2797            store
2798                .record_agent_event(&AgentEvent::spawn(root, 1, c1, "child 1"))
2799                .unwrap();
2800            store
2801                .record_agent_event(&AgentEvent::spawn(root, 1, c2, "child 2"))
2802                .unwrap();
2803            store
2804                .record_agent_event(&AgentEvent::spawn(c1, 1, gc, "grandchild"))
2805                .unwrap();
2806            store
2807                .record_agent_event(&AgentEvent::spawn_refused(root, 2, "agents"))
2808                .unwrap();
2809            store
2810                .record_agent_event(&AgentEvent::budget_draw(c1, 1, 30, 70))
2811                .unwrap();
2812            (root, c1, c2, gc)
2813        };
2814
2815        // A fresh Store over the same file — the process that built the tree is gone.
2816        let store = Store::open(&path).unwrap();
2817        // The parent/child edges rebuild the graph.
2818        assert_eq!(store.children(root).unwrap(), vec![c1, c2]);
2819        assert_eq!(store.children(c1).unwrap(), vec![gc]);
2820        assert_eq!(store.parent(gc).unwrap(), Some(c1));
2821        assert_eq!(store.parent(root).unwrap(), None);
2822        assert_eq!(store.depth(gc).unwrap(), 2);
2823
2824        // Spawns, the refusal, and the draw are all recorded.
2825        let root_events = store.agent_events(root).unwrap();
2826        assert_eq!(root_events.iter().filter(|e| e.kind == "spawn").count(), 2);
2827        assert_eq!(
2828            root_events
2829                .iter()
2830                .filter(|e| e.kind == "spawn_refused")
2831                .count(),
2832            1
2833        );
2834        let draws = store.agent_events(c1).unwrap();
2835        let draw = draws.iter().find(|e| e.kind == "budget_draw").unwrap();
2836        assert_eq!(draw.tokens, Some(30));
2837        assert_eq!(draw.remaining, Some(70));
2838    }
2839
2840    #[test]
2841    fn a_pre_0_5_database_migrates_and_keeps_its_rows() {
2842        let dir = tempfile::tempdir().unwrap();
2843        let path = dir.path().join("runs.db");
2844
2845        // A 0.4.0-shaped database: runs (no parent_run_id/depth), steps, and the
2846        // policy tables, with a row.
2847        {
2848            let conn = rusqlite::Connection::open(&path).unwrap();
2849            conn.execute_batch(
2850                "CREATE TABLE runs (id INTEGER PRIMARY KEY AUTOINCREMENT, goal TEXT NOT NULL,
2851                     file TEXT NOT NULL, outcome TEXT, provider TEXT);
2852                 CREATE TABLE steps (id INTEGER PRIMARY KEY AUTOINCREMENT, run_id INTEGER NOT NULL,
2853                     step INTEGER NOT NULL, decision TEXT NOT NULL, result TEXT NOT NULL,
2854                     prompt TEXT NOT NULL DEFAULT '', tool_call TEXT NOT NULL DEFAULT '',
2855                     tokens INTEGER NOT NULL DEFAULT 0);
2856                 INSERT INTO runs (goal, file) VALUES ('old', 'old.txt');",
2857            )
2858            .unwrap();
2859        }
2860
2861        let store = Store::open(&path).unwrap();
2862        // The pre-existing row survives and reads as a root at depth 0.
2863        assert_eq!(store.parent(1).unwrap(), None);
2864        assert_eq!(store.depth(1).unwrap(), 0);
2865        // The new table is usable.
2866        let child = store.start_child_run("c", "ws", 1, 1).unwrap();
2867        assert_eq!(store.children(1).unwrap(), vec![child]);
2868    }
2869
2870    #[test]
2871    fn a_pre_0_8_database_migrates_in_place_and_keeps_its_rows() {
2872        let dir = tempfile::tempdir().unwrap();
2873        let path = dir.path().join("runs.db");
2874
2875        // A 0.7.0-shaped database: everything through checkpoints, and no
2876        // mcp_events table.
2877        {
2878            let store = Store::open(&path).unwrap();
2879            let run = store.start_run("old goal", "old.txt").unwrap();
2880            store
2881                .checkpoint_step(run, &StepRecord::new(1, "wrote", "ok"))
2882                .unwrap();
2883            store
2884                .record_event(run, &PolicyEvent::refusal(1, "write", "secrets/k"))
2885                .unwrap();
2886            store
2887                .conn
2888                .execute("DROP TABLE IF EXISTS mcp_events", [])
2889                .unwrap();
2890        }
2891
2892        // Reopening migrates it: the old rows are intact and the new table works.
2893        let store = Store::open(&path).unwrap();
2894        assert_eq!(store.last_step(1).unwrap(), 1);
2895        assert_eq!(store.events(1).unwrap().len(), 1);
2896        assert!(store.mcp_events(1).unwrap().is_empty());
2897        store
2898            .record_mcp(1, &McpEvent::connected("files", "stdio"))
2899            .unwrap();
2900        let events = store.mcp_events(1).unwrap();
2901        assert_eq!(events.len(), 1);
2902        assert_eq!(events[0].detail.as_deref(), Some("stdio"));
2903
2904        // And a 0.7.0 binary, which never queries mcp_events, still reads it —
2905        // nothing it knows about was altered or rewritten.
2906        assert_eq!(store.steps(1).unwrap().len(), 1);
2907        assert_eq!(store.run_status(1).unwrap(), Some(RunStatus::Running));
2908    }
2909
2910    #[test]
2911    fn full_trace_persists_and_reads_back() {
2912        let store = Store::memory().unwrap();
2913        let run = store.start_run("goal", "out.txt").unwrap();
2914        store
2915            .record(
2916                run,
2917                &StepRecord::new(1, "wrote file", "content v1").with_trace(
2918                    "the prompt",
2919                    r#"{"content":"content v1"}"#,
2920                    128,
2921                ),
2922            )
2923            .unwrap();
2924        store
2925            .record(run, &StepRecord::new(2, "verified", "ok"))
2926            .unwrap();
2927        store.finish_run(run, "success").unwrap();
2928
2929        let steps = store.steps(run).unwrap();
2930        assert_eq!(steps.len(), 2);
2931        assert_eq!(steps[0].decision, "wrote file");
2932        assert_eq!(steps[0].prompt, "the prompt");
2933        assert_eq!(steps[0].tokens, 128);
2934        assert_eq!(steps[1].result, "ok");
2935        assert_eq!(store.last_step(run).unwrap(), 2);
2936    }
2937
2938    #[test]
2939    fn migrates_a_0_1_0_steps_table_in_place() {
2940        // A 0.1.0 database: `steps` without the trace columns, with a row.
2941        let conn = Connection::open_in_memory().unwrap();
2942        conn.execute_batch(
2943            "CREATE TABLE runs (id INTEGER PRIMARY KEY AUTOINCREMENT, goal TEXT NOT NULL, file TEXT NOT NULL, outcome TEXT);
2944             CREATE TABLE steps (id INTEGER PRIMARY KEY AUTOINCREMENT, run_id INTEGER NOT NULL, step INTEGER NOT NULL, decision TEXT NOT NULL, result TEXT NOT NULL);
2945             INSERT INTO runs (goal, file) VALUES ('g', 'f');
2946             INSERT INTO steps (run_id, step, decision, result) VALUES (1, 1, 'wrote file', 'old');",
2947        )
2948        .unwrap();
2949
2950        // Opening through Store migrates it; the old row survives with defaults.
2951        let store = Store::from_conn(conn).unwrap();
2952        let steps = store.steps(1).unwrap();
2953        assert_eq!(steps.len(), 1);
2954        assert_eq!(steps[0].result, "old");
2955        assert_eq!(steps[0].prompt, "");
2956        assert_eq!(steps[0].tokens, 0);
2957    }
2958
2959    #[test]
2960    fn provider_is_recorded_and_read_back() {
2961        let store = Store::memory().unwrap();
2962        let run = store.start_run("g", "f").unwrap();
2963        assert_eq!(store.provider(run).unwrap(), None);
2964        store.set_provider(run, "anthropic").unwrap();
2965        assert_eq!(store.provider(run).unwrap().as_deref(), Some("anthropic"));
2966    }
2967
2968    #[test]
2969    fn migrates_a_pre_0_3_runs_table_adding_provider() {
2970        // A 0.1/0.2 database: `runs` without the provider column.
2971        let conn = Connection::open_in_memory().unwrap();
2972        conn.execute_batch(
2973            "CREATE TABLE runs (id INTEGER PRIMARY KEY AUTOINCREMENT, goal TEXT NOT NULL, file TEXT NOT NULL, outcome TEXT);
2974             CREATE TABLE steps (id INTEGER PRIMARY KEY AUTOINCREMENT, run_id INTEGER NOT NULL, step INTEGER NOT NULL, decision TEXT NOT NULL, result TEXT NOT NULL);
2975             INSERT INTO runs (goal, file) VALUES ('g', 'f');",
2976        )
2977        .unwrap();
2978
2979        // Opening through Store adds the provider column; the old row survives.
2980        let store = Store::from_conn(conn).unwrap();
2981        assert_eq!(store.provider(1).unwrap(), None);
2982        store.set_provider(1, "openai").unwrap();
2983        assert_eq!(store.provider(1).unwrap().as_deref(), Some("openai"));
2984    }
2985
2986    // ---- 0.7.0: durable checkpoint + resume ----
2987
2988    #[test]
2989    fn checkpoint_step_commits_the_step_and_its_event_together() {
2990        let store = Store::memory().unwrap();
2991        let run = store.start_run("goal", "root").unwrap();
2992        store
2993            .checkpoint_step(run, &StepRecord::new(1, "act", "ok"))
2994            .unwrap();
2995        store
2996            .checkpoint_step(run, &StepRecord::new(2, "act", "ok"))
2997            .unwrap();
2998
2999        assert_eq!(store.last_step(run).unwrap(), 2);
3000        assert_eq!(store.steps(run).unwrap().len(), 2);
3001        let cps: Vec<_> = store
3002            .checkpoint_events(run)
3003            .unwrap()
3004            .into_iter()
3005            .filter(|e| e.kind == "checkpoint")
3006            .collect();
3007        assert_eq!(cps.len(), 2);
3008        // NF4: a checkpoint event carries no file content — only step metadata.
3009        assert!(cps.iter().all(|e| e.detail.is_none()));
3010    }
3011
3012    #[test]
3013    fn a_rolled_back_step_leaves_the_prior_checkpoint_intact() {
3014        // The committed checkpoint is the completion marker: a step whose
3015        // transaction never commits (a crash mid-commit) vanishes entirely and
3016        // the prior checkpoint stands — never a torn half recorded as done.
3017        let store = Store::memory().unwrap();
3018        let run = store.start_run("goal", "root").unwrap();
3019        store
3020            .checkpoint_step(run, &StepRecord::new(1, "act", "ok"))
3021            .unwrap();
3022
3023        // Simulate a crash mid-commit: open the step's transaction, write both
3024        // rows, then drop without committing (as a killed process would).
3025        {
3026            let tx = store.conn.unchecked_transaction().unwrap();
3027            tx.execute(
3028                "INSERT INTO steps (run_id, step, decision, result) VALUES (?1, 2, 'act', 'ok')",
3029                [run],
3030            )
3031            .unwrap();
3032            tx.execute(
3033                "INSERT INTO checkpoint_events (run_id, step, kind) VALUES (?1, 2, 'checkpoint')",
3034                [run],
3035            )
3036            .unwrap();
3037            // no tx.commit() — dropped here, rolling back.
3038        }
3039
3040        assert_eq!(
3041            store.last_step(run).unwrap(),
3042            1,
3043            "the torn step must not survive"
3044        );
3045        assert_eq!(store.steps(run).unwrap().len(), 1);
3046    }
3047
3048    #[test]
3049    fn check_resumable_refuses_a_newer_format_and_a_missing_run() {
3050        let store = Store::memory().unwrap();
3051        let run = store.start_run("goal", "root").unwrap();
3052        assert!(store.check_resumable(run).is_ok());
3053
3054        // A run id that does not exist is a typed Resume error, not a panic.
3055        assert!(matches!(
3056            store.check_resumable(9999),
3057            Err(Error::Resume { .. })
3058        ));
3059
3060        // A store written by a newer checkpoint format is refused rather than
3061        // misread.
3062        store
3063            .conn
3064            .execute_batch(&format!("PRAGMA user_version = {}", CHECKPOINT_FORMAT + 1))
3065            .unwrap();
3066        assert!(matches!(
3067            store.check_resumable(run),
3068            Err(Error::Resume { .. })
3069        ));
3070    }
3071
3072    #[test]
3073    fn spent_tokens_and_elapsed_are_durable_reads() {
3074        let store = Store::memory().unwrap();
3075        let run = store.start_run("goal", "root").unwrap();
3076        store
3077            .checkpoint_step(run, &StepRecord::new(1, "a", "ok").with_trace("p", "t", 30))
3078            .unwrap();
3079        store
3080            .checkpoint_step(run, &StepRecord::new(2, "a", "ok").with_trace("p", "t", 12))
3081            .unwrap();
3082        assert_eq!(store.spent_tokens(run).unwrap(), 42);
3083        assert!(store.elapsed_secs(run).unwrap() >= 0.0);
3084    }
3085
3086    #[test]
3087    fn tree_aggregate_reads_span_root_and_descendants() {
3088        let store = Store::memory().unwrap();
3089        let root = store.start_run("goal", "root").unwrap();
3090        let child = store.start_child_run("sub", "root", root, 1).unwrap();
3091        let grandchild = store.start_child_run("subsub", "root", child, 2).unwrap();
3092        store
3093            .checkpoint_step(
3094                root,
3095                &StepRecord::new(1, "a", "ok").with_trace("p", "t", 10),
3096            )
3097            .unwrap();
3098        store
3099            .checkpoint_step(
3100                child,
3101                &StepRecord::new(1, "a", "ok").with_trace("p", "t", 20),
3102            )
3103            .unwrap();
3104        store
3105            .checkpoint_step(
3106                grandchild,
3107                &StepRecord::new(1, "a", "ok").with_trace("p", "t", 5),
3108            )
3109            .unwrap();
3110
3111        assert_eq!(
3112            store.tree_run_ids(root).unwrap(),
3113            vec![root, child, grandchild]
3114        );
3115        assert_eq!(store.spent_tokens_tree(root).unwrap(), 35);
3116        assert_eq!(store.agent_count_tree(root).unwrap(), 3);
3117    }
3118
3119    #[test]
3120    fn status_round_trips_and_a_pre_0_7_database_migrates() {
3121        // A 0.6.0-shaped database: runs without status/started_at.
3122        let conn = Connection::open_in_memory().unwrap();
3123        conn.execute_batch(
3124            "CREATE TABLE runs (id INTEGER PRIMARY KEY AUTOINCREMENT, goal TEXT NOT NULL, file TEXT NOT NULL, outcome TEXT, provider TEXT, parent_run_id INTEGER, depth INTEGER NOT NULL DEFAULT 0);
3125             INSERT INTO runs (goal, file) VALUES ('g', 'f');",
3126        )
3127        .unwrap();
3128        let store = Store::from_conn(conn).unwrap();
3129        // The old row gains a default status and no start stamp.
3130        assert_eq!(store.status(1).unwrap().as_deref(), Some("running"));
3131        store.set_status(1, "completed").unwrap();
3132        assert_eq!(store.status(1).unwrap().as_deref(), Some("completed"));
3133    }
3134
3135    // ---- 0.10.0: durable cross-run memory ----
3136
3137    #[test]
3138    fn the_entry_count_cap_evicts_oldest_first_and_never_the_new_entry() {
3139        let store = Store::memory().unwrap();
3140        for i in 0..MEMORY_MAX_ENTRIES {
3141            let evicted = store.memory_put("ws", &format!("k{i}"), "v", 1, 1).unwrap();
3142            assert!(evicted.is_empty(), "no eviction while under the cap");
3143        }
3144        assert_eq!(store.memory_list("ws").unwrap().len(), MEMORY_MAX_ENTRIES);
3145
3146        // Three more writes cost exactly the three oldest keys, in order.
3147        let mut evicted = Vec::new();
3148        for i in 0..3 {
3149            evicted.extend(
3150                store
3151                    .memory_put("ws", &format!("new{i}"), "v", 2, 2)
3152                    .unwrap(),
3153            );
3154        }
3155        assert_eq!(evicted, vec!["k0", "k1", "k2"]);
3156
3157        let keys: Vec<String> = store
3158            .memory_list("ws")
3159            .unwrap()
3160            .into_iter()
3161            .map(|e| e.key)
3162            .collect();
3163        assert_eq!(
3164            keys.len(),
3165            MEMORY_MAX_ENTRIES,
3166            "the cap holds after eviction"
3167        );
3168        assert!(!keys.contains(&"k0".to_string()));
3169        // The entry just written is never the one evicted to make room for it.
3170        for i in 0..3 {
3171            assert!(keys.contains(&format!("new{i}")));
3172        }
3173    }
3174
3175    #[test]
3176    fn the_total_chars_cap_evicts_before_the_count_cap_is_reached() {
3177        let store = Store::memory().unwrap();
3178        let big = "x".repeat(MEMORY_MAX_ENTRY_CHARS);
3179        let mut evicted = Vec::new();
3180        // 10 entries of 2_000 chars = 20_000, past the 16_000 char cap while the
3181        // 64-entry cap is nowhere near.
3182        for i in 0..10 {
3183            evicted.extend(
3184                store
3185                    .memory_put("ws", &format!("k{i}"), &big, 1, 1)
3186                    .unwrap(),
3187            );
3188        }
3189        assert_eq!(
3190            evicted,
3191            vec!["k0", "k1"],
3192            "oldest first, count cap untouched"
3193        );
3194
3195        let entries = store.memory_list("ws").unwrap();
3196        assert!(entries.len() < MEMORY_MAX_ENTRIES);
3197        let total: usize = entries.iter().map(|e| e.value.chars().count()).sum();
3198        assert!(total <= MEMORY_MAX_CHARS, "{total} chars is over the cap");
3199    }
3200
3201    #[test]
3202    fn an_oversized_value_is_truncated_with_a_marker_not_rejected() {
3203        let store = Store::memory().unwrap();
3204        // Multibyte throughout, so a byte-wise cut would not be valid UTF-8.
3205        let huge = "é".repeat(MEMORY_MAX_ENTRY_CHARS * 2);
3206        assert!(store.memory_put("ws", "k", &huge, 1, 1).is_ok());
3207
3208        let stored = store.memory_get("ws", "k").unwrap().unwrap().value;
3209        assert_eq!(stored.chars().count(), MEMORY_MAX_ENTRY_CHARS);
3210        assert!(stored.ends_with(MEMORY_TRUNCATED), "the cut is visible");
3211        // Cut on a char boundary: every kept char is the whole 'é', never a half.
3212        let kept = MEMORY_MAX_ENTRY_CHARS - MEMORY_TRUNCATED.chars().count();
3213        assert!(stored.chars().take(kept).all(|c| c == 'é'));
3214    }
3215
3216    #[test]
3217    fn a_0_9_1_store_opens_unchanged_and_still_resumes() {
3218        let dir = tempfile::tempdir().unwrap();
3219        let path = dir.path().join("runs.db");
3220
3221        // A 0.9.1-shaped database: every table through mcp_events, rows in the
3222        // ones a resume reads, and no `memory` table.
3223        let before_format: i64 = {
3224            let store = Store::open(&path).unwrap();
3225            let run = store.start_run("old goal", "old.txt").unwrap();
3226            store
3227                .checkpoint_step(run, &StepRecord::new(1, "wrote", "ok"))
3228                .unwrap();
3229            store
3230                .record_event(run, &PolicyEvent::refusal(1, "write", "secrets/k"))
3231                .unwrap();
3232            store
3233                .put_pending(run, 1, "write", "src/a.rs", None)
3234                .unwrap();
3235            let child = store.start_child_run("sub", "ws", run, 1).unwrap();
3236            store
3237                .record_agent_event(&AgentEvent::spawn(run, 1, child, "sub"))
3238                .unwrap();
3239            store
3240                .record_sandbox_event(&SandboxEvent::create(run, 1, "proc"))
3241                .unwrap();
3242            store
3243                .record_spawn(run, 1, child, "sub", "out.txt", "ok", None, "[]")
3244                .unwrap();
3245            store
3246                .record_mcp(run, &McpEvent::connected("files", "stdio"))
3247                .unwrap();
3248            store.conn.execute("DROP TABLE memory", []).unwrap();
3249            store
3250                .conn
3251                .query_row("PRAGMA user_version", [], |r| r.get(0))
3252                .unwrap()
3253        };
3254
3255        // Reopening under 0.10.0 adds `memory` and touches nothing else.
3256        let store = Store::open(&path).unwrap();
3257        let after_format: i64 = store
3258            .conn
3259            .query_row("PRAGMA user_version", [], |r| r.get(0))
3260            .unwrap();
3261        assert_eq!(
3262            after_format, before_format,
3263            "the checkpoint format must not move — a 0.9.1 checkpoint still resumes"
3264        );
3265        assert_eq!(after_format, CHECKPOINT_FORMAT);
3266        // The 0.7.0 durability promise: the pre-existing run still resumes.
3267        assert!(store.check_resumable(1).is_ok());
3268
3269        // Every pre-existing table is intact, with its rows.
3270        assert_eq!(store.steps(1).unwrap().len(), 1);
3271        assert_eq!(store.last_step(1).unwrap(), 1);
3272        assert_eq!(store.events(1).unwrap().len(), 1);
3273        assert_eq!(store.pending(1).unwrap().unwrap().act, "write");
3274        assert_eq!(store.checkpoint_events(1).unwrap().len(), 1);
3275        assert_eq!(store.agent_events(1).unwrap().len(), 1);
3276        assert_eq!(store.sandbox_events(1).unwrap().len(), 1);
3277        assert_eq!(store.mcp_events(1).unwrap().len(), 1);
3278        assert_eq!(store.children(1).unwrap(), vec![2]);
3279        assert!(store.find_spawn(1, 1, "sub").is_ok());
3280        assert_eq!(store.run_status(1).unwrap(), Some(RunStatus::Running));
3281        // And the new table is there and usable.
3282        assert!(store.memory_list("ws").unwrap().is_empty());
3283        store.memory_put("ws", "k", "v", 1, 1).unwrap();
3284        assert_eq!(store.memory_get("ws", "k").unwrap().unwrap().value, "v");
3285    }
3286
3287    #[test]
3288    fn a_layered_policy_reads_back_exactly_as_it_was_recorded() {
3289        let store = Store::memory().unwrap();
3290        let run = store.start_run("goal", "root").unwrap();
3291        let policy = Policy::default()
3292            .layer("task")
3293            .deny_write("vendor/**")
3294            .rule(
3295                crate::policy::Act::Exec,
3296                crate::policy::Effect::Allow,
3297                "cargo",
3298            );
3299
3300        store.record_run_policy(run, &policy).unwrap();
3301
3302        // Equal, not merely similar: the layers, their order, and the defaults
3303        // are the boundary, so a lossy round trip is a wrong boundary.
3304        assert_eq!(store.run_policy(run).unwrap(), Some(policy));
3305    }
3306
3307    #[test]
3308    fn a_permissive_policy_reads_back_permissive() {
3309        let store = Store::memory().unwrap();
3310        let run = store.start_run("goal", "root").unwrap();
3311        store.record_run_policy(run, &Policy::permissive()).unwrap();
3312
3313        let back = store.run_policy(run).unwrap().expect("a row was recorded");
3314        assert!(back.is_permissive());
3315    }
3316
3317    #[test]
3318    fn a_run_with_no_recorded_policy_reads_back_none_not_permissive() {
3319        let store = Store::memory().unwrap();
3320        let unrecorded = store.start_run("goal", "root").unwrap();
3321        let permissive = store.start_run("goal", "root").unwrap();
3322        store
3323            .record_run_policy(permissive, &Policy::permissive())
3324            .unwrap();
3325
3326        // The distinction the table exists for: a 0.12.0 run wrote no row, and
3327        // "nobody recorded a policy" must never be read as "the caller chose to
3328        // enforce nothing".
3329        assert_eq!(store.run_policy(unrecorded).unwrap(), None);
3330        assert!(store.run_policy(permissive).unwrap().is_some());
3331    }
3332
3333    #[test]
3334    fn re_recording_a_policy_for_the_same_run_replaces_it() {
3335        let store = Store::memory().unwrap();
3336        let run = store.start_run("goal", "root").unwrap();
3337        store.record_run_policy(run, &Policy::permissive()).unwrap();
3338        store.record_run_policy(run, &Policy::default()).unwrap();
3339
3340        assert_eq!(store.run_policy(run).unwrap(), Some(Policy::default()));
3341        let rows: i64 = store
3342            .conn
3343            .query_row(
3344                "SELECT COUNT(*) FROM run_policies WHERE run_id = ?1",
3345                [run],
3346                |r| r.get(0),
3347            )
3348            .unwrap();
3349        assert_eq!(rows, 1);
3350    }
3351}