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;
14use crate::pricing::{PriceTable, Spend};
15use crate::provider::Usage;
16use crate::web::{Citation, ServerToolCall};
17
18/// The group a call with no recorded model falls into. Named rather than
19/// silently merged into a neighbour, and counted as unpriced.
20///
21/// ```
22/// use io_harness::pricing::PriceTable;
23/// use io_harness::{ProviderCall, Store, Usage, UNKNOWN_MODEL};
24///
25/// # fn main() -> io_harness::Result<()> {
26/// # let store = Store::memory()?;
27/// # let run_id = store.start_run("goal", "NOTES.md")?;
28/// // A custom provider that reports no model — a test double, or a wire that
29/// // omits it. The tokens are real and are summed; the money is not knowable.
30/// store.record_provider_call(run_id, &ProviderCall {
31///     step: 1,
32///     provider: "scripted".into(),
33///     usage: Some(Usage { total_tokens: 12, ..Default::default() }),
34///     ..Default::default()
35/// })?;
36///
37/// let spend = store.spend_by_model(&PriceTable::new("2026-07-29"))?;
38/// assert_eq!(spend[0].key, UNKNOWN_MODEL);
39/// assert_eq!(spend[0].usage.total_tokens, 12);
40/// // Counted as unpriced rather than as free, which is the distinction the
41/// // whole accounting release rests on.
42/// assert_eq!((spend[0].cost_micros, spend[0].unpriced_calls), (0, 1));
43/// # Ok(())
44/// # }
45/// ```
46pub const UNKNOWN_MODEL: &str = "(unknown model)";
47
48/// An observation kind as it is stored: the serde rendering, not
49/// [`ObsKind::label`], which is English for a prompt reader and renders `Write`
50/// as "wrote". Going through serde rather than a hand-written match is what
51/// keeps the mapping total — a new variant cannot be added without a rendering,
52/// and the pair below cannot drift apart.
53fn kind_wire(kind: ObsKind) -> String {
54    match serde_json::to_value(kind) {
55        Ok(serde_json::Value::String(s)) => s,
56        // Unreachable for a unit-variant enum with `rename_all`; falling back to
57        // the debug form keeps the write infallible without inventing a kind that
58        // would read back as a different observation.
59        other => format!("{other:?}"),
60    }
61}
62
63/// The inverse of [`kind_wire`]. A kind that does not parse is an error and not
64/// a skipped row: a ledger that came back silently shorter than it was would
65/// assemble a context nobody can account for, which is worse than refusing to
66/// restore it at all.
67fn kind_from_wire(kind: &str, run_id: i64) -> Result<ObsKind> {
68    serde_json::from_value(serde_json::Value::String(kind.to_string())).map_err(|e| Error::Resume {
69        reason: format!("run {run_id} has a ledger observation of unknown kind {kind:?}: {e}"),
70    })
71}
72
73/// The checkpoint layout version stamped into `PRAGMA user_version`. Bump when
74/// the on-disk checkpoint format changes incompatibly. A store whose version is
75/// higher than this is from a newer binary and is refused on resume.
76///
77/// The reason a caller ever reads it: a resume against a store a *newer*
78/// io-harness wrote fails, typed, before anything is replayed — and this constant
79/// is the version to name when reporting that.
80///
81/// ```
82/// use io_harness::{Error, Store, CHECKPOINT_FORMAT};
83///
84/// # fn main() -> io_harness::Result<()> {
85/// let path = std::env::temp_dir().join("io-harness-doc-newer-checkpoint.sqlite3");
86/// let _ = std::fs::remove_file(&path);
87/// {
88///     // Stand in for a store written by a future release.
89///     let conn = rusqlite::Connection::open(&path).unwrap();
90///     conn.pragma_update(None, "user_version", CHECKPOINT_FORMAT + 1).unwrap();
91/// }
92///
93/// let store = Store::open(&path)?;
94/// match store.check_resumable(1) {
95///     Err(Error::Resume { reason }) => {
96///         // Not a panic, and not a half-resume that reads a layout this binary
97///         // does not understand. Tell the operator what to do about it.
98///         assert!(reason.contains("newer"), "{reason}");
99///         eprintln!("this binary understands checkpoint format {CHECKPOINT_FORMAT}: {reason}");
100///     }
101///     other => panic!("a newer store must refuse to resume, got {other:?}"),
102/// }
103/// # let _ = std::fs::remove_file(&path);
104/// # Ok(())
105/// # }
106/// ```
107pub const CHECKPOINT_FORMAT: i64 = 7;
108
109/// The one outcome string that means the run did what it was asked.
110///
111/// Named rather than inlined so that [`RunSummary::success`] and any future
112/// reader agree by construction. Eleven outcome strings exist; exactly one of
113/// them is the task being done.
114///
115/// ```
116/// use io_harness::{Store, SUCCESS_OUTCOME};
117///
118/// # fn main() -> io_harness::Result<()> {
119/// let store = Store::memory()?;
120/// let worked = store.start_run("add a hello function", "src/hello.rs")?;
121/// let gave_up = store.start_run("add a goodbye function", "src/bye.rs")?;
122/// store.finish_run(worked, SUCCESS_OUTCOME)?;
123/// store.finish_run(gave_up, "step_cap_reached")?;
124///
125/// // Scoring a batch of runs: compare against this rather than against the
126/// // literal `"success"`, so a reader and the crate cannot drift apart about
127/// // which of the eleven endings counts.
128/// let succeeded = |id| -> io_harness::Result<bool> {
129///     Ok(store.outcome(id)?.as_deref() == Some(SUCCESS_OUTCOME))
130/// };
131/// assert!(succeeded(worked)?);
132/// assert!(!succeeded(gave_up)?);
133///
134/// // Which is exactly what `RunSummary::success` already carries, computed the
135/// // same way at the moment the run ended.
136/// assert_eq!(store.run_summary(worked)?.map(|s| s.success), Some(true));
137/// # Ok(())
138/// # }
139/// ```
140pub const SUCCESS_OUTCOME: &str = "success";
141
142/// The durable lifecycle status of a run, so a caller can tell a crashed run
143/// (still `Running`) from one paused for a human (`Paused`) or finished
144/// (`Completed`). OS- and rusqlite-free, so it is safe in the public API.
145///
146/// ```
147/// use io_harness::{RunStatus, Store};
148///
149/// # fn main() -> io_harness::Result<()> {
150/// let store = Store::memory()?;
151/// let crashed = store.start_run("summarise the README", "NOTES.md")?;
152/// let waiting = store.start_run("rewrite the changelog", "CHANGELOG.md")?;
153/// let done = store.start_run("add a hello function", "src/hello.rs")?;
154/// store.finish_run(waiting, "awaiting_approval")?;
155/// store.finish_run(done, "success")?;
156///
157/// // The triage a supervisor does on startup. A run left `Running` by a store
158/// // nobody is driving is one whose process died mid-loop: resume it. `Paused` is
159/// // waiting on a human and resumes with their decision, not without it.
160/// let resumable: Vec<i64> = store
161///     .runs()?
162///     .into_iter()
163///     .filter(|id| store.run_status(*id).ok().flatten() == Some(RunStatus::Running))
164///     .collect();
165/// assert_eq!(resumable, [crashed]);
166/// assert_eq!(store.run_status(waiting)?, Some(RunStatus::Paused));
167/// assert_eq!(store.run_status(done)?, Some(RunStatus::Completed));
168/// # Ok(())
169/// # }
170/// ```
171#[derive(Debug, Clone, Copy, PartialEq, Eq)]
172pub enum RunStatus {
173    /// The run is in progress — or was, until the process died mid-loop. A
174    /// `Running` run found in a store is the resume target.
175    Running,
176    /// The run paused for a human decision and can be resumed once it arrives.
177    Paused,
178    /// The run finished (with success or a terminal budget/deny outcome).
179    Completed,
180    /// The run ended in an error.
181    Failed,
182}
183
184impl RunStatus {
185    fn from_str(s: &str) -> Self {
186        match s {
187            "paused" => RunStatus::Paused,
188            "completed" => RunStatus::Completed,
189            "failed" => RunStatus::Failed,
190            _ => RunStatus::Running,
191        }
192    }
193}
194
195/// A persisted spawned-child contract, enough to rebuild and resume that exact
196/// child on a tree resume rather than spawning a duplicate.
197///
198/// ```
199/// use io_harness::Store;
200///
201/// # fn main() -> io_harness::Result<()> {
202/// let store = Store::memory()?;
203/// let parent = store.start_run("summarise the repo", "NOTES.md")?;
204/// let child = store.start_child_run("summarise src/", "NOTES.md", parent, 1)?;
205/// store.record_spawn(parent, 4, child, "summarise src/", "NOTES.md", "#", Some(8), "[]")?;
206///
207/// // What a tree resume does with it: the parent replays step 4, looks the spawn
208/// // up by (parent, step, goal), and adopts the child it already made. Without
209/// // this row the replay would spawn a second child and spend the tree's ledger
210/// // twice for one piece of work.
211/// let row = store.find_spawn(parent, 4, "summarise src/")?.expect("recorded above");
212/// assert_eq!(row.child_run_id, child);
213/// assert_eq!(row.max_steps, Some(8));
214/// // The narrowing the parent applied is stored too, so the adopted child resumes
215/// // under the policy it was contained by rather than the parent's wider one.
216/// assert_eq!(row.deny_write, "[]");
217///
218/// // A step that never spawned has no row, which is how a replay tells "already
219/// // done" from "not done yet".
220/// assert!(store.find_spawn(parent, 5, "summarise src/")?.is_none());
221/// # Ok(())
222/// # }
223/// ```
224#[derive(Debug, Clone, PartialEq)]
225pub struct SpawnRow {
226    /// The child run id already allocated for this spawn.
227    pub child_run_id: i64,
228    /// The child's goal.
229    pub goal: String,
230    /// The workspace-relative file the child's verification reads.
231    pub verify_file: String,
232    /// The substring the child's verification requires.
233    pub needle: String,
234    /// The child's step cap, if the parent set one.
235    pub max_steps: Option<u32>,
236    /// JSON array of `deny_write` globs the parent narrowed the child with.
237    pub deny_write: String,
238}
239
240/// What one finished run cost and whether it worked.
241///
242/// Written once by [`Store::finish_run`] and read back with
243/// [`Store::run_summary`]. Before 0.12.0 a consumer had to assemble this itself
244/// from three different queries plus knowledge of which of eleven outcome
245/// strings count as success — and could not get `duration_ms` at all, because
246/// nothing recorded when a run ended.
247///
248/// Serialisable, so a scoring tool can store or ship it without restating the
249/// shape.
250///
251/// ```
252/// use io_harness::{Store, StepRecord};
253///
254/// # fn main() -> io_harness::Result<()> {
255/// let store = Store::memory()?;
256/// let run_id = store.start_run("add a hello function", "src/hello.rs")?;
257/// store.record(run_id, &StepRecord::new(1, "wrote src/hello.rs", "ok").with_trace("", "", 1_280))?;
258/// store.finish_run(run_id, "step_cap_reached")?;
259///
260/// // One row instead of three queries plus knowledge of which of eleven outcome
261/// // strings mean success. This is what a scoring tool reads per run.
262/// let summary = store.run_summary(run_id)?.expect("the run has finished");
263/// assert!(!summary.success);
264/// // `outcome` and `success` are both kept on purpose: the flag says whether it
265/// // worked, the string says *how* it ended, and a step cap, a stall and a
266/// // human's refusal are three different things to act on.
267/// assert_eq!(summary.outcome, "step_cap_reached");
268/// assert_eq!((summary.steps, summary.tokens), (1, 1_280));
269///
270/// // `None` while a run is unfinished or paused for a human — absent rather than
271/// // a row of zeroes, which would read like a run that did nothing.
272/// let paused = store.start_run("rewrite the changelog", "CHANGELOG.md")?;
273/// store.finish_run(paused, "awaiting_approval")?;
274/// assert!(store.run_summary(paused)?.is_none());
275/// # Ok(())
276/// # }
277/// ```
278#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
279pub struct RunSummary {
280    /// The run this describes.
281    pub run_id: i64,
282    /// The raw outcome string, as written to `runs.outcome`.
283    ///
284    /// Kept alongside [`Self::success`] rather than replaced by it: the string
285    /// says *which* ending, the flag says whether it was the good one, and
286    /// collapsing them would throw away the distinction between a step cap, a
287    /// stall and a human's refusal.
288    pub outcome: String,
289    /// Whether the run achieved what it was asked to do.
290    ///
291    /// True for exactly one outcome — `success`. Every other ending, including
292    /// the ones that are nobody's fault like a rate-limited provider, is not the
293    /// task being done.
294    pub success: bool,
295    /// Steps completed, as `MAX(step)`.
296    ///
297    /// Not `COUNT(*)` over `steps`: a retry writes its own row under the same
298    /// step number, so counting rows counts trace entries rather than agent
299    /// steps. For a tree this is the ROOT agent's step count, not the tree's
300    /// total — each agent has its own summary.
301    pub steps: u32,
302    /// Tokens spent by this run, summed from its committed steps.
303    ///
304    /// Tokens, not money. A provider reports usage and never a price, so the
305    /// crate has nothing to convert with; see
306    /// [`Containment::max_total_cost`](crate::Containment::max_total_cost).
307    pub tokens: u64,
308    /// Wall-clock milliseconds from the run's start to its end.
309    ///
310    /// `None` for a run started before 0.7.0, which has no `started_at` to
311    /// measure from. Includes time the process was not running — a run that
312    /// crashed at midnight and resumed at nine counts the nine hours, because
313    /// that is how long the run took even though it was not working.
314    pub duration_ms: Option<u64>,
315    /// When the run ended, from the database clock.
316    pub finished_at: String,
317}
318
319/// A persisted run store. Use [`Store::open`] for a file, or [`Store::memory`]
320/// for an ephemeral in-memory database.
321///
322/// The store is not a log. It is what makes a run resumable after a crash and
323/// auditable afterwards, so which constructor you choose is a decision about
324/// whether either matters for this run.
325///
326/// ```no_run
327/// use io_harness::{run, OpenRouter, Store, TaskContract, Verification};
328///
329/// # async fn demo() -> io_harness::Result<()> {
330/// // A file store outlives the process: a run that dies mid-loop is resumable
331/// // from its last committed step, and a second process may read the trace while
332/// // this one is still writing it (WAL, plus `BUSY_TIMEOUT`).
333/// let store = Store::open("runs.sqlite3")?;
334///
335/// let contract = TaskContract::new(
336///     "add a hello function returning 42",
337///     "src/hello.rs",
338///     Verification::FileContains("fn hello".into()),
339/// );
340/// let result = run(&contract, &OpenRouter::from_env()?, &store).await?;
341///
342/// // Everything the run did, read back by id: the trace, the budget draws, the
343/// // policy refusals, and what it cost.
344/// for step in store.steps(result.run_id)? {
345///     println!("{}: {} ({} tokens)", step.step, step.decision, step.tokens);
346/// }
347/// println!("{:?}", store.run_summary(result.run_id)?);
348/// # Ok(())
349/// # }
350/// ```
351///
352/// [`Store::memory`] is the same API with no file: a throwaway run, or a test.
353/// Nothing survives the process, so nothing is resumable — which is the right
354/// trade only when a failed run is cheaper to restart than to continue.
355pub struct Store {
356    conn: Connection,
357}
358
359/// One durable checkpoint-lifecycle event: a step was checkpointed, a run was
360/// resumed, or an already-committed step was skipped on resume. Together they
361/// make a crashed-and-resumed run's history reconstructable from the store.
362///
363/// ```
364/// use io_harness::{CheckpointEvent, Store};
365///
366/// # fn main() -> io_harness::Result<()> {
367/// # let store = Store::memory()?;
368/// # let run_id = store.start_run("summarise the repo", "NOTES.md")?;
369/// # store.record_checkpoint_event(&CheckpointEvent::checkpoint(run_id, 1))?;
370/// # store.record_checkpoint_event(&CheckpointEvent::checkpoint(run_id, 2))?;
371/// # store.record_checkpoint_event(&CheckpointEvent::resume(run_id, 3, "restarted after a crash"))?;
372/// # store.record_checkpoint_event(&CheckpointEvent::skipped(run_id, 1))?;
373/// # store.record_checkpoint_event(&CheckpointEvent::skipped(run_id, 2))?;
374/// // Answers the question a crashed run leaves behind: did it restart, and did
375/// // the restart re-do work that was already committed?
376/// let events = store.checkpoint_events(run_id)?;
377/// let resumes = events.iter().filter(|e| e.kind == "resume").count();
378/// let replayed: Vec<u32> = events.iter().filter(|e| e.kind == "skipped").map(|e| e.step).collect();
379///
380/// assert_eq!(resumes, 1, "this run died once and came back");
381/// // Two steps were replayed and recognised as already done, so they cost
382/// // nothing the second time — that is what makes a resume idempotent rather
383/// // than a second charge for the same steps.
384/// assert_eq!(replayed, [1, 2]);
385/// # Ok(())
386/// # }
387/// ```
388#[derive(Debug, Clone, PartialEq)]
389pub struct CheckpointEvent {
390    /// The run this event belongs to.
391    pub run_id: i64,
392    /// The step it concerns.
393    pub step: u32,
394    /// `"checkpoint"`, `"resume"`, or `"skipped"`.
395    pub kind: String,
396    /// Optional human-readable detail (never file contents or secrets).
397    pub detail: Option<String>,
398}
399
400impl CheckpointEvent {
401    /// A step was durably checkpointed.
402    pub fn checkpoint(run_id: i64, step: u32) -> Self {
403        Self {
404            run_id,
405            step,
406            kind: "checkpoint".into(),
407            detail: None,
408        }
409    }
410    /// A run was resumed, re-driving from `step`.
411    pub fn resume(run_id: i64, step: u32, detail: impl Into<String>) -> Self {
412        Self {
413            run_id,
414            step,
415            kind: "resume".into(),
416            detail: Some(detail.into()),
417        }
418    }
419    /// An already-committed step was skipped on resume.
420    pub fn skipped(run_id: i64, step: u32) -> Self {
421        Self {
422            run_id,
423            step,
424            kind: "skipped".into(),
425            detail: None,
426        }
427    }
428}
429
430/// One recorded loop step — the full trace entry, as written and read back.
431///
432/// ```
433/// use io_harness::{StepRecord, Store};
434///
435/// # fn main() -> io_harness::Result<()> {
436/// let store = Store::memory()?;
437/// let run_id = store.start_run("add a hello function", "src/hello.rs")?;
438///
439/// // A transient provider failure, and then the step that worked. Both are step
440/// // 1: a retry writes its own row under the step number it retried.
441/// store.record(run_id, &StepRecord::new(1, "retry 1 after a 503", ""))?;
442///
443/// // `new` alone records a decision and its result; `with_trace` adds the audit
444/// // half — the exact prompt sent, the call the model made, and what it cost.
445/// // Without it the trace says what happened and not why.
446/// store.record(
447///     run_id,
448///     &StepRecord::new(1, "wrote src/hello.rs", "ok").with_trace(
449///         "<the assembled prompt>",
450///         r#"{"name":"write_file","arguments":{"path":"src/hello.rs"}}"#,
451///         1_280,
452///     ),
453/// )?;
454///
455/// // Read back for audit — and the reason `RunSummary::steps` is `MAX(step)`
456/// // rather than a row count: two rows here, one agent step.
457/// let steps = store.steps(run_id)?;
458/// assert_eq!(steps.len(), 2);
459/// assert_eq!(store.last_step(run_id)?, 1);
460/// assert!(steps[1].tool_call.contains("write_file"));
461/// assert_eq!(steps.iter().map(|s| s.tokens).sum::<u64>(), 1_280);
462/// # Ok(())
463/// # }
464/// ```
465#[derive(Debug, Clone, PartialEq)]
466pub struct StepRecord {
467    /// 1-based step number within the run.
468    pub step: u32,
469    /// What the agent decided this step (e.g. "wrote file", "retry 1 after error").
470    pub decision: String,
471    /// Intermediate result / model text for the step.
472    pub result: String,
473    /// The prompt sent to the model this step.
474    pub prompt: String,
475    /// The tool call the model made, as JSON, or "" if none.
476    pub tool_call: String,
477    /// Total tokens used this step, 0 if the provider reported none.
478    pub tokens: u64,
479}
480
481impl StepRecord {
482    /// A trace entry with the audit fields empty — for callers that only record
483    /// a decision and result.
484    pub fn new(step: u32, decision: impl Into<String>, result: impl Into<String>) -> Self {
485        Self {
486            step,
487            decision: decision.into(),
488            result: result.into(),
489            prompt: String::new(),
490            tool_call: String::new(),
491            tokens: 0,
492        }
493    }
494
495    /// Attach the prompt, tool call, and token count for the full trace.
496    pub fn with_trace(
497        mut self,
498        prompt: impl Into<String>,
499        tool_call: impl Into<String>,
500        tokens: u64,
501    ) -> Self {
502        self.prompt = prompt.into();
503        self.tool_call = tool_call.into();
504        self.tokens = tokens;
505        self
506    }
507}
508
509/// One policy event in the trace: an action refused, or a human decision.
510///
511/// Records the path, command, rule, layer, and decision — never file contents
512/// or credentials. (The write payload of a *deferred* action is held separately
513/// in the pending-approval row, because resuming it requires replaying exactly
514/// what was approved.)
515///
516/// ```
517/// use io_harness::{PolicyEvent, Store};
518///
519/// # fn main() -> io_harness::Result<()> {
520/// # let store = Store::memory()?;
521/// # let run_id = store.start_run("tidy the repo", "src/lib.rs")?;
522/// # store.record_event(run_id, &PolicyEvent::refusal(3, "write", "secrets/id_rsa")
523/// #     .with_rule("secrets/*", "app"))?;
524/// # store.record_event(run_id, &PolicyEvent::decision(4, "exec", "git push", "approve", "stdin")
525/// #     .with_performed("git push --dry-run"))?;
526/// // The audit an operator actually asks for: what did the agent try that it was
527/// // not allowed to do, and which rule stopped it?
528/// let events = store.events(run_id)?;
529/// let refused = events.iter().find(|e| e.kind == "refusal").expect("one refusal");
530/// assert_eq!(refused.target, "secrets/id_rsa");
531/// // Attributed to the rule and the layer, so "why was this denied" is answered
532/// // from the store rather than by re-deriving the policy stack by hand.
533/// assert_eq!((refused.rule.as_deref(), refused.layer.as_deref()), (Some("secrets/*"), Some("app")));
534///
535/// // And the case that is easy to miss: an approval that changed the action. The
536/// // agent asked for one command and a human let a different one through.
537/// let approved = events.iter().find(|e| e.kind == "decision").expect("one decision");
538/// assert_eq!(approved.decision.as_deref(), Some("approve"));
539/// assert_eq!(approved.performed.as_deref(), Some("git push --dry-run"));
540/// # Ok(())
541/// # }
542/// ```
543#[derive(Debug, Clone, PartialEq, Eq)]
544pub struct PolicyEvent {
545    /// 1-based step the event occurred on.
546    pub step: u32,
547    /// `"refusal"` or `"decision"`.
548    pub kind: String,
549    /// `"read"`, `"write"`, or `"exec"`.
550    pub act: String,
551    /// The path, or the binary name plus argv for an exec.
552    pub target: String,
553    /// The glob that decided, when a rule rather than a tier default did.
554    pub rule: Option<String>,
555    /// The layer the deciding rule came from.
556    pub layer: Option<String>,
557    /// `"approve"`, `"deny"`, or `"defer"` for a decision.
558    pub decision: Option<String>,
559    /// Which approver decided, or `"remembered"` when a remembered rule did.
560    pub source: Option<String>,
561    /// The action actually performed, when approve-with-changes altered it.
562    pub performed: Option<String>,
563}
564
565impl PolicyEvent {
566    /// An action refused by the policy.
567    pub fn refusal(step: u32, act: impl Into<String>, target: impl Into<String>) -> Self {
568        Self {
569            step,
570            kind: "refusal".into(),
571            act: act.into(),
572            target: target.into(),
573            rule: None,
574            layer: None,
575            decision: None,
576            source: None,
577            performed: None,
578        }
579    }
580
581    /// A human (or built-in approver) decision on a sensitive action.
582    pub fn decision(
583        step: u32,
584        act: impl Into<String>,
585        target: impl Into<String>,
586        decision: impl Into<String>,
587        source: impl Into<String>,
588    ) -> Self {
589        Self {
590            kind: "decision".into(),
591            decision: Some(decision.into()),
592            source: Some(source.into()),
593            ..Self::refusal(step, act, target)
594        }
595    }
596
597    /// Attribute the event to the rule and layer that produced it.
598    pub fn with_rule(mut self, rule: impl Into<String>, layer: impl Into<String>) -> Self {
599        self.rule = Some(rule.into());
600        self.layer = Some(layer.into());
601        self
602    }
603
604    /// Record that the action performed differed from the one requested.
605    pub fn with_performed(mut self, performed: impl Into<String>) -> Self {
606        self.performed = Some(performed.into());
607        self
608    }
609}
610
611/// An action paused awaiting a human decision, persisted so it outlives the
612/// process that requested it.
613///
614/// ```
615/// use io_harness::Store;
616///
617/// # fn main() -> io_harness::Result<()> {
618/// let store = Store::memory()?;
619/// let run_id = store.start_run("update the deploy config", "deploy/prod.yaml")?;
620///
621/// // The agent asked to write a production file, the policy said `Ask`, and the
622/// // approver deferred. The payload is stored with it, because resuming has to
623/// // replay exactly what was approved and not whatever the model would say now.
624/// let request_id = store.put_pending(run_id, 2, "write", "deploy/prod.yaml", Some("replicas: 4"))?;
625///
626/// // This process may now exit. A reviewer — a web UI, a CLI, a person the next
627/// // morning — reads the request back and decides.
628/// let pending = store.pending(request_id)?.expect("just written");
629/// assert_eq!((pending.act.as_str(), pending.target.as_str()), ("write", "deploy/prod.yaml"));
630/// assert_eq!(pending.content.as_deref(), Some("replicas: 4"));
631/// assert!(pending.resolved.is_none(), "nobody has decided yet");
632///
633/// store.resolve_pending(request_id, "approve")?;
634/// assert_eq!(store.pending(request_id)?.unwrap().resolved.as_deref(), Some("approve"));
635/// // `resume_with_decision` then continues the run from here.
636/// # Ok(())
637/// # }
638/// ```
639#[derive(Debug, Clone, PartialEq, Eq)]
640pub struct Pending {
641    /// The request id, as returned by [`Store::put_pending`].
642    pub id: i64,
643    /// The run this action belongs to.
644    pub run_id: i64,
645    /// The step it paused on.
646    pub step: u32,
647    /// `"read"`, `"write"`, or `"exec"`.
648    pub act: String,
649    /// The target path or binary.
650    pub target: String,
651    /// The write payload, needed to replay exactly what was approved.
652    pub content: Option<String>,
653    /// `None` while pending; otherwise `"approve"` or `"deny"`.
654    pub resolved: Option<String>,
655}
656
657/// One event in a tree of agents: a parent spawning a child, a spawn refused by
658/// the containment boundary, or a draw against the tree's shared spend ceiling.
659///
660/// Together with each run's `parent_run_id` these make the tree a reconstructable
661/// graph — who spawned whom, what was refused, and what the tree spent — long
662/// after the process that ran it has exited.
663///
664/// ```
665/// use io_harness::{AgentEvent, Store};
666///
667/// # fn main() -> io_harness::Result<()> {
668/// # let store = Store::memory()?;
669/// # let run_id = store.start_run("summarise the repo", "NOTES.md")?;
670/// # store.record_agent_event(&AgentEvent::budget_draw(run_id, 1, 1_280, 8_720))?;
671/// # store.record_agent_event(&AgentEvent::spawn(run_id, 2, 2, "summarise src/"))?;
672/// # store.record_agent_event(&AgentEvent::budget_draw(run_id, 2, 4_100, 4_620))?;
673/// # store.record_agent_event(&AgentEvent::spawn_refused(run_id, 3, "agents"))?;
674/// # store.record_agent_event(&AgentEvent::budget_draw(run_id, 3, 3_900, 720))?;
675/// // The only audit of what each step drew against the tree's *shared* ceiling.
676/// // A run's own token total does not show this: the ceiling is tree-wide, so
677/// // what matters is how fast `remaining` is falling for everyone.
678/// let events = store.agent_events(run_id)?;
679///
680/// let drawn: u64 = events.iter().filter_map(|e| e.tokens).sum();
681/// let left = events.iter().filter_map(|e| e.remaining).last();
682/// assert_eq!(drawn, 9_280);
683/// // 720 left after three steps that averaged over 3,000: this tree ends in
684/// // `BudgetCeilingReached` next step, and the row says so before it happens.
685/// assert_eq!(left, Some(720));
686///
687/// // The same table carries the shape of the tree and what it was denied.
688/// let children: Vec<i64> = events.iter().filter_map(|e| e.child_run_id).collect();
689/// let refusals: Vec<&str> = events
690///     .iter()
691///     .filter(|e| e.kind == "spawn_refused")
692///     .filter_map(|e| e.detail.as_deref())
693///     .collect();
694/// assert_eq!(children, [2]);
695/// // A parent that wanted a second child and hit the agent cap — a run doing
696/// // less work than it planned to, which is invisible in its outcome.
697/// assert_eq!(refusals, ["agents"]);
698/// # Ok(())
699/// # }
700/// ```
701#[derive(Debug, Clone, PartialEq, Eq)]
702pub struct AgentEvent {
703    /// The agent this event belongs to (the parent, for a spawn; the drawing
704    /// agent, for a budget draw).
705    pub run_id: i64,
706    /// The step it occurred on.
707    pub step: u32,
708    /// `"spawn"`, `"spawn_refused"`, or `"budget_draw"`.
709    pub kind: String,
710    /// The spawned child's run id, for a `"spawn"`.
711    pub child_run_id: Option<i64>,
712    /// Free-form detail: the child's goal for a spawn, the breached cap for a
713    /// refusal.
714    pub detail: Option<String>,
715    /// Tokens drawn, for a `"budget_draw"`.
716    pub tokens: Option<u64>,
717    /// The tree's remaining tokens after the draw.
718    pub remaining: Option<u64>,
719}
720
721impl AgentEvent {
722    /// A parent spawned a child.
723    pub fn spawn(run_id: i64, step: u32, child_run_id: i64, goal: impl Into<String>) -> Self {
724        Self {
725            run_id,
726            step,
727            kind: "spawn".into(),
728            child_run_id: Some(child_run_id),
729            detail: Some(goal.into()),
730            tokens: None,
731            remaining: None,
732        }
733    }
734
735    /// A spawn was refused by the containment boundary.
736    pub fn spawn_refused(run_id: i64, step: u32, cap: &str) -> Self {
737        Self {
738            run_id,
739            step,
740            kind: "spawn_refused".into(),
741            child_run_id: None,
742            detail: Some(cap.into()),
743            tokens: None,
744            remaining: None,
745        }
746    }
747
748    /// An agent drew `tokens` against the tree, leaving `remaining`.
749    pub fn budget_draw(run_id: i64, step: u32, tokens: u64, remaining: u64) -> Self {
750        Self {
751            run_id,
752            step,
753            kind: "budget_draw".into(),
754            child_run_id: None,
755            detail: None,
756            tokens: Some(tokens),
757            remaining: Some(remaining),
758        }
759    }
760}
761
762/// One event in the life of a sandboxed execution: the sandbox created for a
763/// run, a command run in it (with the backend that isolated it), a resource cap
764/// that killed it, a denied network attempt, or the sandbox torn down.
765///
766/// Together these let an operator audit not just *what* code ran but *where* and
767/// *how* it was isolated, reconstructable from the store alone after the run.
768///
769/// ```
770/// use io_harness::{SandboxEvent, Store};
771///
772/// # fn main() -> io_harness::Result<()> {
773/// # let store = Store::memory()?;
774/// # let run_id = store.start_run("make the tests pass", "src/lib.rs")?;
775/// # store.record_sandbox_event(&SandboxEvent::create(run_id, 4, "macos-sandbox-exec"))?;
776/// # store.record_sandbox_event(&SandboxEvent::exec(run_id, 4, "macos-sandbox-exec", "rustc --test subject.rs"))?;
777/// # store.record_sandbox_event(&SandboxEvent::gate_phase_failed(run_id, 4, "criterion-compile"))?;
778/// # store.record_sandbox_event(&SandboxEvent::destroy(run_id, 4))?;
779/// let events = store.sandbox_events(run_id)?;
780///
781/// // Which backend actually isolated the run. The crate picks a native one per
782/// // platform and falls back to a portable floor, so "was this really contained"
783/// // is a question about this field and not about the configuration.
784/// let backend = events.iter().find_map(|e| e.backend.as_deref());
785/// assert_eq!(backend, Some("macos-sandbox-exec"));
786///
787/// // And the phase to look for when a run that used to pass stops passing:
788/// // `criterion-compile` means the criterion no longer compiles *against* the
789/// // subject, which before 0.8.1 could be reported as a pass.
790/// let phases: Vec<&str> = events
791///     .iter()
792///     .filter(|e| e.kind == "gate_phase_failed")
793///     .filter_map(|e| e.detail.as_deref())
794///     .collect();
795/// assert_eq!(phases, ["criterion-compile"]);
796/// # Ok(())
797/// # }
798/// ```
799#[derive(Debug, Clone, PartialEq, Eq)]
800pub struct SandboxEvent {
801    /// The run this execution belongs to.
802    pub run_id: i64,
803    /// The step it occurred on.
804    pub step: u32,
805    /// `"create"`, `"exec"`, `"cap_hit"`, `"destroy"`, `"gate_phase_failed"`
806    /// (whose `detail` names the phase), or `"gate_output"` (whose `detail` is
807    /// what a failing gate command printed).
808    ///
809    /// A `"net_deny"` kind was documented here from 0.6.0 to 0.11.0 and never
810    /// existed: nothing constructed it and nothing emitted it. It was removed in
811    /// 0.12.0 rather than implemented, because a sandbox denies egress
812    /// *structurally* — the backend gives the child no route out, so there is no
813    /// attempt to observe and nothing to count. Network decisions the harness
814    /// actually makes are in `policy_events` with `act = "net"`.
815    pub kind: String,
816    /// The backend that isolated the run (e.g. `"macos-sandbox-exec"`).
817    pub backend: Option<String>,
818    /// The argv for an `"exec"`, the breached cap for a `"cap_hit"`, or the
819    /// bounded output of a failing gate command for a `"gate_output"`. Never the
820    /// agent's file contents and never credentials — the command line, or what
821    /// the caller's own criterion printed.
822    pub detail: Option<String>,
823}
824
825impl SandboxEvent {
826    /// A sandbox was created for a run, isolated by `backend`.
827    pub fn create(run_id: i64, step: u32, backend: &str) -> Self {
828        Self {
829            run_id,
830            step,
831            kind: "create".into(),
832            backend: Some(backend.into()),
833            detail: None,
834        }
835    }
836
837    /// A command ran in the sandbox under `backend`.
838    pub fn exec(run_id: i64, step: u32, backend: &str, argv: &str) -> Self {
839        Self {
840            run_id,
841            step,
842            kind: "exec".into(),
843            backend: Some(backend.into()),
844            detail: Some(argv.into()),
845        }
846    }
847
848    /// A resource cap killed the run.
849    pub fn cap_hit(run_id: i64, step: u32, cap: &str) -> Self {
850        Self {
851            run_id,
852            step,
853            kind: "cap_hit".into(),
854            backend: None,
855            detail: Some(cap.into()),
856        }
857    }
858
859    /// The sandbox was torn down (workdir removed, processes reaped).
860    pub fn destroy(run_id: i64, step: u32) -> Self {
861        Self {
862            run_id,
863            step,
864            kind: "destroy".into(),
865            backend: None,
866            detail: None,
867        }
868    }
869
870    /// Which phase of an execution gate failed: `"subject-compile"` (the file
871    /// under verification does not compile), `"criterion-compile"` (the
872    /// criterion does not compile *against* it), `"test-run"` (it compiled and
873    /// the test failed), or `"subject-emptied"` (the file compiled but a
874    /// crate-level attribute stripped its items, so nothing was type-checked —
875    /// the compile-only gates).
876    ///
877    /// 0.8.1 added this because the release deliberately makes some previously
878    /// passing runs fail. `criterion-compile` is the one to look for: before
879    /// 0.8.1 the subject and the criterion were one crate, so a subject could
880    /// shadow the names the criterion used — or delete it outright — and be
881    /// reported as passing. An operator whose run stopped passing on upgrade can
882    /// tell that case from an ordinary failed criterion without reading the
883    /// harness's source.
884    ///
885    /// A new `kind` value, not a new table or column: a 0.8.0 store takes it
886    /// with no migration.
887    pub fn gate_phase_failed(run_id: i64, step: u32, phase: &str) -> Self {
888        Self {
889            run_id,
890            step,
891            kind: "gate_phase_failed".into(),
892            backend: None,
893            detail: Some(phase.into()),
894        }
895    }
896
897    /// What a failing gate command printed, already bounded by the caller
898    /// (0.17.0).
899    ///
900    /// [`Verification::Command`](crate::Verification::Command) can run any
901    /// command in any language, so "the criterion did not pass" stops being a
902    /// self-explaining outcome: `cargo test` failing because the agent's change
903    /// is wrong and `npm test` failing because the machine has no `node_modules`
904    /// are the same discriminant and need opposite responses. The command's own
905    /// output is the only thing that tells them apart.
906    ///
907    /// This is the one `detail` that is not a command line. It is a *gate*
908    /// command's output — a caller-supplied criterion, never the agent's file
909    /// contents — and it is what makes a language-agnostic gate diagnosable at
910    /// all. Bounded by the caller before it arrives here.
911    ///
912    /// ```
913    /// use io_harness::{SandboxEvent, Store};
914    ///
915    /// # fn main() -> io_harness::Result<()> {
916    /// let store = Store::memory()?;
917    /// let run_id = store.start_run("make the suite pass", "/repo")?;
918    /// store.record_sandbox_event(&SandboxEvent::gate_output(
919    ///     run_id, 3, "FAIL test/parse.test.js\n  expected 2 received 1",
920    /// ))?;
921    ///
922    /// let why = store.sandbox_events(run_id)?;
923    /// assert!(why.iter().any(|e| e.kind == "gate_output"
924    ///     && e.detail.as_deref().is_some_and(|d| d.contains("expected 2"))));
925    /// # Ok(()) }
926    /// ```
927    ///
928    /// A new `kind` value, not a new table or column: a 0.6.0 store takes it with
929    /// no migration.
930    pub fn gate_output(run_id: i64, step: u32, output: &str) -> Self {
931        Self {
932            run_id,
933            step,
934            kind: "gate_output".into(),
935            backend: None,
936            detail: Some(output.into()),
937        }
938    }
939}
940
941/// One event in the life of an MCP connection: a server connected, a tool it
942/// offered, a tool called (with how long it took and whether it worked), or a
943/// server disconnected.
944///
945/// The `net` half of a run's egress history lives in [`PolicyEvent`] — an MCP
946/// server's host is checked by the same policy as any other outbound call — so
947/// this table is about the MCP conversation itself, not about permission.
948///
949/// ```
950/// use io_harness::{McpEvent, Store};
951///
952/// # fn main() -> io_harness::Result<()> {
953/// # let store = Store::memory()?;
954/// # let run_id = store.start_run("summarise the repo", "NOTES.md")?;
955/// # store.record_mcp(run_id, &McpEvent::connected("files", "stdio").with_millis(42))?;
956/// # store.record_mcp(run_id, &McpEvent::discovered("files", "mcp__files__read"))?;
957/// # store.record_mcp(run_id, &McpEvent::discovered("files", "mcp__files__list"))?;
958/// # store.record_mcp(run_id, &McpEvent::called("files", "mcp__files__read", true).at_step(2).with_millis(31))?;
959/// # store.record_mcp(run_id, &McpEvent::called("files", "mcp__files__read", false).at_step(3).with_millis(30_000).with_detail("timeout"))?;
960/// let events = store.mcp_events(run_id)?;
961///
962/// // What a server actually offered this run — the answer to "why did the model
963/// // not use the tool I configured", which is usually that it was never
964/// // discovered. Namespaced, so a server can never shadow `write_file`.
965/// let offered: Vec<&str> = events
966///     .iter()
967///     .filter(|e| e.kind == "discovered")
968///     .filter_map(|e| e.tool.as_deref())
969///     .collect();
970/// assert_eq!(offered, ["mcp__files__read", "mcp__files__list"]);
971///
972/// // And the call that went wrong, with how long it took before it did. Detail
973/// // carries a short note only — never arguments or results, which can carry
974/// // secrets.
975/// let failed = events.iter().find(|e| e.ok == Some(false)).expect("one failure");
976/// assert_eq!((failed.step, failed.millis), (3, Some(30_000)));
977/// assert_eq!(failed.detail.as_deref(), Some("timeout"));
978/// # Ok(())
979/// # }
980/// ```
981#[derive(Debug, Clone, PartialEq, Eq)]
982pub struct McpEvent {
983    /// The step it occurred on. `0` for connect/discover, which happen before
984    /// the run's first step.
985    pub step: u32,
986    /// `"connected"`, `"discovered"`, `"called"`, or `"disconnected"`.
987    pub kind: String,
988    /// The configured server's id.
989    pub server: String,
990    /// The namespaced tool name, for `"discovered"` and `"called"`.
991    pub tool: Option<String>,
992    /// Whether a `"called"` tool succeeded.
993    pub ok: Option<bool>,
994    /// How long a connect or call took, in milliseconds.
995    pub millis: Option<u64>,
996    /// Transport for a connect, or a note such as `"truncated"`. Never tool
997    /// arguments or results — those can carry secrets.
998    pub detail: Option<String>,
999}
1000
1001impl McpEvent {
1002    fn new(kind: &str, server: &str) -> Self {
1003        Self {
1004            step: 0,
1005            kind: kind.into(),
1006            server: server.into(),
1007            tool: None,
1008            ok: None,
1009            millis: None,
1010            detail: None,
1011        }
1012    }
1013
1014    /// A server connected over `transport`.
1015    pub fn connected(server: &str, transport: &str) -> Self {
1016        Self::new("connected", server).with_detail(transport)
1017    }
1018
1019    /// A server offered a tool, under its namespaced name.
1020    pub fn discovered(server: &str, tool: &str) -> Self {
1021        let mut e = Self::new("discovered", server);
1022        e.tool = Some(tool.into());
1023        e
1024    }
1025
1026    /// A tool was called, and whether it worked.
1027    pub fn called(server: &str, tool: &str, ok: bool) -> Self {
1028        let mut e = Self::new("called", server);
1029        e.tool = Some(tool.into());
1030        e.ok = Some(ok);
1031        e
1032    }
1033
1034    /// A server was disconnected.
1035    pub fn disconnected(server: &str) -> Self {
1036        Self::new("disconnected", server)
1037    }
1038
1039    /// Attach the step this happened on.
1040    pub fn at_step(mut self, step: u32) -> Self {
1041        self.step = step;
1042        self
1043    }
1044
1045    /// Attach a duration in milliseconds.
1046    pub fn with_millis(mut self, millis: u64) -> Self {
1047        self.millis = Some(millis);
1048        self
1049    }
1050
1051    /// Attach a short note. An empty note is dropped rather than stored blank.
1052    pub fn with_detail(mut self, detail: impl Into<String>) -> Self {
1053        let detail = detail.into();
1054        self.detail = (!detail.is_empty()).then_some(detail);
1055        self
1056    }
1057}
1058
1059// ---- 0.10.0: durable cross-run memory ----
1060
1061/// One durable memory entry: a fact or decision an agent wrote deliberately,
1062/// keyed to a workspace rather than to a run.
1063///
1064/// ```
1065/// use io_harness::Store;
1066///
1067/// # fn main() -> io_harness::Result<()> {
1068/// let store = Store::memory()?;
1069/// let first = store.start_run("make the tests pass", "src/lib.rs")?;
1070///
1071/// // What an agent learned the expensive way, written once so the next run over
1072/// // this workspace does not spend three steps rediscovering it.
1073/// store.memory_put("/repo", "test-command", "cargo test --features documents", first, 6)?;
1074///
1075/// // A later run — a different process, days afterwards — recalls it by key.
1076/// let entry = store.memory_get("/repo", "test-command")?.expect("written above");
1077/// assert_eq!(entry.value, "cargo test --features documents");
1078/// // Attributed, which is what makes a stale fact traceable: this came from run
1079/// // 1, step 6, and you can go and read what that step actually did.
1080/// assert_eq!((entry.run_id, entry.step), (first, 6));
1081///
1082/// // Keyed to the workspace, never to the run: another workspace sees nothing.
1083/// assert!(store.memory_get("/other-repo", "test-command")?.is_none());
1084/// # Ok(())
1085/// # }
1086/// ```
1087#[derive(Debug, Clone, PartialEq)]
1088pub struct MemoryEntry {
1089    /// The name it is recalled by, unique within its workspace.
1090    pub key: String,
1091    /// The remembered text.
1092    pub value: String,
1093    /// The run that wrote it, so a later reader knows where a fact came from.
1094    pub run_id: i64,
1095    /// The step of that run which wrote it.
1096    pub step: u32,
1097    /// UTC write time, refreshed on every overwrite so ordering is by recency.
1098    pub created_at: String,
1099}
1100
1101/// One turn of a conversation: what was asked, which run answered it, and which
1102/// earlier turn it answers from.
1103///
1104/// A node of the tree [`Session`](crate::Session) walks. `parent_turn_id` is what
1105/// makes it a tree rather than a list: two turns may share a parent, which is
1106/// what a branch is, and nothing is rewritten to create one.
1107///
1108/// ```
1109/// use io_harness::Store;
1110///
1111/// # fn main() -> io_harness::Result<()> {
1112/// let store = Store::memory()?;
1113/// let session = store.create_session("/repo")?;
1114///
1115/// // A turn is recorded against the run that will serve it, before that run
1116/// // starts, so a turn whose process dies is still in the tree.
1117/// let run = store.start_run("where is the retry policy applied?", "/repo")?;
1118/// let first = store.record_turn(session, None, run, "where is the retry policy applied?")?;
1119/// store.finish_turn(first, Some("in `complete_with_retry`"), "finished")?;
1120///
1121/// let turn = store.session_turn(first)?.expect("recorded above");
1122/// assert_eq!(turn.run_id, run);
1123/// assert_eq!(turn.parent_turn_id, None); // the root of this conversation
1124/// assert_eq!(turn.reply.as_deref(), Some("in `complete_with_retry`"));
1125///
1126/// // Everything that turn cost, refused or committed is in the run tables under
1127/// // `turn.run_id` — a turn adds a conversation over a run, not a second trace.
1128/// assert!(store.run_status(turn.run_id)?.is_some());
1129/// # Ok(())
1130/// # }
1131/// ```
1132#[derive(Debug, Clone, PartialEq)]
1133pub struct Turn {
1134    /// The turn's own id, and what [`Session::branch_from`](crate::Session::branch_from) takes.
1135    pub id: i64,
1136    /// The session it belongs to.
1137    pub session_id: i64,
1138    /// The turn it answers from, or `None` for the root of a conversation.
1139    pub parent_turn_id: Option<i64>,
1140    /// The run that served it. Every step, refusal and budget draw is under this id.
1141    pub run_id: i64,
1142    /// What the operator said.
1143    pub prompt: String,
1144    /// What the agent said back, when it said anything. `None` while the turn is
1145    /// still running, and for a turn that stopped without a closing message.
1146    pub reply: Option<String>,
1147    /// Why the turn stopped, as the run's outcome string. `None` while it runs.
1148    pub outcome: Option<String>,
1149    /// UTC creation time.
1150    pub created_at: String,
1151}
1152
1153/// Read one `session_turns` row. One place, so the two queries that read the
1154/// table cannot drift in their column order.
1155fn turn_row(r: &rusqlite::Row<'_>) -> rusqlite::Result<Turn> {
1156    Ok(Turn {
1157        id: r.get(0)?,
1158        session_id: r.get(1)?,
1159        parent_turn_id: r.get(2)?,
1160        run_id: r.get(3)?,
1161        prompt: r.get(4)?,
1162        reply: r.get(5)?,
1163        outcome: r.get(6)?,
1164        created_at: r.get(7)?,
1165    })
1166}
1167
1168/// One question the agent asked the operator, and its answer if it has one (0.21.0).
1169///
1170/// The mirror of [`Pending`], for the other channel: [`Pending`] is an action awaiting
1171/// permission, this is an intent awaiting clarification. An answer here is text the
1172/// model reads and authorizes nothing.
1173///
1174/// ```
1175/// use io_harness::{Question, Store};
1176///
1177/// # fn main() -> io_harness::Result<()> {
1178/// let store = Store::memory()?;
1179/// let run = store.start_run("port the parser", "/repo")?;
1180///
1181/// let id = store.put_question(run, 3, &Question::new("Which config should I edit?")
1182///     .with_choices(["io.toml", "io.local.toml"]))?;
1183///
1184/// // Unanswered, and readable by whatever process is going to answer it.
1185/// let q = store.question(id)?.expect("just written");
1186/// assert!(!q.resolved && q.answer.is_none());
1187/// assert_eq!(q.choices, ["io.toml", "io.local.toml"]);
1188///
1189/// store.answer_question(id, "io.local.toml", "human")?;
1190/// let q = store.question(id)?.unwrap();
1191/// assert_eq!(q.answer.as_deref(), Some("io.local.toml"));
1192/// assert_eq!(q.answered_by.as_deref(), Some("human"));
1193///
1194/// // Answering twice is an error, not a silent second write.
1195/// assert!(store.answer_question(id, "io.toml", "human").is_err());
1196/// # Ok(())
1197/// # }
1198/// ```
1199#[derive(Debug, Clone, PartialEq, Eq)]
1200pub struct PendingQuestion {
1201    /// The question's own id, and what [`resume_with_answer`](crate::resume_with_answer) takes.
1202    pub id: i64,
1203    /// The run that asked.
1204    pub run_id: i64,
1205    /// The step it was asked on.
1206    ///
1207    /// That step **is** committed before the run pauses, so a resume starts after it and
1208    /// the `ask_question` call is not replayed;
1209    /// [`resume_with_answer`](crate::resume_with_answer) delivers the answer as a ledger
1210    /// observation. (A *parent* whose child asked is the different case: its own spawn
1211    /// step is left uncommitted so the resume re-adopts that child.)
1212    pub step: u32,
1213    /// What the agent asked.
1214    pub question: String,
1215    /// What the agent already knew, if it said.
1216    pub context: Option<String>,
1217    /// Options the agent offered. An answer need not be one of them.
1218    pub choices: Vec<String>,
1219    /// The answer, once there is one.
1220    pub answer: Option<String>,
1221    /// `"responder"` if a [`Responder`](crate::Responder) in the run's own process
1222    /// answered, `"human"` if the answer arrived through a resume after a pause.
1223    /// "The machine decided" and "a person decided" are different facts about a run.
1224    pub answered_by: Option<String>,
1225    /// Whether it has been answered.
1226    pub resolved: bool,
1227}
1228
1229/// Read one `pending_questions` row. One place, so the three queries that read the
1230/// table cannot drift in their column order.
1231fn question_row(r: &rusqlite::Row<'_>) -> rusqlite::Result<PendingQuestion> {
1232    let choices: Option<String> = r.get(5)?;
1233    Ok(PendingQuestion {
1234        id: r.get(0)?,
1235        run_id: r.get(1)?,
1236        step: r.get(2)?,
1237        question: r.get(3)?,
1238        context: r.get(4)?,
1239        choices: choices
1240            .and_then(|c| serde_json::from_str(&c).ok())
1241            .unwrap_or_default(),
1242        answer: r.get(6)?,
1243        answered_by: r.get(7)?,
1244        resolved: r.get::<_, i64>(8)? != 0,
1245    })
1246}
1247
1248/// Where one item of an agent's plan has got to (0.21.0).
1249///
1250/// Three states and no more. A plan is for an operator to read at a glance, and
1251/// every state past these three — blocked, cancelled, deferred, in-review — is a
1252/// distinction the harness would have to define, the model would have to choose
1253/// correctly, and nothing would ever check.
1254///
1255/// ```
1256/// use io_harness::TodoState;
1257///
1258/// // The wire form is what the model writes and what the column holds.
1259/// assert_eq!(TodoState::Active.as_str(), "active");
1260/// assert_eq!(TodoState::parse("done"), Some(TodoState::Done));
1261///
1262/// // An unknown state is not silently a default: a plan that says something the
1263/// // crate does not understand is a plan whose author should hear about it.
1264/// assert_eq!(TodoState::parse("blocked"), None);
1265/// ```
1266#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
1267#[serde(rename_all = "snake_case")]
1268pub enum TodoState {
1269    /// Not started.
1270    Pending,
1271    /// Being worked on now. Nothing enforces that only one item is active.
1272    Active,
1273    /// Finished, as far as the agent is concerned. Nothing verifies it.
1274    Done,
1275}
1276
1277impl TodoState {
1278    /// The wire and column form.
1279    pub fn as_str(&self) -> &'static str {
1280        match self {
1281            Self::Pending => "pending",
1282            Self::Active => "active",
1283            Self::Done => "done",
1284        }
1285    }
1286
1287    /// Parse the wire form, or `None` for anything else.
1288    ///
1289    /// `None` rather than a default, so a model that invents a state gets told
1290    /// instead of having its plan quietly rewritten.
1291    pub fn parse(s: &str) -> Option<Self> {
1292        match s {
1293            "pending" => Some(Self::Pending),
1294            "active" => Some(Self::Active),
1295            "done" => Some(Self::Done),
1296            _ => None,
1297        }
1298    }
1299}
1300
1301/// One line of an agent's plan (0.21.0).
1302///
1303/// A plan is the agent's *stated intent*: it is written by the agent, read by an
1304/// operator, and enforced by nothing. No [`RunOutcome`](crate::RunOutcome) depends
1305/// on it, no verification consults it, and writing one is not an act the policy
1306/// gates — see the plan section of `docs/CONTRACT.md`. What it buys is a long run
1307/// that can be recognised as going the wrong way before it ends.
1308///
1309/// There is no item id. The whole list is replaced on every write, which is why
1310/// there is nothing for a model to mis-address.
1311///
1312/// ```
1313/// use io_harness::{Store, TodoItem, TodoState};
1314///
1315/// # fn main() -> io_harness::Result<()> {
1316/// let store = Store::memory()?;
1317/// let run = store.start_run("port the parser", "/repo")?;
1318///
1319/// store.write_todos(run, &[
1320///     TodoItem::new("read the current parser", TodoState::Done),
1321///     TodoItem::new("port the tokenizer", TodoState::Active),
1322///     TodoItem::new("port the error paths", TodoState::Pending),
1323/// ])?;
1324///
1325/// // Read back in the order it was written — which is the order an operator reads.
1326/// let plan = store.todos(run)?;
1327/// assert_eq!(plan.len(), 3);
1328/// assert_eq!(plan[1].text, "port the tokenizer");
1329/// assert_eq!(plan[1].state, TodoState::Active);
1330///
1331/// // A second write replaces the list rather than merging into it.
1332/// store.write_todos(run, &[TodoItem::new("ship it", TodoState::Active)])?;
1333/// assert_eq!(store.todos(run)?.len(), 1);
1334/// # Ok(())
1335/// # }
1336/// ```
1337#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
1338pub struct TodoItem {
1339    /// What the agent said it would do, in its own words.
1340    pub text: String,
1341    /// Where it says that has got to.
1342    pub state: TodoState,
1343}
1344
1345impl TodoItem {
1346    /// One item.
1347    pub fn new(text: impl Into<String>, state: TodoState) -> Self {
1348        Self {
1349            text: text.into(),
1350            state,
1351        }
1352    }
1353}
1354
1355/// Most items one plan may hold.
1356///
1357/// A plan longer than this is not a plan an operator reads; it is a transcript.
1358/// A write past the cap is truncated to the cap rather than refused, and the tool
1359/// says so in its observation — the same shape as every other bounded tool result
1360/// in the crate, which caps and tells rather than failing.
1361///
1362/// ```
1363/// use io_harness::{Store, TodoItem, TodoState, TODO_MAX_ITEMS};
1364///
1365/// # fn main() -> io_harness::Result<()> {
1366/// let store = Store::memory()?;
1367/// let run = store.start_run("a very long plan", "/repo")?;
1368///
1369/// // Two past the cap. The plan is held to the cap and the overflow is reported,
1370/// // rather than the write being refused.
1371/// let long: Vec<TodoItem> = (0..TODO_MAX_ITEMS + 2)
1372///     .map(|i| TodoItem::new(format!("step {i}"), TodoState::Pending))
1373///     .collect();
1374/// let dropped = store.write_todos(run, &long)?;
1375///
1376/// assert_eq!(dropped, 2);
1377/// assert_eq!(store.todos(run)?.len(), TODO_MAX_ITEMS);
1378/// # Ok(())
1379/// # }
1380/// ```
1381pub const TODO_MAX_ITEMS: usize = 64;
1382
1383/// Longest one item's text may be, in characters.
1384///
1385/// Truncated to fit rather than refused, like every other bounded tool result here.
1386///
1387/// ```
1388/// use io_harness::{Store, TodoItem, TodoState, TODO_TEXT_CAP};
1389///
1390/// # fn main() -> io_harness::Result<()> {
1391/// let store = Store::memory()?;
1392/// let run = store.start_run("a wordy plan", "/repo")?;
1393///
1394/// let wordy = "x".repeat(TODO_TEXT_CAP + 50);
1395/// store.write_todos(run, &[TodoItem::new(wordy, TodoState::Active)])?;
1396///
1397/// assert_eq!(store.todos(run)?[0].text.chars().count(), TODO_TEXT_CAP);
1398/// # Ok(())
1399/// # }
1400/// ```
1401pub const TODO_TEXT_CAP: usize = 200;
1402
1403/// Most entries one workspace may hold.
1404///
1405/// A write past the cap is not refused — the oldest entry is evicted to make
1406/// room, and the evicted keys come back so the caller can record the loss in the
1407/// trace rather than discovering it later as a fact that quietly stopped existing.
1408///
1409/// ```
1410/// use io_harness::{Store, MEMORY_MAX_ENTRIES};
1411///
1412/// # fn main() -> io_harness::Result<()> {
1413/// let store = Store::memory()?;
1414/// for i in 0..MEMORY_MAX_ENTRIES {
1415///     let evicted = store.memory_put("/repo", &format!("fact-{i}"), "small", 1, 1)?;
1416///     assert!(evicted.is_empty(), "everything fits until the cap");
1417/// }
1418///
1419/// // One more. The oldest goes, oldest first, and is named.
1420/// let evicted = store.memory_put("/repo", "fact-new", "small", 1, 2)?;
1421/// assert_eq!(evicted, ["fact-0"]);
1422/// assert_eq!(store.memory_list("/repo")?.len(), MEMORY_MAX_ENTRIES);
1423/// assert!(store.memory_get("/repo", "fact-0")?.is_none());
1424/// # Ok(())
1425/// # }
1426/// ```
1427pub const MEMORY_MAX_ENTRIES: usize = 64;
1428
1429/// Most characters one workspace's entries may total.
1430///
1431/// The second of the two caps, and the one that actually binds: sixty-four
1432/// entries of a paragraph each is a context section nobody budgeted for, so the
1433/// character total evicts even when the entry count is nowhere near its limit.
1434///
1435/// ```
1436/// use io_harness::{Store, MEMORY_MAX_CHARS, MEMORY_MAX_ENTRIES, MEMORY_MAX_ENTRY_CHARS};
1437///
1438/// # fn main() -> io_harness::Result<()> {
1439/// let store = Store::memory()?;
1440/// let long_note = "x".repeat(MEMORY_MAX_ENTRY_CHARS);
1441///
1442/// // Eight full-size notes fill the workspace exactly — well inside the entry
1443/// // count, and exactly on the character ceiling.
1444/// for i in 0..8 {
1445///     assert!(store.memory_put("/repo", &format!("note-{i}"), &long_note, 1, 1)?.is_empty());
1446/// }
1447/// assert!(8 < MEMORY_MAX_ENTRIES);
1448/// assert_eq!(long_note.chars().count() * 8, MEMORY_MAX_CHARS);
1449///
1450/// // The ninth evicts, on characters rather than on count.
1451/// assert_eq!(store.memory_put("/repo", "note-8", &long_note, 1, 2)?, ["note-0"]);
1452/// # Ok(())
1453/// # }
1454/// ```
1455pub const MEMORY_MAX_CHARS: usize = 16_000;
1456
1457/// Most characters one entry may hold. A longer value is cut down to this and
1458/// marked, never refused — a too-long fact is still worth remembering.
1459///
1460/// ```
1461/// use io_harness::{Store, MEMORY_MAX_ENTRY_CHARS};
1462///
1463/// # fn main() -> io_harness::Result<()> {
1464/// let store = Store::memory()?;
1465/// let rambling = "y".repeat(MEMORY_MAX_ENTRY_CHARS * 3);
1466/// store.memory_put("/repo", "build-notes", &rambling, 1, 4)?;
1467///
1468/// // Cut to the ceiling and *marked*, so a later reader can see the value is a
1469/// // fragment. A silent truncation reads like a complete fact that happens to
1470/// // end mid-sentence.
1471/// let stored = store.memory_get("/repo", "build-notes")?.expect("written above");
1472/// assert_eq!(stored.value.chars().count(), MEMORY_MAX_ENTRY_CHARS);
1473/// assert!(stored.value.ends_with("[truncated]"));
1474/// # Ok(())
1475/// # }
1476/// ```
1477pub const MEMORY_MAX_ENTRY_CHARS: usize = MEMORY_MAX_CHARS / 8;
1478
1479/// The visible marker appended to a value that was cut to the per-entry ceiling.
1480const MEMORY_TRUNCATED: &str = "…[truncated]";
1481
1482/// Cut `value` to [`MEMORY_MAX_ENTRY_CHARS`] on a char boundary, marking the
1483/// cut. Returned unchanged when it already fits.
1484fn truncate_memory_value(value: &str) -> String {
1485    if value.chars().count() <= MEMORY_MAX_ENTRY_CHARS {
1486        return value.to_string();
1487    }
1488    let keep = MEMORY_MAX_ENTRY_CHARS - MEMORY_TRUNCATED.chars().count();
1489    let mut out: String = value.chars().take(keep).collect();
1490    out.push_str(MEMORY_TRUNCATED);
1491    out
1492}
1493
1494// ---- 0.10.0: what the context assembler decided ----
1495
1496/// One decision the context assembler made: the section it built for a turn, or
1497/// a stale read it re-read (or was refused).
1498///
1499/// One row per turn plus one per re-read — never one per elided observation,
1500/// which would put a row explosion in the trace to say nothing new. Together
1501/// they answer "why did the model not see the thing it read at step 3", and
1502/// `est_tokens` beside `reported_tokens` is what records the estimator's drift
1503/// from the provider's own count (see [`crate::context::estimate_tokens`]).
1504///
1505/// ```
1506/// use io_harness::{ContextEvent, Store};
1507///
1508/// # fn main() -> io_harness::Result<()> {
1509/// # let store = Store::memory()?;
1510/// # let run_id = store.start_run("summarise the repo", "NOTES.md")?;
1511/// # store.record_context_event(run_id, &ContextEvent::assembled(3, "carried=3 stubbed=5 reread=1", 4_100))?;
1512/// # store.record_context_reported(run_id, 3, 4_690)?;
1513/// # store.record_context_event(run_id, &ContextEvent::reread_refused(3, "secrets/id_rsa: denied by policy"))?;
1514/// let events = store.context_events(run_id)?;
1515///
1516/// // Why the model did not see something it read earlier: five observations were
1517/// // stubbed to fit the budget, and one stale read could not be refreshed at all
1518/// // because the policy now refuses that path.
1519/// let refused: Vec<&str> = events
1520///     .iter()
1521///     .filter(|e| e.kind == "reread_refused")
1522///     .filter_map(|e| e.detail.as_deref())
1523///     .collect();
1524/// assert_eq!(refused, ["secrets/id_rsa: denied by policy"]);
1525///
1526/// // And the pair that makes the budget trustworthy: what the assembler
1527/// // estimated, beside what the provider actually charged for the same turn.
1528/// // Drift here is the estimator being wrong, not the budget being generous.
1529/// let assembled = events.iter().find(|e| e.kind == "assembled").expect("one per turn");
1530/// assert_eq!((assembled.est_tokens, assembled.reported_tokens), (Some(4_100), Some(4_690)));
1531/// # Ok(())
1532/// # }
1533/// ```
1534#[derive(Debug, Clone, PartialEq, Eq)]
1535pub struct ContextEvent {
1536    /// The step it belongs to.
1537    pub step: u32,
1538    /// `"assembled"`, `"reread"`, or `"reread_refused"`.
1539    pub kind: String,
1540    /// For `"assembled"`, the turn's summary (`carried=3 stubbed=5 reread=1`);
1541    /// for a re-read, the path and why it was or was not re-read.
1542    pub detail: Option<String>,
1543    /// The assembler's own estimate for the section it built.
1544    pub est_tokens: Option<u64>,
1545    /// What the provider said the request actually cost, when it says anything.
1546    pub reported_tokens: Option<u64>,
1547}
1548
1549impl ContextEvent {
1550    /// The section built for one turn.
1551    pub fn assembled(step: u32, detail: impl Into<String>, est_tokens: u64) -> Self {
1552        Self {
1553            step,
1554            kind: "assembled".into(),
1555            detail: Some(detail.into()),
1556            est_tokens: Some(est_tokens),
1557            reported_tokens: None,
1558        }
1559    }
1560
1561    /// A stale read was re-read at assembly time.
1562    pub fn reread(step: u32, detail: impl Into<String>) -> Self {
1563        Self {
1564            step,
1565            kind: "reread".into(),
1566            detail: Some(detail.into()),
1567            est_tokens: None,
1568            reported_tokens: None,
1569        }
1570    }
1571
1572    /// A stale read could not be re-read — the policy refused it, or it is gone.
1573    pub fn reread_refused(step: u32, detail: impl Into<String>) -> Self {
1574        Self {
1575            step,
1576            kind: "reread_refused".into(),
1577            detail: Some(detail.into()),
1578            est_tokens: None,
1579            reported_tokens: None,
1580        }
1581    }
1582
1583    /// The agent recorded a durable note for later runs over this workspace.
1584    pub fn memory_write(step: u32, detail: impl Into<String>) -> Self {
1585        Self::of("memory_write", step, detail)
1586    }
1587
1588    /// A note was dropped to hold the workspace's memory caps.
1589    pub fn memory_evict(step: u32, detail: impl Into<String>) -> Self {
1590        Self::of("memory_evict", step, detail)
1591    }
1592
1593    /// Notes from earlier runs were carried into this turn's context.
1594    pub fn memory_recall(step: u32, detail: impl Into<String>) -> Self {
1595        Self::of("memory_recall", step, detail)
1596    }
1597
1598    /// The agent wrote down its plan at this step (0.21.0). The detail is the shape
1599    /// of the plan — how many items and how many done — rather than its text, which
1600    /// is already in the `todos` table and would otherwise be stored twice.
1601    pub fn todo_write(step: u32, detail: impl Into<String>) -> Self {
1602        Self::of("todo_write", step, detail)
1603    }
1604
1605    /// The agent asked the operator about intent at this step (0.21.0).
1606    pub fn question_asked(step: u32, detail: impl Into<String>) -> Self {
1607        Self::of("question_asked", step, detail)
1608    }
1609
1610    /// An answer to that question entered the run (0.21.0). The detail names who
1611    /// answered — a `Responder` in the process, or a human after a pause.
1612    pub fn question_answered(step: u32, detail: impl Into<String>) -> Self {
1613        Self::of("question_answered", step, detail)
1614    }
1615
1616    /// An operator's mid-turn message entered the conversation at this step
1617    /// (0.20.0). Recorded rather than only delivered: a turn that changed course
1618    /// because a human said something must be readable as that afterwards, and the
1619    /// detail is the message's length rather than its text — the message itself is
1620    /// already in the step's observations.
1621    pub fn steered(step: u32, detail: impl Into<String>) -> Self {
1622        Self::of("steered", step, detail)
1623    }
1624
1625    /// Which provider actually answered this step.
1626    ///
1627    /// Recorded only when the answer is not obvious from configuration — a
1628    /// [`Fallback`](crate::provider::Fallback) that fell over. `runs.provider` is one
1629    /// label for a whole run and stops being true the moment a run can use two.
1630    pub fn served(step: u32, provider: impl Into<String>) -> Self {
1631        Self::of("served", step, provider)
1632    }
1633
1634    /// The agent made no progress and was told once to change approach. The run
1635    /// continues.
1636    ///
1637    /// Split from [`Self::stalled`] in 0.12.0. Both were recorded under the one
1638    /// `"stalled"` kind, distinguishable only by prose in `detail` — so anything
1639    /// scoring a run could not tell "was nudged and carried on" from "gave up"
1640    /// without string-matching an English sentence the crate never promised.
1641    pub fn replan(step: u32, detail: impl Into<String>) -> Self {
1642        Self::of("replan", step, detail)
1643    }
1644
1645    /// The agent made no progress, had already been told once, and the run is
1646    /// ending here. Terminal.
1647    ///
1648    /// A nudge that did not work is [`Self::replan`]; see there for why the two
1649    /// are separate kinds since 0.12.0.
1650    pub fn stalled(step: u32, detail: impl Into<String>) -> Self {
1651        Self::of("stalled", step, detail)
1652    }
1653
1654    fn of(kind: &str, step: u32, detail: impl Into<String>) -> Self {
1655        Self {
1656            step,
1657            kind: kind.into(),
1658            detail: Some(detail.into()),
1659            est_tokens: None,
1660            reported_tokens: None,
1661        }
1662    }
1663}
1664
1665/// One call to a provider: what it cost, which model served it, and how long it
1666/// took (0.18.0).
1667///
1668/// **Per call, not per step.** A step that failed twice and then answered is
1669/// three of these, and the two that failed are kept because a model that
1670/// produced tokens before erroring was still billed for them. `steps.tokens`
1671/// holds one integer per step and cannot express any of that.
1672///
1673/// ```
1674/// use io_harness::{ProviderCall, Store, Usage};
1675///
1676/// # fn main() -> io_harness::Result<()> {
1677/// # let store = Store::memory()?;
1678/// # let run_id = store.start_run("summarise the repo", "NOTES.md")?;
1679/// // The first attempt died after the model had already produced tokens...
1680/// store.record_provider_call(run_id, &ProviderCall {
1681///     step: 1,
1682///     attempt: 0,
1683///     provider: "anthropic".into(),
1684///     model: Some("claude-sonnet-4".into()),
1685///     usage: Some(Usage { prompt_tokens: 900, completion_tokens: 20, total_tokens: 920,
1686///                         ..Default::default() }),
1687///     latency_ms: 4_100,
1688///     failure: Some("Server (HTTP 529)".into()),
1689///     ..Default::default()
1690/// })?;
1691/// // ...and the retry answered.
1692/// store.record_provider_call(run_id, &ProviderCall {
1693///     step: 1,
1694///     attempt: 1,
1695///     provider: "anthropic".into(),
1696///     model: Some("claude-sonnet-4".into()),
1697///     usage: Some(Usage { prompt_tokens: 900, completion_tokens: 310, total_tokens: 1_210,
1698///                         cache_read_tokens: 850, ..Default::default() }),
1699///     latency_ms: 2_300,
1700///     ttft_ms: Some(420),
1701///     finish_reason: Some("end_turn".into()),
1702///     ..Default::default()
1703/// })?;
1704///
1705/// let calls = store.provider_calls(run_id)?;
1706/// assert_eq!(calls.len(), 2);
1707///
1708/// // What the step actually cost is the sum over its calls — including the one
1709/// // that failed, which a trace keeping only the winner would have hidden.
1710/// let billed: u64 = calls.iter().filter_map(|c| c.usage).map(|u| u.total_tokens).sum();
1711/// assert_eq!(billed, 2_130);
1712/// assert_eq!(calls[0].failure.as_deref(), Some("Server (HTTP 529)"));
1713/// # Ok(())
1714/// # }
1715/// ```
1716#[derive(Debug, Clone, Default, PartialEq, Eq)]
1717pub struct ProviderCall {
1718    /// The step this call was made for.
1719    pub step: u32,
1720    /// Which attempt at that step this was, counting from 0.
1721    pub attempt: u32,
1722    /// The provider that was asked, by [`Provider::name`](crate::Provider::name).
1723    /// For a [`Fallback`](crate::provider::Fallback) this is the combined label;
1724    /// [`ProviderCall::model`] is what identifies who actually answered.
1725    pub provider: String,
1726    /// The model that served the call, as the provider named it. `None` when it
1727    /// did not say — a custom provider, or a wire that omits it.
1728    pub model: Option<String>,
1729    /// Tokens, as reported. `None` when the provider reported none, which is not
1730    /// the same fact as zero.
1731    pub usage: Option<Usage>,
1732    /// Milliseconds the whole call took, measured by the harness around
1733    /// [`Provider::complete`](crate::Provider::complete) — so it includes the
1734    /// crate's own request building and stream consumption, not only the
1735    /// provider's part.
1736    pub latency_ms: u64,
1737    /// Milliseconds to the first content-bearing chunk, where the path streamed
1738    /// and measured it. `None`, never zero, when nothing measured it.
1739    pub ttft_ms: Option<u64>,
1740    /// The provider's own word for why the model stopped, verbatim.
1741    pub finish_reason: Option<String>,
1742    /// `None` when the call answered; the failure's short name when it did not.
1743    pub failure: Option<String>,
1744}
1745
1746/// One file change a run made, and how many lines it added and removed
1747/// (0.18.0).
1748///
1749/// The counts come from comparing the file's lines before and after, trimming
1750/// the common head and tail — not from a minimal diff. A one-line replacement is
1751/// therefore one added and one removed, and a rewrite of the middle of a file is
1752/// the size of that middle. It answers "how much did this run change" without
1753/// the crate carrying a diff algorithm it has no other use for.
1754///
1755/// ```
1756/// use io_harness::{Edit, Store};
1757///
1758/// # fn main() -> io_harness::Result<()> {
1759/// # let store = Store::memory()?;
1760/// # let run_id = store.start_run("tidy the notes", "NOTES.md")?;
1761/// // A run writes rows like this itself; recorded by hand here so the example
1762/// // needs no model.
1763/// store.record_edit(run_id, &Edit::measure(
1764///     2,
1765///     "edit_file",
1766///     "src/parse.rs",
1767///     "fn parse() {}\nfn other() {}\n",
1768///     "fn parse(s: &str) {}\nfn other() {}\n",
1769/// ))?;
1770///
1771/// let edits = store.edits(run_id)?;
1772/// assert_eq!(edits[0].path, "src/parse.rs");
1773/// // One line out, one line in — the untouched line is not counted as either.
1774/// assert_eq!((edits[0].lines_added, edits[0].lines_removed), (1, 1));
1775/// # Ok(())
1776/// # }
1777/// ```
1778#[derive(Debug, Clone, Default, PartialEq, Eq)]
1779pub struct Edit {
1780    /// The step that made the change.
1781    pub step: u32,
1782    /// The tool that made it — `write_file` or `edit_file`.
1783    pub tool: String,
1784    /// The path, as the agent named it (relative to the workspace root).
1785    pub path: String,
1786    /// Lines present after and not before.
1787    pub lines_added: u64,
1788    /// Lines present before and not after.
1789    pub lines_removed: u64,
1790}
1791
1792impl Edit {
1793    /// The change `new` makes to `old`, by common head and tail comparison.
1794    ///
1795    /// ```
1796    /// use io_harness::Edit;
1797    ///
1798    /// // A one-line replacement inside a file: one line out, one line in.
1799    /// let edit = Edit::measure(1, "edit_file", "a.rs", "fn one() {}\nfn two() {}\n",
1800    ///                                                 "fn one() {}\nfn three() {}\n");
1801    /// assert_eq!((edit.lines_added, edit.lines_removed), (1, 1));
1802    ///
1803    /// // A new file is all addition; rewriting a file with what it already held
1804    /// // is neither, which is the fact a byte-count would have missed.
1805    /// assert_eq!(Edit::measure(1, "write_file", "b.rs", "", "one\ntwo\n").lines_added, 2);
1806    /// let same = Edit::measure(1, "write_file", "b.rs", "one\ntwo\n", "one\ntwo\n");
1807    /// assert_eq!((same.lines_added, same.lines_removed), (0, 0));
1808    /// ```
1809    pub fn measure(step: u32, tool: &str, path: &str, old: &str, new: &str) -> Self {
1810        let old: Vec<&str> = old.lines().collect();
1811        let new: Vec<&str> = new.lines().collect();
1812        let head = old
1813            .iter()
1814            .zip(&new)
1815            .take_while(|(a, b)| a == b)
1816            .count()
1817            .min(old.len().min(new.len()));
1818        let tail = old[head..]
1819            .iter()
1820            .rev()
1821            .zip(new[head..].iter().rev())
1822            .take_while(|(a, b)| a == b)
1823            .count();
1824        Self {
1825            step,
1826            tool: tool.to_string(),
1827            path: path.to_string(),
1828            lines_added: (new.len() - head - tail) as u64,
1829            lines_removed: (old.len() - head - tail) as u64,
1830        }
1831    }
1832}
1833
1834/// How long a contended statement waits for the writer before giving up, set on
1835/// every store opened from a file. Without it rusqlite's default is to fail
1836/// immediately with `SQLITE_BUSY`, which turns a moment of contention into an
1837/// error rather than a short wait.
1838///
1839/// ```no_run
1840/// use io_harness::{Store, BUSY_TIMEOUT};
1841///
1842/// # fn main() -> io_harness::Result<()> {
1843/// // A dashboard tailing a run another process is still writing. `Store::open`
1844/// // sets WAL and this timeout, so this read waits for the writer instead of
1845/// // failing — which is why watching a live run needs no coordination with it.
1846/// let store = Store::open("runs.sqlite3")?;
1847/// let run_id = store.last_run()?.expect("at least one run in the store");
1848/// println!("step {} so far", store.last_step(run_id)?);
1849///
1850/// // The value is public because it is the bound on how long such a read can
1851/// // block: a poller on a shorter interval than this can queue up behind a busy
1852/// // writer, and it should size its own deadline knowing that.
1853/// assert!(BUSY_TIMEOUT >= std::time::Duration::from_secs(1));
1854/// # Ok(())
1855/// # }
1856/// ```
1857pub const BUSY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
1858
1859impl Store {
1860    /// Open (creating if absent) a store at `path` and ensure the schema exists.
1861    ///
1862    /// Sets `journal_mode = WAL` and a [`BUSY_TIMEOUT`], so a second process may
1863    /// read the trace while a run is still writing it without either side
1864    /// blocking or aborting the other. Before 0.12.0 this was a bare
1865    /// `Connection::open`, which left every reader to configure the file itself
1866    /// — reaching around this API to do it, and having to do it before the
1867    /// harness opened the file at all.
1868    ///
1869    /// WAL is a persistent property of the database file, not of this
1870    /// connection: a store opened once by 0.12.0 stays in WAL mode afterwards.
1871    /// That is why it is documented as a migration.
1872    pub fn open(path: impl AsRef<std::path::Path>) -> Result<Self> {
1873        let conn = Connection::open(path)?;
1874        conn.busy_timeout(BUSY_TIMEOUT)?;
1875        // `query_row` rather than `execute`: this pragma returns the resulting
1876        // mode as a row, and rusqlite's `execute` rejects a statement that
1877        // yields rows. The returned mode is not asserted — a database on a
1878        // filesystem that cannot support WAL stays in its previous journal mode
1879        // and still works, just without concurrent readers.
1880        let _: String = conn.query_row("PRAGMA journal_mode = WAL", [], |r| r.get(0))?;
1881        Self::from_conn(conn)
1882    }
1883
1884    /// An in-memory store, for tests and throwaway runs.
1885    pub fn memory() -> Result<Self> {
1886        Self::from_conn(Connection::open_in_memory()?)
1887    }
1888
1889    fn from_conn(conn: Connection) -> Result<Self> {
1890        conn.execute_batch(
1891            "CREATE TABLE IF NOT EXISTS runs (
1892                 id       INTEGER PRIMARY KEY AUTOINCREMENT,
1893                 goal     TEXT NOT NULL,
1894                 file     TEXT NOT NULL,
1895                 outcome  TEXT,
1896                 provider TEXT
1897             );
1898             CREATE TABLE IF NOT EXISTS steps (
1899                 id       INTEGER PRIMARY KEY AUTOINCREMENT,
1900                 run_id   INTEGER NOT NULL REFERENCES runs(id),
1901                 step     INTEGER NOT NULL,
1902                 decision TEXT NOT NULL,
1903                 result   TEXT NOT NULL,
1904                 prompt    TEXT NOT NULL DEFAULT '',
1905                 tool_call TEXT NOT NULL DEFAULT '',
1906                 tokens    INTEGER NOT NULL DEFAULT 0
1907             );",
1908        )?;
1909
1910        // Migrate a 0.1.0 database whose `steps` table predates the trace
1911        // columns. ADD COLUMN errors on an already-present column; ignore it.
1912        for col in [
1913            "prompt TEXT NOT NULL DEFAULT ''",
1914            "tool_call TEXT NOT NULL DEFAULT ''",
1915            "tokens INTEGER NOT NULL DEFAULT 0",
1916        ] {
1917            let _ = conn.execute(&format!("ALTER TABLE steps ADD COLUMN {col}"), []);
1918        }
1919        // 0.3.0: record which provider ran. Additive — a 0.1/0.2 database gains
1920        // the column and a 0.2 binary still reads a migrated database.
1921        let _ = conn.execute("ALTER TABLE runs ADD COLUMN provider TEXT", []);
1922
1923        // 0.4.0: policy refusals/decisions, and actions paused awaiting a human.
1924        // New tables only — a 0.3.0 database gains them and a 0.3.0 binary,
1925        // which never queries them, still reads a migrated database.
1926        conn.execute_batch(
1927            "CREATE TABLE IF NOT EXISTS policy_events (
1928                 id        INTEGER PRIMARY KEY AUTOINCREMENT,
1929                 run_id    INTEGER NOT NULL,
1930                 step      INTEGER NOT NULL,
1931                 kind      TEXT NOT NULL,
1932                 act       TEXT NOT NULL,
1933                 target    TEXT NOT NULL,
1934                 rule      TEXT,
1935                 layer     TEXT,
1936                 decision  TEXT,
1937                 source    TEXT,
1938                 performed TEXT
1939             );
1940             CREATE TABLE IF NOT EXISTS pending_approvals (
1941                 id       INTEGER PRIMARY KEY AUTOINCREMENT,
1942                 run_id   INTEGER NOT NULL,
1943                 step     INTEGER NOT NULL,
1944                 act      TEXT NOT NULL,
1945                 target   TEXT NOT NULL,
1946                 content  TEXT,
1947                 resolved TEXT
1948             );",
1949        )?;
1950
1951        // 0.5.0: sub-agent trees. Runs gain a parent edge and a depth; a new
1952        // table records spawns, spawn refusals, and draws against the tree's
1953        // shared spend ceiling. All additive — a 0.4.0 database gains the column
1954        // and table and a 0.4.0 binary still reads a migrated database.
1955        let _ = conn.execute("ALTER TABLE runs ADD COLUMN parent_run_id INTEGER", []);
1956        let _ = conn.execute(
1957            "ALTER TABLE runs ADD COLUMN depth INTEGER NOT NULL DEFAULT 0",
1958            [],
1959        );
1960        conn.execute_batch(
1961            "CREATE TABLE IF NOT EXISTS agent_events (
1962                 id           INTEGER PRIMARY KEY AUTOINCREMENT,
1963                 run_id       INTEGER NOT NULL,
1964                 step         INTEGER NOT NULL,
1965                 kind         TEXT NOT NULL,
1966                 child_run_id INTEGER,
1967                 detail       TEXT,
1968                 tokens       INTEGER,
1969                 remaining    INTEGER
1970             );",
1971        )?;
1972
1973        // 0.6.0: sandbox lifecycle events (create, exec+backend, cap hit, net
1974        // deny, destroy). New table only — a 0.5.0 database gains it and a 0.5.0
1975        // binary, which never queries it, still reads a migrated database.
1976        conn.execute_batch(
1977            "CREATE TABLE IF NOT EXISTS sandbox_events (
1978                 id       INTEGER PRIMARY KEY AUTOINCREMENT,
1979                 run_id   INTEGER NOT NULL,
1980                 step     INTEGER NOT NULL,
1981                 kind     TEXT NOT NULL,
1982                 backend  TEXT,
1983                 detail   TEXT
1984             );",
1985        )?;
1986
1987        // 0.7.0: durable checkpoint + resume. `runs` gains a resumable status and
1988        // a start timestamp so wall-clock elapsed survives a restart; a new table
1989        // records checkpoint / resume / step-skipped events so a multi-crash run's
1990        // history is reconstructable from the store alone. All additive — a 0.6.0
1991        // database gains the columns/table and a 0.6.0 binary still reads it.
1992        let _ = conn.execute(
1993            "ALTER TABLE runs ADD COLUMN status TEXT NOT NULL DEFAULT 'running'",
1994            [],
1995        );
1996        let _ = conn.execute("ALTER TABLE runs ADD COLUMN started_at TEXT", []);
1997        conn.execute_batch(
1998            "CREATE TABLE IF NOT EXISTS checkpoint_events (
1999                 id     INTEGER PRIMARY KEY AUTOINCREMENT,
2000                 run_id INTEGER NOT NULL,
2001                 step   INTEGER NOT NULL,
2002                 kind   TEXT NOT NULL,
2003                 detail TEXT
2004             );
2005             CREATE TABLE IF NOT EXISTS spawns (
2006                 id            INTEGER PRIMARY KEY AUTOINCREMENT,
2007                 parent_run_id INTEGER NOT NULL,
2008                 step          INTEGER NOT NULL,
2009                 child_run_id  INTEGER NOT NULL,
2010                 goal          TEXT NOT NULL,
2011                 verify_file   TEXT NOT NULL,
2012                 needle        TEXT NOT NULL,
2013                 max_steps     INTEGER,
2014                 deny_write    TEXT NOT NULL DEFAULT '[]'
2015             );",
2016        )?;
2017
2018        // 0.8.0: the MCP conversation — connects, tool discovery, tool calls,
2019        // disconnects. New table only, so a 0.7.0 database gains it and a 0.7.0
2020        // binary, which never queries it, still reads a migrated database. The
2021        // network *verdicts* deliberately do not live here: they go to
2022        // policy_events beside every other permission decision, because an
2023        // operator auditing "what was this run allowed to do" should find them
2024        // in one place.
2025        conn.execute_batch(
2026            "CREATE TABLE IF NOT EXISTS mcp_events (
2027                 id     INTEGER PRIMARY KEY AUTOINCREMENT,
2028                 run_id INTEGER NOT NULL,
2029                 step   INTEGER NOT NULL,
2030                 kind   TEXT NOT NULL,
2031                 server TEXT NOT NULL,
2032                 tool   TEXT,
2033                 ok     INTEGER,
2034                 millis INTEGER,
2035                 detail TEXT
2036             );",
2037        )?;
2038
2039        // 0.10.0: durable cross-run memory — facts and decisions an agent wrote
2040        // deliberately, keyed to a *workspace* instead of a run, so a later run
2041        // recalls what an earlier one learned. New table only, so a 0.9.1
2042        // database gains it and a 0.9.1 binary, which never queries it, still
2043        // reads a migrated database. Deliberately NOT a CHECKPOINT_FORMAT bump:
2044        // no checkpoint layout changed, and bumping it would make
2045        // [`Store::check_resumable`] refuse every 0.9.1 checkpoint.
2046        conn.execute_batch(
2047            "CREATE TABLE IF NOT EXISTS memory (
2048                 id         INTEGER PRIMARY KEY,
2049                 workspace  TEXT NOT NULL,
2050                 key        TEXT NOT NULL,
2051                 value      TEXT NOT NULL,
2052                 run_id     INTEGER NOT NULL,
2053                 step       INTEGER NOT NULL,
2054                 created_at TEXT NOT NULL,
2055                 UNIQUE(workspace, key)
2056             );",
2057        )?;
2058
2059        // 0.10.0: what the context assembler decided each turn — one row per turn
2060        // plus one per re-read. New table only, so a 0.9.1 database gains it and a
2061        // 0.9.1 binary, which never queries it, still opens and resumes a migrated
2062        // database. Deliberately NOT a `CHECKPOINT_FORMAT` bump: nothing about a
2063        // checkpoint's layout changed, and bumping it would refuse every 0.9.1
2064        // store on resume for an additive audit table.
2065        conn.execute_batch(
2066            "CREATE TABLE IF NOT EXISTS context_events (
2067                 id              INTEGER PRIMARY KEY,
2068                 run_id          INTEGER NOT NULL,
2069                 step            INTEGER NOT NULL,
2070                 kind            TEXT NOT NULL,
2071                 detail          TEXT,
2072                 est_tokens      INTEGER,
2073                 reported_tokens INTEGER
2074             );",
2075        )?;
2076
2077        // 0.12.0: one row per finished run, so "did it work, how long did it take,
2078        // what did it cost" is one read rather than a reconstruction.
2079        //
2080        // Every field but the end stamp was already derivable, and derivable was
2081        // not good enough: a consumer had to know that success is one of eleven
2082        // free-text strings, that steps means MAX(step) and not COUNT(*) because
2083        // retry rows share a step number, and that spend is SUM(steps.tokens). That
2084        // is schema knowledge the crate never promised, so io-eval would have been
2085        // coupled to internals from its first line.
2086        //
2087        // `finished_at` is the genuinely new fact. Nothing in the schema recorded
2088        // when a run ENDED — only `runs.started_at` — and `Store::elapsed_secs`
2089        // measures against `julianday('now')`, so it keeps growing after the run is
2090        // over and cannot reconstruct a finished run's latency. Stamped from
2091        // SQLite's clock for the same reason `started_at` is: the pair must come
2092        // from one clock or the difference is meaningless.
2093        //
2094        // A separate table rather than columns on `runs`: additive, and `runs` is
2095        // read by resume on the hot path. New table only, so a 0.11.0 database gains
2096        // it and a 0.11.0 binary, which never queries it, still opens and resumes a
2097        // migrated database. Deliberately NOT a `CHECKPOINT_FORMAT` bump — no
2098        // checkpoint layout changed, and bumping it would make
2099        // [`Store::check_resumable`] refuse every 0.11.0 store for an additive table.
2100        conn.execute_batch(
2101            "CREATE TABLE IF NOT EXISTS run_outcomes (
2102                 run_id      INTEGER PRIMARY KEY,
2103                 outcome     TEXT NOT NULL,
2104                 success     INTEGER NOT NULL,
2105                 steps       INTEGER NOT NULL,
2106                 tokens      INTEGER NOT NULL,
2107                 duration_ms INTEGER,
2108                 finished_at TEXT NOT NULL
2109             );",
2110        )?;
2111
2112        // 0.13.0: the policy a run was started under, kept so a later resume can
2113        // tell what boundary the caller enforced instead of guessing. Nothing in
2114        // the schema recorded it: `policy_events` holds the decisions a policy
2115        // produced, which is the opposite direction — a run that was never asked
2116        // to do anything forbidden leaves no events at all, and a permissive run
2117        // leaves none either, so the two are indistinguishable after the fact.
2118        //
2119        // Stored as JSON in one column rather than shredded into rule rows: the
2120        // only reader wants the whole [`Policy`] back, and a serialised blob
2121        // cannot drift from the type the way a hand-written flattening would.
2122        //
2123        // New table only, so a 0.12.0 database gains it and a 0.12.0 binary, which
2124        // never queries it, still opens and resumes a migrated database.
2125        // Deliberately NOT a `CHECKPOINT_FORMAT` bump: no checkpoint layout
2126        // changed, and bumping it would make [`Store::check_resumable`] refuse
2127        // every 0.12.0 store for an additive table.
2128        conn.execute_batch(
2129            "CREATE TABLE IF NOT EXISTS run_policies (
2130                 run_id INTEGER PRIMARY KEY,
2131                 policy TEXT NOT NULL
2132             );",
2133        )?;
2134
2135        // 0.13.0: the observation ledger the context assembler builds, made
2136        // durable so a resumed run restores the context it had instead of
2137        // re-deriving one from the workspace.
2138        //
2139        // The text was already durable — `steps.result` holds one step's
2140        // observations concatenated — but concatenated is the problem: a step with
2141        // three observations stores one string, and the typed triple assembly
2142        // actually reasons about (`step`, `kind`, `target`) is not recoverable
2143        // from it at all. `ObsKind::target_is_the_subject` decides supersession
2144        // from `kind`, so a ledger rebuilt from `steps.result` would assemble
2145        // differently from the one it replaced, which is worse than the honest
2146        // re-derivation it would be replacing.
2147        //
2148        // One row per observation, ordered by `id` like every other event table
2149        // here, because the ledger is an ordered log and `step` alone does not
2150        // order the observations within a step.
2151        //
2152        // New table only, so a 0.12.0 database gains it and a 0.12.0 binary, which
2153        // never queries it, still opens and resumes a migrated database.
2154        // Deliberately NOT a `CHECKPOINT_FORMAT` bump: no checkpoint layout
2155        // changed, and bumping it would make [`Store::check_resumable`] refuse
2156        // every 0.12.0 store for an additive table.
2157        conn.execute_batch(
2158            "CREATE TABLE IF NOT EXISTS ledger_observations (
2159                 id     INTEGER PRIMARY KEY,
2160                 run_id INTEGER NOT NULL,
2161                 step   INTEGER NOT NULL,
2162                 kind   TEXT NOT NULL,
2163                 target TEXT,
2164                 text   TEXT NOT NULL
2165             );",
2166        )?;
2167
2168        // 0.18.0: accounting. One row per provider call and one per file change.
2169        //
2170        // `provider_calls` is per CALL, not per step, which is the whole point:
2171        // `steps.tokens` holds one integer for a step, so a step that retried
2172        // twice and then fell over to a second vendor collapsed into a single
2173        // number attributed to nothing. A row per attempt keeps what was actually
2174        // paid for — including the attempts that failed after the model had
2175        // already produced tokens.
2176        //
2177        // No cost column, deliberately. Money is derived at query time from a
2178        // price table the operator owns ([`crate::pricing`]), because a stored
2179        // dollar figure is wrong the moment a price changes or was entered wrong,
2180        // and cannot then be repaired without rewriting history.
2181        //
2182        // `at` comes from SQLite's clock, like `runs.started_at`, so the day a
2183        // call is grouped into and the run's elapsed time come from one clock
2184        // rather than two that can disagree.
2185        //
2186        // New tables only, so a 0.17.0 database gains them and a 0.17.0 binary,
2187        // which never queries them, still opens and resumes a migrated database.
2188        // Deliberately NOT a `CHECKPOINT_FORMAT` bump: no checkpoint layout
2189        // changed, and bumping it would make [`Store::check_resumable`] refuse
2190        // every 0.17.0 store for two additive tables.
2191        conn.execute_batch(
2192            "CREATE TABLE IF NOT EXISTS provider_calls (
2193                 id                   INTEGER PRIMARY KEY AUTOINCREMENT,
2194                 run_id               INTEGER NOT NULL,
2195                 step                 INTEGER NOT NULL,
2196                 attempt              INTEGER NOT NULL,
2197                 provider             TEXT NOT NULL,
2198                 model                TEXT,
2199                 prompt_tokens        INTEGER,
2200                 completion_tokens    INTEGER,
2201                 total_tokens         INTEGER,
2202                 cache_read_tokens    INTEGER,
2203                 cache_write_tokens   INTEGER,
2204                 reasoning_tokens     INTEGER,
2205                 server_tool_requests INTEGER,
2206                 latency_ms           INTEGER NOT NULL,
2207                 ttft_ms              INTEGER,
2208                 finish_reason        TEXT,
2209                 failure              TEXT,
2210                 at                   TEXT NOT NULL DEFAULT (datetime('now'))
2211             );
2212             CREATE TABLE IF NOT EXISTS edits (
2213                 id            INTEGER PRIMARY KEY AUTOINCREMENT,
2214                 run_id        INTEGER NOT NULL,
2215                 step          INTEGER NOT NULL,
2216                 tool          TEXT NOT NULL,
2217                 path          TEXT NOT NULL,
2218                 lines_added   INTEGER NOT NULL,
2219                 lines_removed INTEGER NOT NULL,
2220                 at            TEXT NOT NULL DEFAULT (datetime('now'))
2221             );",
2222        )?;
2223
2224        // 0.20.0 — the session layer. A conversation is a tree of turns and a turn
2225        // is a run, so the only new state is the tree itself: which turns a session
2226        // has, which turn each one answers, and which run served it. Everything a
2227        // turn cost, refused, or committed is already in the tables above under its
2228        // `run_id`.
2229        //
2230        // New tables only, as every addition since 0.13.0 has been, and deliberately
2231        // NOT a `CHECKPOINT_FORMAT` bump: no checkpoint layout changed, and bumping
2232        // it would make [`Store::check_resumable`] refuse every 0.19.0 store for two
2233        // additive tables. A 0.19.0 binary never queries them and opens a migrated
2234        // database unchanged.
2235        //
2236        // `head_turn_id` is a column rather than "the last row", because branching
2237        // means the head is a choice and not a maximum.
2238        conn.execute_batch(
2239            "CREATE TABLE IF NOT EXISTS sessions (
2240                 id           INTEGER PRIMARY KEY AUTOINCREMENT,
2241                 root         TEXT NOT NULL,
2242                 head_turn_id INTEGER,
2243                 created_at   TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
2244             );
2245             CREATE TABLE IF NOT EXISTS session_turns (
2246                 id             INTEGER PRIMARY KEY AUTOINCREMENT,
2247                 session_id     INTEGER NOT NULL,
2248                 parent_turn_id INTEGER,
2249                 run_id         INTEGER NOT NULL,
2250                 prompt         TEXT NOT NULL,
2251                 reply          TEXT,
2252                 outcome        TEXT,
2253                 created_at     TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
2254             );",
2255        )?;
2256
2257        // 0.21.0. The agent's plan, one row per item, and the question channel that
2258        // asks an operator what they actually wanted.
2259        //
2260        // New tables again, and again deliberately NOT a `CHECKPOINT_FORMAT` bump:
2261        // no checkpoint layout changed, and bumping it would make
2262        // [`Store::check_resumable`] refuse every 0.20.0 store over two additive
2263        // tables a 0.20.0 binary never queries.
2264        //
2265        // `todos.position` is a column rather than "the rowid order", because the
2266        // list is replaced wholesale and an operator reads it in the order the agent
2267        // wrote it — which after a replace is not the order the ids run in.
2268        //
2269        // `pending_questions` mirrors `pending_approvals` field for field, including
2270        // the `resolved` marker, so a question survives a process exit for exactly
2271        // the reason a pending approval does. `answer` is NULL until a human writes
2272        // one, and `answered_by` records whether it was a `Responder` in the process
2273        // or a person after a pause — "the machine decided" and "a person decided"
2274        // are different facts about a run.
2275        conn.execute_batch(
2276            "CREATE TABLE IF NOT EXISTS todos (
2277                 id         INTEGER PRIMARY KEY AUTOINCREMENT,
2278                 run_id     INTEGER NOT NULL,
2279                 position   INTEGER NOT NULL,
2280                 text       TEXT NOT NULL,
2281                 state      TEXT NOT NULL,
2282                 written_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
2283             );
2284             CREATE INDEX IF NOT EXISTS todos_run ON todos(run_id, position);
2285             CREATE TABLE IF NOT EXISTS pending_questions (
2286                 id          INTEGER PRIMARY KEY AUTOINCREMENT,
2287                 run_id      INTEGER NOT NULL,
2288                 step        INTEGER NOT NULL,
2289                 question    TEXT NOT NULL,
2290                 context     TEXT,
2291                 choices     TEXT,
2292                 answer      TEXT,
2293                 answered_by TEXT,
2294                 resolved    INTEGER NOT NULL DEFAULT 0,
2295                 asked_at    TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
2296             );",
2297        )?;
2298
2299        // 0.22.0 — provider-executed web search and fetch. Two more additive
2300        // tables and, for the same reasons as the two above, NOT a
2301        // `CHECKPOINT_FORMAT` bump: no checkpoint layout changed and a 0.21.0
2302        // binary never queries either of them.
2303        //
2304        // `citations` is what the provider said it drew on, per run and step. The
2305        // crate does not fetch the url or check the page, so these rows are a
2306        // record of what was returned rather than of what is true.
2307        //
2308        // `server_tool_calls` is what the provider *ran*, and exists because a
2309        // failed search arrives inside an HTTP 200 as an error object: without a
2310        // row carrying `error`, a search that broke and a search that found
2311        // nothing are the same empty result set, which is the quiet failure this
2312        // release exists to prevent.
2313        conn.execute_batch(
2314            "CREATE TABLE IF NOT EXISTS citations (
2315                 id         INTEGER PRIMARY KEY AUTOINCREMENT,
2316                 run_id     INTEGER NOT NULL,
2317                 step       INTEGER NOT NULL,
2318                 url        TEXT NOT NULL,
2319                 title      TEXT,
2320                 cited_text TEXT,
2321                 cited_at   TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
2322             );
2323             CREATE INDEX IF NOT EXISTS citations_run ON citations(run_id, step);
2324             CREATE TABLE IF NOT EXISTS server_tool_calls (
2325                 id       INTEGER PRIMARY KEY AUTOINCREMENT,
2326                 run_id   INTEGER NOT NULL,
2327                 step     INTEGER NOT NULL,
2328                 provider TEXT NOT NULL,
2329                 tool     TEXT NOT NULL,
2330                 error    TEXT,
2331                 ran_at   TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
2332             );
2333             CREATE INDEX IF NOT EXISTS server_tool_calls_run ON server_tool_calls(run_id, step);",
2334        )?;
2335
2336        // Stamp the checkpoint-format version. A fresh or pre-0.7.0 database reads
2337        // back 0; we bump it to the current format. A database written by a NEWER
2338        // format reads back a higher number and [`Store::check_resumable`] refuses
2339        // it with a typed [`Error::Resume`] rather than resuming a layout it does
2340        // not understand.
2341        let format: i64 = conn.query_row("PRAGMA user_version", [], |r| r.get(0))?;
2342        if format < CHECKPOINT_FORMAT {
2343            conn.execute_batch(&format!("PRAGMA user_version = {CHECKPOINT_FORMAT}"))?;
2344        }
2345
2346        Ok(Self { conn })
2347    }
2348
2349    /// Record a policy refusal or a human decision against a run.
2350    pub fn record_event(&self, run_id: i64, e: &PolicyEvent) -> Result<()> {
2351        self.conn.execute(
2352            "INSERT INTO policy_events
2353                 (run_id, step, kind, act, target, rule, layer, decision, source, performed)
2354             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
2355            (
2356                run_id,
2357                e.step,
2358                &e.kind,
2359                &e.act,
2360                &e.target,
2361                &e.rule,
2362                &e.layer,
2363                &e.decision,
2364                &e.source,
2365                &e.performed,
2366            ),
2367        )?;
2368        Ok(())
2369    }
2370
2371    /// Every policy event recorded for a run, in order.
2372    pub fn events(&self, run_id: i64) -> Result<Vec<PolicyEvent>> {
2373        let mut stmt = self.conn.prepare(
2374            "SELECT step, kind, act, target, rule, layer, decision, source, performed
2375             FROM policy_events WHERE run_id = ?1 ORDER BY id ASC",
2376        )?;
2377        let rows = stmt.query_map([run_id], |r| {
2378            Ok(PolicyEvent {
2379                step: r.get::<_, i64>(0)? as u32,
2380                kind: r.get(1)?,
2381                act: r.get(2)?,
2382                target: r.get(3)?,
2383                rule: r.get(4)?,
2384                layer: r.get(5)?,
2385                decision: r.get(6)?,
2386                source: r.get(7)?,
2387                performed: r.get(8)?,
2388            })
2389        })?;
2390        Ok(rows.collect::<std::result::Result<_, _>>()?)
2391    }
2392
2393    /// Persist an action awaiting a human decision; returns its request id.
2394    pub fn put_pending(
2395        &self,
2396        run_id: i64,
2397        step: u32,
2398        act: &str,
2399        target: &str,
2400        content: Option<&str>,
2401    ) -> Result<i64> {
2402        self.conn.execute(
2403            "INSERT INTO pending_approvals (run_id, step, act, target, content)
2404             VALUES (?1, ?2, ?3, ?4, ?5)",
2405            (run_id, step, act, target, content),
2406        )?;
2407        Ok(self.conn.last_insert_rowid())
2408    }
2409
2410    /// Read a pending action back by request id.
2411    pub fn pending(&self, request_id: i64) -> Result<Option<Pending>> {
2412        let mut stmt = self.conn.prepare(
2413            "SELECT id, run_id, step, act, target, content, resolved
2414             FROM pending_approvals WHERE id = ?1",
2415        )?;
2416        let mut rows = stmt.query_map([request_id], |r| {
2417            Ok(Pending {
2418                id: r.get(0)?,
2419                run_id: r.get(1)?,
2420                step: r.get::<_, i64>(2)? as u32,
2421                act: r.get(3)?,
2422                target: r.get(4)?,
2423                content: r.get(5)?,
2424                resolved: r.get(6)?,
2425            })
2426        })?;
2427        Ok(rows.next().transpose()?)
2428    }
2429
2430    /// Mark a pending action decided, so a resume knows what the human chose.
2431    pub fn resolve_pending(&self, request_id: i64, decision: &str) -> Result<()> {
2432        self.conn.execute(
2433            "UPDATE pending_approvals SET resolved = ?1 WHERE id = ?2",
2434            (decision, request_id),
2435        )?;
2436        Ok(())
2437    }
2438
2439    /// Start a run row; returns its id. Stamps `started_at` (UTC, from SQLite's
2440    /// clock) so a 24h wall-clock budget survives a restart, and marks the run
2441    /// `running`.
2442    pub fn start_run(&self, goal: &str, file: &str) -> Result<i64> {
2443        self.conn.execute(
2444            "INSERT INTO runs (goal, file, status, started_at)
2445             VALUES (?1, ?2, 'running', strftime('%Y-%m-%dT%H:%M:%fZ','now'))",
2446            (goal, file),
2447        )?;
2448        Ok(self.conn.last_insert_rowid())
2449    }
2450
2451    /// Start a child run under `parent_run_id` at `depth`, so the tree records
2452    /// who spawned whom. Returns the child's run id.
2453    pub fn start_child_run(
2454        &self,
2455        goal: &str,
2456        file: &str,
2457        parent_run_id: i64,
2458        depth: u32,
2459    ) -> Result<i64> {
2460        self.conn.execute(
2461            "INSERT INTO runs (goal, file, parent_run_id, depth, status, started_at)
2462             VALUES (?1, ?2, ?3, ?4, 'running', strftime('%Y-%m-%dT%H:%M:%fZ','now'))",
2463            (goal, file, parent_run_id, depth),
2464        )?;
2465        Ok(self.conn.last_insert_rowid())
2466    }
2467
2468    /// Record a spawn, a spawn refusal, or a budget draw against the tree.
2469    pub fn record_agent_event(&self, e: &AgentEvent) -> Result<()> {
2470        self.conn.execute(
2471            "INSERT INTO agent_events (run_id, step, kind, child_run_id, detail, tokens, remaining)
2472             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
2473            (
2474                e.run_id,
2475                e.step,
2476                &e.kind,
2477                e.child_run_id,
2478                &e.detail,
2479                e.tokens,
2480                e.remaining,
2481            ),
2482        )?;
2483        Ok(())
2484    }
2485
2486    /// Every agent event recorded for a run, in order.
2487    pub fn agent_events(&self, run_id: i64) -> Result<Vec<AgentEvent>> {
2488        let mut stmt = self.conn.prepare(
2489            "SELECT run_id, step, kind, child_run_id, detail, tokens, remaining
2490             FROM agent_events WHERE run_id = ?1 ORDER BY id ASC",
2491        )?;
2492        let rows = stmt.query_map([run_id], |r| {
2493            Ok(AgentEvent {
2494                run_id: r.get(0)?,
2495                step: r.get::<_, i64>(1)? as u32,
2496                kind: r.get(2)?,
2497                child_run_id: r.get(3)?,
2498                detail: r.get(4)?,
2499                tokens: r.get::<_, Option<i64>>(5)?.map(|n| n as u64),
2500                remaining: r.get::<_, Option<i64>>(6)?.map(|n| n as u64),
2501            })
2502        })?;
2503        Ok(rows.collect::<std::result::Result<_, _>>()?)
2504    }
2505
2506    /// Record one sandbox lifecycle event against a run.
2507    pub fn record_sandbox_event(&self, e: &SandboxEvent) -> Result<()> {
2508        self.conn.execute(
2509            "INSERT INTO sandbox_events (run_id, step, kind, backend, detail)
2510             VALUES (?1, ?2, ?3, ?4, ?5)",
2511            (e.run_id, e.step, &e.kind, &e.backend, &e.detail),
2512        )?;
2513        Ok(())
2514    }
2515
2516    /// Record one MCP event.
2517    pub fn record_mcp(&self, run_id: i64, e: &McpEvent) -> Result<()> {
2518        self.conn.execute(
2519            "INSERT INTO mcp_events (run_id, step, kind, server, tool, ok, millis, detail)
2520             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
2521            (
2522                run_id,
2523                e.step,
2524                &e.kind,
2525                &e.server,
2526                &e.tool,
2527                e.ok,
2528                e.millis.map(|m| m as i64),
2529                &e.detail,
2530            ),
2531        )?;
2532        Ok(())
2533    }
2534
2535    /// Every MCP event recorded for a run, in order.
2536    pub fn mcp_events(&self, run_id: i64) -> Result<Vec<McpEvent>> {
2537        let mut stmt = self.conn.prepare(
2538            "SELECT step, kind, server, tool, ok, millis, detail
2539             FROM mcp_events WHERE run_id = ?1 ORDER BY id ASC",
2540        )?;
2541        let rows = stmt.query_map([run_id], |r| {
2542            Ok(McpEvent {
2543                step: r.get::<_, i64>(0)? as u32,
2544                kind: r.get(1)?,
2545                server: r.get(2)?,
2546                tool: r.get(3)?,
2547                ok: r.get(4)?,
2548                millis: r.get::<_, Option<i64>>(5)?.map(|m| m as u64),
2549                detail: r.get(6)?,
2550            })
2551        })?;
2552        Ok(rows.collect::<std::result::Result<_, _>>()?)
2553    }
2554
2555    /// Record one context-assembly event against a run.
2556    pub fn record_context_event(&self, run_id: i64, e: &ContextEvent) -> Result<()> {
2557        self.conn.execute(
2558            "INSERT INTO context_events (run_id, step, kind, detail, est_tokens, reported_tokens)
2559             VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
2560            (
2561                run_id,
2562                e.step,
2563                &e.kind,
2564                &e.detail,
2565                e.est_tokens.map(|n| n as i64),
2566                e.reported_tokens.map(|n| n as i64),
2567            ),
2568        )?;
2569        Ok(())
2570    }
2571
2572    /// Fill in what the provider said one turn's request cost, once the
2573    /// completion has returned. The estimate is left as it was: the pair is the
2574    /// point — one row carries both numbers, so drift is readable.
2575    pub fn record_context_reported(&self, run_id: i64, step: u32, reported: u64) -> Result<()> {
2576        self.conn.execute(
2577            "UPDATE context_events SET reported_tokens = ?1
2578             WHERE run_id = ?2 AND step = ?3 AND kind = 'assembled'",
2579            (reported as i64, run_id, step),
2580        )?;
2581        Ok(())
2582    }
2583
2584    /// Every context-assembly event recorded for a run, in order.
2585    pub fn context_events(&self, run_id: i64) -> Result<Vec<ContextEvent>> {
2586        let mut stmt = self.conn.prepare(
2587            "SELECT step, kind, detail, est_tokens, reported_tokens
2588             FROM context_events WHERE run_id = ?1 ORDER BY id ASC",
2589        )?;
2590        let rows = stmt.query_map([run_id], |r| {
2591            Ok(ContextEvent {
2592                step: r.get::<_, i64>(0)? as u32,
2593                kind: r.get(1)?,
2594                detail: r.get(2)?,
2595                est_tokens: r.get::<_, Option<i64>>(3)?.map(|n| n as u64),
2596                reported_tokens: r.get::<_, Option<i64>>(4)?.map(|n| n as u64),
2597            })
2598        })?;
2599        Ok(rows.collect::<std::result::Result<_, _>>()?)
2600    }
2601
2602    /// Record one call to a provider (0.18.0).
2603    ///
2604    /// Called once per attempt, by the run loop, for a call that answered and
2605    /// for one that failed alike. See [`ProviderCall`] for why the failures are
2606    /// kept.
2607    pub fn record_provider_call(&self, run_id: i64, call: &ProviderCall) -> Result<()> {
2608        let u = call.usage;
2609        self.conn.execute(
2610            "INSERT INTO provider_calls
2611                 (run_id, step, attempt, provider, model, prompt_tokens, completion_tokens,
2612                  total_tokens, cache_read_tokens, cache_write_tokens, reasoning_tokens,
2613                  server_tool_requests, latency_ms, ttft_ms, finish_reason, failure)
2614             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16)",
2615            rusqlite::params![
2616                run_id,
2617                call.step,
2618                call.attempt,
2619                &call.provider,
2620                &call.model,
2621                u.map(|u| u.prompt_tokens),
2622                u.map(|u| u.completion_tokens),
2623                u.map(|u| u.total_tokens),
2624                u.map(|u| u.cache_read_tokens),
2625                u.map(|u| u.cache_write_tokens),
2626                u.map(|u| u.reasoning_tokens),
2627                u.map(|u| u.server_tool_requests),
2628                call.latency_ms,
2629                call.ttft_ms,
2630                &call.finish_reason,
2631                &call.failure,
2632            ],
2633        )?;
2634        Ok(())
2635    }
2636
2637    /// Every provider call recorded for a run, in the order they were made.
2638    ///
2639    /// A run that predates 0.18.0 has no rows, and this returns an empty vector
2640    /// rather than zeros — an unrecorded run and a free one are different facts.
2641    pub fn provider_calls(&self, run_id: i64) -> Result<Vec<ProviderCall>> {
2642        let mut stmt = self.conn.prepare(
2643            "SELECT step, attempt, provider, model, prompt_tokens, completion_tokens,
2644                    total_tokens, cache_read_tokens, cache_write_tokens, reasoning_tokens,
2645                    server_tool_requests, latency_ms, ttft_ms, finish_reason, failure
2646             FROM provider_calls WHERE run_id = ?1 ORDER BY id",
2647        )?;
2648        let rows = stmt.query_map([run_id], |r| {
2649            // `total_tokens` decides whether the provider reported anything at
2650            // all: a NULL there is the `None` the caller stored, and reading it
2651            // back as a zeroed `Usage` would turn "unknown" into "free".
2652            let total: Option<u64> = r.get(6)?;
2653            Ok(ProviderCall {
2654                step: r.get(0)?,
2655                attempt: r.get(1)?,
2656                provider: r.get(2)?,
2657                model: r.get(3)?,
2658                usage: match total {
2659                    Some(total_tokens) => Some(Usage {
2660                        prompt_tokens: r.get::<_, Option<u64>>(4)?.unwrap_or(0),
2661                        completion_tokens: r.get::<_, Option<u64>>(5)?.unwrap_or(0),
2662                        total_tokens,
2663                        cache_read_tokens: r.get::<_, Option<u64>>(7)?.unwrap_or(0),
2664                        cache_write_tokens: r.get::<_, Option<u64>>(8)?.unwrap_or(0),
2665                        reasoning_tokens: r.get::<_, Option<u64>>(9)?.unwrap_or(0),
2666                        server_tool_requests: r.get::<_, Option<u64>>(10)?.unwrap_or(0),
2667                    }),
2668                    None => None,
2669                },
2670                latency_ms: r.get(11)?,
2671                ttft_ms: r.get(12)?,
2672                finish_reason: r.get(13)?,
2673                failure: r.get(14)?,
2674            })
2675        })?;
2676        Ok(rows.collect::<std::result::Result<_, _>>()?)
2677    }
2678
2679    /// Every provider call in the store, with the run and the day it belongs to.
2680    ///
2681    /// The grouped views are built from this one read: pricing is arithmetic the
2682    /// database cannot do, so the rows come back and the grouping happens in
2683    /// Rust rather than in half-SQL that would still need a second pass.
2684    fn all_provider_calls(&self) -> Result<Vec<(i64, String, ProviderCall)>> {
2685        let mut stmt = self.conn.prepare(
2686            "SELECT run_id, date(at), step, attempt, provider, model, prompt_tokens,
2687                    completion_tokens, total_tokens, cache_read_tokens, cache_write_tokens,
2688                    reasoning_tokens, server_tool_requests, latency_ms, ttft_ms, finish_reason,
2689                    failure
2690             FROM provider_calls ORDER BY id",
2691        )?;
2692        let rows = stmt.query_map([], |r| {
2693            let total: Option<u64> = r.get(8)?;
2694            Ok((
2695                r.get::<_, i64>(0)?,
2696                r.get::<_, String>(1)?,
2697                ProviderCall {
2698                    step: r.get(2)?,
2699                    attempt: r.get(3)?,
2700                    provider: r.get(4)?,
2701                    model: r.get(5)?,
2702                    usage: match total {
2703                        Some(total_tokens) => Some(Usage {
2704                            prompt_tokens: r.get::<_, Option<u64>>(6)?.unwrap_or(0),
2705                            completion_tokens: r.get::<_, Option<u64>>(7)?.unwrap_or(0),
2706                            total_tokens,
2707                            cache_read_tokens: r.get::<_, Option<u64>>(9)?.unwrap_or(0),
2708                            cache_write_tokens: r.get::<_, Option<u64>>(10)?.unwrap_or(0),
2709                            reasoning_tokens: r.get::<_, Option<u64>>(11)?.unwrap_or(0),
2710                            server_tool_requests: r.get::<_, Option<u64>>(12)?.unwrap_or(0),
2711                        }),
2712                        None => None,
2713                    },
2714                    latency_ms: r.get(13)?,
2715                    ttft_ms: r.get(14)?,
2716                    finish_reason: r.get(15)?,
2717                    failure: r.get(16)?,
2718                },
2719            ))
2720        })?;
2721        Ok(rows.collect::<std::result::Result<_, _>>()?)
2722    }
2723
2724    /// Spend grouped by the model that served each call, priced by `prices`
2725    /// (0.18.0).
2726    ///
2727    /// Calls whose provider named no model group under `"(unknown model)"` and
2728    /// are counted in [`Spend::unpriced_calls`], because attributing them to
2729    /// anything else would be a guess.
2730    ///
2731    /// ```
2732    /// use io_harness::pricing::{Price, PriceTable};
2733    /// use io_harness::{ProviderCall, Store, Usage};
2734    ///
2735    /// # fn main() -> io_harness::Result<()> {
2736    /// # let store = Store::memory()?;
2737    /// # let run_id = store.start_run("goal", "NOTES.md")?;
2738    /// # store.record_provider_call(run_id, &ProviderCall {
2739    /// #     step: 1, provider: "anthropic".into(), model: Some("m".into()),
2740    /// #     usage: Some(Usage { prompt_tokens: 1_000_000, total_tokens: 1_000_000,
2741    /// #                         ..Default::default() }), ..Default::default() })?;
2742    /// let cheap = PriceTable::new("2026-07-29").with("m", Price { input: 1_000_000, ..Price::ZERO });
2743    /// let dear = PriceTable::new("2026-07-29").with("m", Price { input: 2_000_000, ..Price::ZERO });
2744    ///
2745    /// // The same unchanged trace, two price tables, two answers — which is what
2746    /// // "correcting a price repairs the whole history" means in practice.
2747    /// assert_eq!(store.spend_by_model(&cheap)?[0].cost_micros, 1_000_000);
2748    /// assert_eq!(store.spend_by_model(&dear)?[0].cost_micros, 2_000_000);
2749    /// # Ok(())
2750    /// # }
2751    /// ```
2752    pub fn spend_by_model(&self, prices: &PriceTable) -> Result<Vec<Spend>> {
2753        self.grouped(prices, |_, _, call| {
2754            call.model.clone().unwrap_or_else(|| UNKNOWN_MODEL.into())
2755        })
2756    }
2757
2758    /// Spend grouped by day (`YYYY-MM-DD`, UTC, from the database clock), priced
2759    /// by `prices` (0.18.0).
2760    pub fn spend_by_day(&self, prices: &PriceTable) -> Result<Vec<Spend>> {
2761        self.grouped(prices, |_, day, _| day.to_string())
2762    }
2763
2764    /// Spend grouped by run id, priced by `prices` (0.18.0).
2765    pub fn spend_by_run(&self, prices: &PriceTable) -> Result<Vec<Spend>> {
2766        self.grouped(prices, |run_id, _, _| run_id.to_string())
2767    }
2768
2769    /// The shared body of the three groupings: read once, key by `key`, sum and
2770    /// price each group. Rows come back ordered by key, which is the only
2771    /// ordering promised.
2772    fn grouped(
2773        &self,
2774        prices: &PriceTable,
2775        key: impl Fn(i64, &str, &ProviderCall) -> String,
2776    ) -> Result<Vec<Spend>> {
2777        let calls = self.all_provider_calls()?;
2778        let mut groups: std::collections::BTreeMap<String, Vec<&ProviderCall>> =
2779            std::collections::BTreeMap::new();
2780        for (run_id, day, call) in &calls {
2781            groups
2782                .entry(key(*run_id, day, call))
2783                .or_default()
2784                .push(call);
2785        }
2786        Ok(groups
2787            .into_iter()
2788            .map(|(k, calls)| crate::pricing::group(k, &calls, prices))
2789            .collect())
2790    }
2791
2792    /// Record one file change and its line counts (0.18.0).
2793    pub fn record_edit(&self, run_id: i64, edit: &Edit) -> Result<()> {
2794        self.conn.execute(
2795            "INSERT INTO edits (run_id, step, tool, path, lines_added, lines_removed)
2796             VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
2797            (
2798                run_id,
2799                edit.step,
2800                &edit.tool,
2801                &edit.path,
2802                edit.lines_added,
2803                edit.lines_removed,
2804            ),
2805        )?;
2806        Ok(())
2807    }
2808
2809    /// Every file change recorded for a run, in the order they were made.
2810    pub fn edits(&self, run_id: i64) -> Result<Vec<Edit>> {
2811        let mut stmt = self.conn.prepare(
2812            "SELECT step, tool, path, lines_added, lines_removed
2813             FROM edits WHERE run_id = ?1 ORDER BY id",
2814        )?;
2815        let rows = stmt.query_map([run_id], |r| {
2816            Ok(Edit {
2817                step: r.get(0)?,
2818                tool: r.get(1)?,
2819                path: r.get(2)?,
2820                lines_added: r.get(3)?,
2821                lines_removed: r.get(4)?,
2822            })
2823        })?;
2824        Ok(rows.collect::<std::result::Result<_, _>>()?)
2825    }
2826
2827    /// Record the policy a run was started under.
2828    ///
2829    /// `INSERT OR REPLACE`, like every other per-run row, so recording twice for
2830    /// one run — a resume that re-states its boundary — replaces rather than
2831    /// duplicates or fails.
2832    pub fn record_run_policy(&self, run_id: i64, policy: &Policy) -> Result<()> {
2833        self.conn.execute(
2834            "INSERT OR REPLACE INTO run_policies (run_id, policy) VALUES (?1, ?2)",
2835            (
2836                run_id,
2837                serde_json::to_string(policy).expect("a Policy is always serialisable"),
2838            ),
2839        )?;
2840        Ok(())
2841    }
2842
2843    /// The policy a run was started under, or `None` if none was recorded.
2844    ///
2845    /// `None` is not [`Policy::permissive`] and must never be read as it: a run
2846    /// written by 0.12.0 has no row at all, so the honest answer is "nobody
2847    /// recorded what the boundary was", not "the caller chose to enforce
2848    /// nothing". A caller that needs a policy either way has to decide which to
2849    /// assume, and it should decide that knowingly.
2850    /// Unlike the other getters in this file, a failed read is an error rather
2851    /// than `None`. They can fold the two together because a missing memory
2852    /// entry and an unreadable one lead to the same recovery; here they do not.
2853    /// `None` is what tells [`crate::resume`] the run had no boundary and may be
2854    /// resumed permissively, so a disk error that read as `None` would hand a
2855    /// policy-bearing run an agent with no policy — silently, and by exactly the
2856    /// route this table exists to close.
2857    pub fn run_policy(&self, run_id: i64) -> Result<Option<Policy>> {
2858        let json: Option<String> = match self.conn.query_row(
2859            "SELECT policy FROM run_policies WHERE run_id = ?1",
2860            [run_id],
2861            |r| r.get(0),
2862        ) {
2863            Ok(json) => Some(json),
2864            Err(rusqlite::Error::QueryReturnedNoRows) => None,
2865            Err(e) => return Err(e.into()),
2866        };
2867        json.map(|j| {
2868            serde_json::from_str(&j).map_err(|e| Error::Resume {
2869                reason: format!("run {run_id} has an unreadable recorded policy: {e}"),
2870            })
2871        })
2872        .transpose()
2873    }
2874
2875    /// Append observations to a run's durable ledger, in one transaction.
2876    ///
2877    /// Called once at a committed step boundary rather than once per
2878    /// observation: the step is the unit the rest of the checkpoint works in, and
2879    /// an observation belonging to a step that never committed must not survive a
2880    /// crash the step itself did not survive.
2881    pub fn record_observations(&self, run_id: i64, entries: &[Observation]) -> Result<()> {
2882        if entries.is_empty() {
2883            return Ok(());
2884        }
2885        let tx = self.conn.unchecked_transaction()?;
2886        {
2887            let mut stmt = tx.prepare(
2888                "INSERT INTO ledger_observations (run_id, step, kind, target, text)
2889                 VALUES (?1, ?2, ?3, ?4, ?5)",
2890            )?;
2891            for e in entries {
2892                stmt.execute((run_id, e.step as i64, kind_wire(e.kind), &e.target, &e.text))?;
2893            }
2894        }
2895        tx.commit()?;
2896        Ok(())
2897    }
2898
2899    /// A run's durable ledger, in the order it was observed.
2900    ///
2901    /// Empty for a run that recorded nothing and for a run written before 0.13.0
2902    /// — the two are the same to a reader, and both mean "there is nothing to
2903    /// restore", which is 0.12.0's behaviour and not a lie about it.
2904    pub fn observations(&self, run_id: i64) -> Result<Vec<Observation>> {
2905        let mut stmt = self.conn.prepare(
2906            "SELECT step, kind, target, text
2907             FROM ledger_observations WHERE run_id = ?1 ORDER BY id ASC",
2908        )?;
2909        let rows = stmt.query_map([run_id], |r| {
2910            Ok((
2911                r.get::<_, i64>(0)? as u32,
2912                r.get::<_, String>(1)?,
2913                r.get::<_, Option<String>>(2)?,
2914                r.get::<_, String>(3)?,
2915            ))
2916        })?;
2917        let mut out = Vec::new();
2918        for row in rows {
2919            let (step, kind, target, text) = row?;
2920            out.push(Observation::new(
2921                step,
2922                kind_from_wire(&kind, run_id)?,
2923                target,
2924                text,
2925            ));
2926        }
2927        Ok(out)
2928    }
2929
2930    /// Every sandbox event recorded for a run, in order.
2931    pub fn sandbox_events(&self, run_id: i64) -> Result<Vec<SandboxEvent>> {
2932        let mut stmt = self.conn.prepare(
2933            "SELECT run_id, step, kind, backend, detail
2934             FROM sandbox_events WHERE run_id = ?1 ORDER BY id ASC",
2935        )?;
2936        let rows = stmt.query_map([run_id], |r| {
2937            Ok(SandboxEvent {
2938                run_id: r.get(0)?,
2939                step: r.get::<_, i64>(1)? as u32,
2940                kind: r.get(2)?,
2941                backend: r.get(3)?,
2942                detail: r.get(4)?,
2943            })
2944        })?;
2945        Ok(rows.collect::<std::result::Result<_, _>>()?)
2946    }
2947
2948    /// The run ids of the direct children of `run_id`, in spawn order.
2949    pub fn children(&self, run_id: i64) -> Result<Vec<i64>> {
2950        let mut stmt = self
2951            .conn
2952            .prepare("SELECT id FROM runs WHERE parent_run_id = ?1 ORDER BY id ASC")?;
2953        let rows = stmt.query_map([run_id], |r| r.get(0))?;
2954        Ok(rows.collect::<std::result::Result<_, _>>()?)
2955    }
2956
2957    /// The parent run id of `run_id`, or `None` for a root run.
2958    pub fn parent(&self, run_id: i64) -> Result<Option<i64>> {
2959        Ok(self.conn.query_row(
2960            "SELECT parent_run_id FROM runs WHERE id = ?1",
2961            [run_id],
2962            |r| r.get(0),
2963        )?)
2964    }
2965
2966    /// The nesting depth recorded for a run (0 at the root).
2967    pub fn depth(&self, run_id: i64) -> Result<u32> {
2968        let d: i64 =
2969            self.conn
2970                .query_row("SELECT depth FROM runs WHERE id = ?1", [run_id], |r| {
2971                    r.get(0)
2972                })?;
2973        Ok(d as u32)
2974    }
2975
2976    /// Record one step's full trace entry.
2977    pub fn record(&self, run_id: i64, step: &StepRecord) -> Result<()> {
2978        self.conn.execute(
2979            "INSERT INTO steps (run_id, step, decision, result, prompt, tool_call, tokens)
2980             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
2981            (
2982                run_id,
2983                step.step,
2984                &step.decision,
2985                &step.result,
2986                &step.prompt,
2987                &step.tool_call,
2988                step.tokens,
2989            ),
2990        )?;
2991        Ok(())
2992    }
2993
2994    /// Durably checkpoint one completed step: the step's trace row and its
2995    /// checkpoint event are written in a single transaction, so a crash leaves
2996    /// either both (the step is done) or neither (it replays) — never a torn
2997    /// half. The committed checkpoint is the step's completion marker.
2998    pub fn checkpoint_step(&self, run_id: i64, step: &StepRecord) -> Result<()> {
2999        let tx = self.conn.unchecked_transaction()?;
3000        tx.execute(
3001            "INSERT INTO steps (run_id, step, decision, result, prompt, tool_call, tokens)
3002             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
3003            (
3004                run_id,
3005                step.step,
3006                &step.decision,
3007                &step.result,
3008                &step.prompt,
3009                &step.tool_call,
3010                step.tokens,
3011            ),
3012        )?;
3013        tx.execute(
3014            "INSERT INTO checkpoint_events (run_id, step, kind, detail)
3015             VALUES (?1, ?2, 'checkpoint', NULL)",
3016            (run_id, step.step),
3017        )?;
3018        tx.commit()?;
3019        Ok(())
3020    }
3021
3022    /// Record a checkpoint/resume/skipped event on its own (not tied to a step
3023    /// commit) — used for resume and skip markers.
3024    pub fn record_checkpoint_event(&self, e: &CheckpointEvent) -> Result<()> {
3025        self.conn.execute(
3026            "INSERT INTO checkpoint_events (run_id, step, kind, detail) VALUES (?1, ?2, ?3, ?4)",
3027            (e.run_id, e.step, &e.kind, &e.detail),
3028        )?;
3029        Ok(())
3030    }
3031
3032    /// Every checkpoint-lifecycle event recorded for a run, in order.
3033    pub fn checkpoint_events(&self, run_id: i64) -> Result<Vec<CheckpointEvent>> {
3034        let mut stmt = self.conn.prepare(
3035            "SELECT run_id, step, kind, detail
3036             FROM checkpoint_events WHERE run_id = ?1 ORDER BY id ASC",
3037        )?;
3038        let rows = stmt.query_map([run_id], |r| {
3039            Ok(CheckpointEvent {
3040                run_id: r.get(0)?,
3041                step: r.get::<_, i64>(1)? as u32,
3042                kind: r.get(2)?,
3043                detail: r.get(3)?,
3044            })
3045        })?;
3046        Ok(rows.collect::<std::result::Result<_, _>>()?)
3047    }
3048
3049    /// Set the durable run status (`running`, `paused`, `completed`, `failed`).
3050    pub fn set_status(&self, run_id: i64, status: &str) -> Result<()> {
3051        self.conn.execute(
3052            "UPDATE runs SET status = ?1 WHERE id = ?2",
3053            (status, run_id),
3054        )?;
3055        Ok(())
3056    }
3057
3058    /// The durable run status, if the run exists.
3059    pub fn status(&self, run_id: i64) -> Result<Option<String>> {
3060        Ok(self
3061            .conn
3062            .query_row("SELECT status FROM runs WHERE id = ?1", [run_id], |r| {
3063                r.get(0)
3064            })
3065            .ok())
3066    }
3067
3068    /// Real wall-clock seconds elapsed since the run's `started_at`, from the
3069    /// database clock — so a budget over duration counts time that passed while
3070    /// the process was down, not just this process's uptime. Zero if the run has
3071    /// no start stamp (a pre-0.7.0 run).
3072    pub fn elapsed_secs(&self, run_id: i64) -> Result<f64> {
3073        let secs: Option<f64> = self.conn.query_row(
3074            "SELECT (julianday('now') - julianday(started_at)) * 86400.0
3075             FROM runs WHERE id = ?1",
3076            [run_id],
3077            |r| r.get(0),
3078        )?;
3079        Ok(secs.unwrap_or(0.0).max(0.0))
3080    }
3081
3082    /// Total tokens recorded across this run's steps — the durable spend, so a
3083    /// resume restores the token budget instead of restarting it at zero.
3084    pub fn spent_tokens(&self, run_id: i64) -> Result<u64> {
3085        let n: i64 = self.conn.query_row(
3086            "SELECT COALESCE(SUM(tokens), 0) FROM steps WHERE run_id = ?1",
3087            [run_id],
3088            |r| r.get(0),
3089        )?;
3090        Ok(n as u64)
3091    }
3092
3093    /// Every run id in the tree rooted at `root` (the root plus all descendants),
3094    /// via the `parent_run_id` edge — the set a tree-level resume re-drives.
3095    pub fn tree_run_ids(&self, root: i64) -> Result<Vec<i64>> {
3096        let mut stmt = self.conn.prepare(
3097            "WITH RECURSIVE tree(id) AS (
3098                 SELECT id FROM runs WHERE id = ?1
3099                 UNION ALL
3100                 SELECT r.id FROM runs r JOIN tree t ON r.parent_run_id = t.id
3101             )
3102             SELECT id FROM tree ORDER BY id ASC",
3103        )?;
3104        let rows = stmt.query_map([root], |r| r.get(0))?;
3105        Ok(rows.collect::<std::result::Result<_, _>>()?)
3106    }
3107
3108    /// Total tokens spent across the whole tree rooted at `root` — the durable
3109    /// aggregate-ledger spend restored on a tree resume.
3110    pub fn spent_tokens_tree(&self, root: i64) -> Result<u64> {
3111        let n: i64 = self.conn.query_row(
3112            "WITH RECURSIVE tree(id) AS (
3113                 SELECT id FROM runs WHERE id = ?1
3114                 UNION ALL
3115                 SELECT r.id FROM runs r JOIN tree t ON r.parent_run_id = t.id
3116             )
3117             SELECT COALESCE(SUM(s.tokens), 0)
3118             FROM steps s JOIN tree ON s.run_id = tree.id",
3119            [root],
3120            |r| r.get(0),
3121        )?;
3122        Ok(n as u64)
3123    }
3124
3125    /// Number of agents (run rows) in the tree rooted at `root` — the durable
3126    /// agent count restored on a tree resume.
3127    pub fn agent_count_tree(&self, root: i64) -> Result<u32> {
3128        let n: i64 = self.conn.query_row(
3129            "WITH RECURSIVE tree(id) AS (
3130                 SELECT id FROM runs WHERE id = ?1
3131                 UNION ALL
3132                 SELECT r.id FROM runs r JOIN tree t ON r.parent_run_id = t.id
3133             )
3134             SELECT COUNT(*) FROM tree",
3135            [root],
3136            |r| r.get(0),
3137        )?;
3138        Ok(n as u32)
3139    }
3140
3141    /// Persist a spawned child's contract so a crashed tree can rebuild and
3142    /// resume that exact child on resume instead of spawning a duplicate. Keyed
3143    /// by (parent, step, goal) so a replayed spawn step adopts the existing child.
3144    #[allow(clippy::too_many_arguments)]
3145    pub fn record_spawn(
3146        &self,
3147        parent_run_id: i64,
3148        step: u32,
3149        child_run_id: i64,
3150        goal: &str,
3151        verify_file: &str,
3152        needle: &str,
3153        max_steps: Option<u32>,
3154        deny_write_json: &str,
3155    ) -> Result<()> {
3156        self.conn.execute(
3157            "INSERT INTO spawns
3158                 (parent_run_id, step, child_run_id, goal, verify_file, needle, max_steps, deny_write)
3159             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
3160            (
3161                parent_run_id,
3162                step,
3163                child_run_id,
3164                goal,
3165                verify_file,
3166                needle,
3167                max_steps,
3168                deny_write_json,
3169            ),
3170        )?;
3171        Ok(())
3172    }
3173
3174    /// Find the child spawned by `parent_run_id` at `step` for `goal`, if any —
3175    /// the adopt-on-resume lookup that makes a replayed spawn step idempotent.
3176    pub fn find_spawn(
3177        &self,
3178        parent_run_id: i64,
3179        step: u32,
3180        goal: &str,
3181    ) -> Result<Option<SpawnRow>> {
3182        Ok(self
3183            .conn
3184            .query_row(
3185                "SELECT child_run_id, goal, verify_file, needle, max_steps, deny_write
3186                 FROM spawns WHERE parent_run_id = ?1 AND step = ?2 AND goal = ?3
3187                 ORDER BY id ASC LIMIT 1",
3188                (parent_run_id, step, goal),
3189                |r| {
3190                    Ok(SpawnRow {
3191                        child_run_id: r.get(0)?,
3192                        goal: r.get(1)?,
3193                        verify_file: r.get(2)?,
3194                        needle: r.get(3)?,
3195                        max_steps: r.get::<_, Option<i64>>(4)?.map(|n| n as u32),
3196                        deny_write: r.get(5)?,
3197                    })
3198                },
3199            )
3200            .ok())
3201    }
3202
3203    /// Check a run can be resumed from its checkpoint, or return a typed
3204    /// [`Error::Resume`]. Refuses a store written by a newer checkpoint format
3205    /// (rather than misreading a layout it does not understand) and a run id that
3206    /// does not exist. An already-`completed` run is resumable as a no-op, so it
3207    /// is not refused here.
3208    pub fn check_resumable(&self, run_id: i64) -> Result<()> {
3209        let format: i64 = self
3210            .conn
3211            .query_row("PRAGMA user_version", [], |r| r.get(0))?;
3212        if format > CHECKPOINT_FORMAT {
3213            return Err(Error::Resume {
3214                reason: format!(
3215                    "checkpoint format {format} is newer than supported {CHECKPOINT_FORMAT}; \
3216                     upgrade io-harness to resume this run"
3217                ),
3218            });
3219        }
3220        let exists: bool = self
3221            .conn
3222            .query_row("SELECT 1 FROM runs WHERE id = ?1", [run_id], |_| Ok(true))
3223            .unwrap_or(false);
3224        if !exists {
3225            return Err(Error::Resume {
3226                reason: format!("no run with id {run_id} in the store"),
3227            });
3228        }
3229        Ok(())
3230    }
3231
3232    /// Record which provider ran this run, for the audit trace.
3233    pub fn set_provider(&self, run_id: i64, provider: &str) -> Result<()> {
3234        self.conn.execute(
3235            "UPDATE runs SET provider = ?1 WHERE id = ?2",
3236            (provider, run_id),
3237        )?;
3238        Ok(())
3239    }
3240
3241    /// The provider recorded for a run, if any.
3242    pub fn provider(&self, run_id: i64) -> Result<Option<String>> {
3243        Ok(self
3244            .conn
3245            .query_row("SELECT provider FROM runs WHERE id = ?1", [run_id], |r| {
3246                r.get(0)
3247            })?)
3248    }
3249
3250    /// Record the run's final outcome, and derive the durable status from it:
3251    /// `success` completes the run, `awaiting_approval` pauses it, any other
3252    /// terminal outcome completes it (finished, just not with success). A run
3253    /// that crashed mid-loop never reaches here, so it stays `running` and is
3254    /// resumable.
3255    pub fn finish_run(&self, run_id: i64, outcome: &str) -> Result<()> {
3256        let status = match outcome {
3257            "awaiting_approval" => "paused",
3258            _ => "completed",
3259        };
3260        self.conn.execute(
3261            "UPDATE runs SET outcome = ?1, status = ?2 WHERE id = ?3",
3262            (outcome, status, run_id),
3263        )?;
3264        // A paused run has not finished — it is waiting for a human and will be
3265        // resumed — so it gets no summary yet. It gets one when it really ends.
3266        if status == "completed" {
3267            self.write_summary(run_id, outcome)?;
3268        }
3269        Ok(())
3270    }
3271
3272    /// Record the run's outcome summary. Called by [`Self::finish_run`].
3273    ///
3274    /// Written here rather than assembled by the caller because a run that
3275    /// escalates or is refused returns `Err` and never reaches a
3276    /// [`RunResult`](crate::RunResult) at all — so a summary built at the call
3277    /// site would be missing for exactly the endings a scoring tool most wants to
3278    /// count.
3279    fn write_summary(&self, run_id: i64, outcome: &str) -> Result<()> {
3280        // Both stamps from the database clock, like `started_at`. Mixing SQLite's
3281        // clock with the process's would make the difference meaningless.
3282        let (finished_at, duration_ms): (String, Option<f64>) = self.conn.query_row(
3283            "SELECT strftime('%Y-%m-%dT%H:%M:%fZ','now'),
3284                    (julianday('now') - julianday(started_at)) * 86400000.0
3285             FROM runs WHERE id = ?1",
3286            [run_id],
3287            |r| Ok((r.get(0)?, r.get(1)?)),
3288        )?;
3289        let duration_ms = duration_ms.map(|ms| ms.max(0.0) as u64);
3290        // `INSERT OR REPLACE`, because `finish_run` is reachable more than once for
3291        // one run: a paused run resumes and finishes, and a resume of an already
3292        // finished run is documented as idempotent. The last ending is the true one.
3293        self.conn.execute(
3294            "INSERT OR REPLACE INTO run_outcomes
3295                 (run_id, outcome, success, steps, tokens, duration_ms, finished_at)
3296             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
3297            (
3298                run_id,
3299                outcome,
3300                i64::from(outcome == SUCCESS_OUTCOME),
3301                self.last_step(run_id)?,
3302                self.spent_tokens(run_id)?,
3303                duration_ms,
3304                &finished_at,
3305            ),
3306        )?;
3307        Ok(())
3308    }
3309
3310    /// What a finished run cost and whether it worked.
3311    ///
3312    /// `None` if the run has not finished, is paused awaiting a human, or was
3313    /// finished by a pre-0.12.0 binary — a missing summary is reported as absent
3314    /// rather than as a row of zeroes, which would be indistinguishable from a run
3315    /// that did nothing.
3316    pub fn run_summary(&self, run_id: i64) -> Result<Option<RunSummary>> {
3317        let mut q = self.conn.prepare(
3318            "SELECT run_id, outcome, success, steps, tokens, duration_ms, finished_at
3319             FROM run_outcomes WHERE run_id = ?1",
3320        )?;
3321        let mut rows = q.query_map([run_id], |r| {
3322            Ok(RunSummary {
3323                run_id: r.get(0)?,
3324                outcome: r.get(1)?,
3325                success: r.get::<_, i64>(2)? != 0,
3326                steps: r.get(3)?,
3327                tokens: r.get(4)?,
3328                duration_ms: r.get(5)?,
3329                finished_at: r.get(6)?,
3330            })
3331        })?;
3332        match rows.next() {
3333            Some(row) => Ok(Some(row?)),
3334            None => Ok(None),
3335        }
3336    }
3337
3338    /// Every run in this store, newest first.
3339    ///
3340    /// Exists because an escalation returns `Err` rather than a
3341    /// [`RunResult`](crate::RunResult), so a caller whose run escalated has no
3342    /// `run_id` to resume with and therefore no way to reach
3343    /// [`RunOutcome::Escalated`](crate::RunOutcome::Escalated) — the outcome added
3344    /// for exactly that case. A caller who did not record the id before starting
3345    /// can find it here.
3346    pub fn runs(&self) -> Result<Vec<i64>> {
3347        let mut stmt = self.conn.prepare("SELECT id FROM runs ORDER BY id DESC")?;
3348        let rows = stmt.query_map([], |r| r.get(0))?;
3349        Ok(rows.collect::<std::result::Result<_, _>>()?)
3350    }
3351
3352    /// The most recently started run, if this store holds one.
3353    ///
3354    /// A convenience over [`Store::runs`] for the common single-run case. With
3355    /// concurrent runs in one store, "most recent" is by insertion order and a
3356    /// caller that cares should track its own ids.
3357    pub fn last_run(&self) -> Result<Option<i64>> {
3358        Ok(self.runs()?.into_iter().next())
3359    }
3360
3361    /// The recorded final outcome string of a run, if it has finished.
3362    pub fn outcome(&self, run_id: i64) -> Result<Option<String>> {
3363        Ok(self
3364            .conn
3365            .query_row("SELECT outcome FROM runs WHERE id = ?1", [run_id], |r| {
3366                r.get(0)
3367            })
3368            .ok()
3369            .flatten())
3370    }
3371
3372    /// The durable run status as a typed [`RunStatus`], if the run exists.
3373    pub fn run_status(&self, run_id: i64) -> Result<Option<RunStatus>> {
3374        Ok(self.status(run_id)?.map(|s| RunStatus::from_str(&s)))
3375    }
3376
3377    /// The highest step number recorded for a run, or 0 if none — the resume
3378    /// point for [`crate::resume`].
3379    pub fn last_step(&self, run_id: i64) -> Result<u32> {
3380        let n: i64 = self.conn.query_row(
3381            "SELECT COALESCE(MAX(step), 0) FROM steps WHERE run_id = ?1",
3382            [run_id],
3383            |r| r.get(0),
3384        )?;
3385        Ok(n as u32)
3386    }
3387
3388    /// Read every step of a run back, in order, as the full trace.
3389    /// The run's trace reduced to the part that two identical runs must match,
3390    /// as diffable text.
3391    ///
3392    /// This is the crate's definition of "the same run twice", and it exists
3393    /// because equality could not be row identity: `steps` has no
3394    /// `UNIQUE(run_id, step)` and a retry inserts its own row under the step
3395    /// number the eventual commit will reuse, so counting or comparing rows
3396    /// compares trace entries rather than agent behaviour.
3397    ///
3398    /// # What is compared
3399    ///
3400    /// Every `steps` row — step number, decision, result, prompt, tool call and
3401    /// tokens — and every `context_events` row's step, kind and detail. Between
3402    /// them these are what the agent was shown, what it decided, what it did, and
3403    /// what that cost.
3404    ///
3405    /// # What is excluded, and why
3406    ///
3407    /// Everything whose value is a fact about *this* execution rather than about
3408    /// the run:
3409    ///
3410    /// - **Wall-clock stamps** — `runs.started_at`, `memory.created_at`,
3411    ///   `run_outcomes.finished_at` and `duration_ms`. Two runs of the same case
3412    ///   take different amounts of time; that is not a divergence.
3413    /// - **`mcp_events.millis`** — a measured duration, for the same reason.
3414    /// - **`sandbox_events.detail`** — it carries the argv, and the argv carries
3415    ///   an ephemeral tempdir path that is different every run by design.
3416    /// - **Run and child ids** — `AUTOINCREMENT` values, meaningful only within
3417    ///   one store.
3418    ///
3419    /// Excluding a field is a decision that this crate cannot promise it, not a
3420    /// convenience. Anything added to this list should be added to this doc with
3421    /// its reason, because a comparison that quietly excludes what it cannot
3422    /// match is a comparison that asserts nothing.
3423    ///
3424    /// # What it assumes
3425    ///
3426    /// That each run being compared has its **own fresh store**. Run ids are
3427    /// excluded from the text, but a child agent's run id is embedded in the
3428    /// parent's composed observation (`[child 5 "goal" -> …]`), which is real
3429    /// content the model was shown. In a fresh store those ids start at 1 and are
3430    /// allocated in spawn order, so they match; in a shared store the second run's
3431    /// ids are higher and the traces differ for a reason that has nothing to do
3432    /// with the agent.
3433    ///
3434    /// Deterministic replay also requires the provider to answer identically —
3435    /// see [`Replay`](crate::provider::Replay) — and the same workspace state to
3436    /// start from.
3437    pub fn canonical_trace(&self, run_id: i64) -> Result<String> {
3438        let mut out = String::new();
3439        for s in self.steps(run_id)? {
3440            out.push_str(&format!(
3441                "step {} | tokens {} | decision {} | tool_call {} | prompt {} | result {}\n",
3442                s.step, s.tokens, s.decision, s.tool_call, s.prompt, s.result
3443            ));
3444        }
3445        for e in self.context_events(run_id)? {
3446            out.push_str(&format!(
3447                "context {} | {} | {}\n",
3448                e.step,
3449                e.kind,
3450                e.detail.as_deref().unwrap_or("")
3451            ));
3452        }
3453        Ok(out)
3454    }
3455
3456    pub fn steps(&self, run_id: i64) -> Result<Vec<StepRecord>> {
3457        let mut stmt = self.conn.prepare(
3458            "SELECT step, decision, result, prompt, tool_call, tokens
3459             FROM steps WHERE run_id = ?1 ORDER BY step ASC, id ASC",
3460        )?;
3461        let rows = stmt.query_map([run_id], |r| {
3462            Ok(StepRecord {
3463                step: r.get::<_, i64>(0)? as u32,
3464                decision: r.get(1)?,
3465                result: r.get(2)?,
3466                prompt: r.get(3)?,
3467                tool_call: r.get(4)?,
3468                tokens: r.get::<_, i64>(5)? as u64,
3469            })
3470        })?;
3471        Ok(rows.collect::<std::result::Result<_, _>>()?)
3472    }
3473
3474    // ---- 0.21.0: the question channel ----
3475
3476    /// Persist a question nobody in this process could answer, and return its id.
3477    ///
3478    /// The mirror of [`Self::put_pending`], deliberately: a question survives a
3479    /// process exit for exactly the reason a pending approval does, and the two stay
3480    /// in separate tables because they are separate things — one asks whether an act
3481    /// is permitted, the other what the operator meant.
3482    pub fn put_question(
3483        &self,
3484        run_id: i64,
3485        step: u32,
3486        q: &crate::approve::Question,
3487    ) -> Result<i64> {
3488        let choices = if q.choices.is_empty() {
3489            None
3490        } else {
3491            Some(serde_json::to_string(&q.choices).unwrap_or_default())
3492        };
3493        self.conn.execute(
3494            "INSERT INTO pending_questions (run_id, step, question, context, choices)
3495             VALUES (?1, ?2, ?3, ?4, ?5)",
3496            rusqlite::params![run_id, step, q.question, q.context, choices],
3497        )?;
3498        Ok(self.conn.last_insert_rowid())
3499    }
3500
3501    /// Read one question by id, answered or not.
3502    pub fn question(&self, question_id: i64) -> Result<Option<PendingQuestion>> {
3503        let mut stmt = self.conn.prepare(
3504            "SELECT id, run_id, step, question, context, choices, answer, answered_by, resolved
3505             FROM pending_questions WHERE id = ?1",
3506        )?;
3507        let mut rows = stmt.query([question_id])?;
3508        match rows.next()? {
3509            Some(r) => Ok(Some(question_row(r)?)),
3510            None => Ok(None),
3511        }
3512    }
3513
3514    /// The answer already recorded for this exact question on this run and step, if
3515    /// there is one.
3516    ///
3517    /// A query for a caller reconstructing a run, **not** the mechanism a resume uses.
3518    /// The step that asks a question is committed before the run pauses, so a resume
3519    /// starts at the step after it and the `ask_question` call is never replayed —
3520    /// [`resume_with_answer`](crate::resume_with_answer) delivers the answer as an
3521    /// observation instead. See [`Self::questions`] for the whole conversation.
3522    pub fn answered_question(
3523        &self,
3524        run_id: i64,
3525        step: u32,
3526        question: &str,
3527    ) -> Result<Option<PendingQuestion>> {
3528        let mut stmt = self.conn.prepare(
3529            "SELECT id, run_id, step, question, context, choices, answer, answered_by, resolved
3530             FROM pending_questions
3531             WHERE run_id = ?1 AND step = ?2 AND question = ?3 AND resolved = 1
3532             ORDER BY id DESC LIMIT 1",
3533        )?;
3534        let mut rows = stmt.query(rusqlite::params![run_id, step, question])?;
3535        match rows.next()? {
3536            Some(r) => Ok(Some(question_row(r)?)),
3537            None => Ok(None),
3538        }
3539    }
3540
3541    /// Record an answer and mark the question resolved.
3542    ///
3543    /// `by` is `"responder"` or `"human"`. Answering an already-answered question is
3544    /// an [`Error::Resume`] rather than a silent second write, the way
3545    /// [`Self::resolve_pending`] refuses a second decision: two answers to one
3546    /// question means one of them was never acted on, and a caller should hear which.
3547    pub fn answer_question(&self, question_id: i64, answer: &str, by: &str) -> Result<()> {
3548        let existing = self.question(question_id)?;
3549        match existing {
3550            None => {
3551                return Err(Error::Resume {
3552                    reason: format!("no question {question_id} to answer"),
3553                })
3554            }
3555            Some(q) if q.resolved => {
3556                return Err(Error::Resume {
3557                    reason: format!("question {question_id} was already answered"),
3558                })
3559            }
3560            Some(_) => {}
3561        }
3562        self.conn.execute(
3563            "UPDATE pending_questions SET answer = ?2, answered_by = ?3, resolved = 1
3564             WHERE id = ?1",
3565            rusqlite::params![question_id, answer, by],
3566        )?;
3567        Ok(())
3568    }
3569
3570    /// Every question asked on a run, in the order they were asked.
3571    pub fn questions(&self, run_id: i64) -> Result<Vec<PendingQuestion>> {
3572        let mut stmt = self.conn.prepare(
3573            "SELECT id, run_id, step, question, context, choices, answer, answered_by, resolved
3574             FROM pending_questions WHERE run_id = ?1 ORDER BY id",
3575        )?;
3576        let rows = stmt.query_map([run_id], question_row)?;
3577        Ok(rows.collect::<std::result::Result<_, _>>()?)
3578    }
3579
3580    // ---- 0.21.0: the agent's plan ----
3581
3582    /// Replace this run's plan with `items`.
3583    ///
3584    /// Wholesale, in one transaction: the old rows go and the new ones land, so a
3585    /// reader on another connection sees the previous plan or the next one and never
3586    /// a half-written mixture of the two. That atomicity is the whole reason an
3587    /// operator can read a plan mid-run and trust what they see.
3588    ///
3589    /// Bounded like every other tool result in the crate rather than refused: at most
3590    /// [`TODO_MAX_ITEMS`] items, each at most [`TODO_TEXT_CAP`] characters. Returns
3591    /// how many items were dropped to hold the cap, so the caller can say so in the
3592    /// observation instead of letting a plan quietly lose its tail.
3593    ///
3594    /// Writes no trace row of its own — the run loop records the write where the
3595    /// step number is known, exactly as it does for [`Self::memory_put`].
3596    pub fn write_todos(&self, run_id: i64, items: &[TodoItem]) -> Result<usize> {
3597        let kept = items.len().min(TODO_MAX_ITEMS);
3598        let dropped = items.len() - kept;
3599        // One transaction: a reader on another connection sees the old plan or the
3600        // new one, never both halves.
3601        let tx = self.conn.unchecked_transaction()?;
3602        tx.execute("DELETE FROM todos WHERE run_id = ?1", [run_id])?;
3603        {
3604            let mut stmt = tx.prepare(
3605                "INSERT INTO todos (run_id, position, text, state) VALUES (?1, ?2, ?3, ?4)",
3606            )?;
3607            for (i, item) in items.iter().take(kept).enumerate() {
3608                let text: String = item.text.chars().take(TODO_TEXT_CAP).collect();
3609                stmt.execute(rusqlite::params![
3610                    run_id,
3611                    i as i64,
3612                    text,
3613                    item.state.as_str()
3614                ])?;
3615            }
3616        }
3617        tx.commit()?;
3618        Ok(dropped)
3619    }
3620
3621    /// This run's plan, in the order the agent wrote it.
3622    ///
3623    /// Empty for a run that never wrote one, and empty — not absent — for a run that
3624    /// cleared its plan, because an agent that finished its work and emptied its list
3625    /// is not an agent that never had one.
3626    ///
3627    /// A row whose `state` is not one [`TodoState`] understands is skipped rather than
3628    /// guessed at; the writer above only ever writes the three, so this can only
3629    /// happen to a database another program has written to.
3630    pub fn todos(&self, run_id: i64) -> Result<Vec<TodoItem>> {
3631        let mut stmt = self
3632            .conn
3633            .prepare("SELECT text, state FROM todos WHERE run_id = ?1 ORDER BY position")?;
3634        let rows = stmt.query_map([run_id], |r| {
3635            let text: String = r.get(0)?;
3636            let state: String = r.get(1)?;
3637            Ok((text, state))
3638        })?;
3639        let mut out = Vec::new();
3640        for row in rows {
3641            let (text, state) = row?;
3642            if let Some(state) = TodoState::parse(&state) {
3643                out.push(TodoItem { text, state });
3644            }
3645        }
3646        Ok(out)
3647    }
3648
3649    // ---- 0.22.0: what the provider looked up ----
3650
3651    /// Record the sources a completion cited, at the step that made it.
3652    ///
3653    /// Verbatim, and without judgement: this crate never fetches the url, so a row
3654    /// says the provider cited a page, not that the page says what the model
3655    /// claimed. A url already recorded for the same run and step is not written
3656    /// twice — a vendor repeats it on every sentence it supports.
3657    ///
3658    /// ```
3659    /// use io_harness::{Citation, Store};
3660    ///
3661    /// # fn main() -> io_harness::Result<()> {
3662    /// let store = Store::memory()?;
3663    /// let run = store.start_run("what shipped this week", "anthropic")?;
3664    /// store.record_citations(run, 1, &[Citation {
3665    ///     url: "https://docs.rs/io-harness".into(),
3666    ///     title: Some("io-harness".into()),
3667    ///     cited_text: None,
3668    /// }])?;
3669    ///
3670    /// // Readable afterwards from the store alone, which is what makes "where did
3671    /// // that claim come from" answerable once the process that ran it is gone.
3672    /// let cited = store.citations(run)?;
3673    /// assert_eq!(cited.len(), 1);
3674    /// assert_eq!(cited[0].url, "https://docs.rs/io-harness");
3675    /// # Ok(())
3676    /// # }
3677    /// ```
3678    pub fn record_citations(&self, run_id: i64, step: u32, citations: &[Citation]) -> Result<()> {
3679        if citations.is_empty() {
3680            return Ok(());
3681        }
3682        let tx = self.conn.unchecked_transaction()?;
3683        {
3684            let mut stmt = tx.prepare(
3685                "INSERT INTO citations (run_id, step, url, title, cited_text)
3686                 SELECT ?1, ?2, ?3, ?4, ?5
3687                 WHERE NOT EXISTS (
3688                     SELECT 1 FROM citations WHERE run_id = ?1 AND step = ?2 AND url = ?3
3689                 )",
3690            )?;
3691            for citation in citations {
3692                stmt.execute(rusqlite::params![
3693                    run_id,
3694                    step,
3695                    &citation.url,
3696                    &citation.title,
3697                    &citation.cited_text,
3698                ])?;
3699            }
3700        }
3701        tx.commit()?;
3702        Ok(())
3703    }
3704
3705    /// Every source this run cited, in the order the steps ran.
3706    ///
3707    /// Empty for a run that never searched — which is every run before a
3708    /// [`WebAccess`](crate::WebAccess) declaration, and every run whose model
3709    /// answered without looking anything up.
3710    ///
3711    /// ```
3712    /// use io_harness::Store;
3713    ///
3714    /// # fn main() -> io_harness::Result<()> {
3715    /// let store = Store::memory()?;
3716    /// let run = store.start_run("a task with no searching in it", "anthropic")?;
3717    /// // Nothing cited is an empty list, not an error: a run that answered from
3718    /// // what it already knew is a normal run.
3719    /// assert!(store.citations(run)?.is_empty());
3720    /// # Ok(())
3721    /// # }
3722    /// ```
3723    pub fn citations(&self, run_id: i64) -> Result<Vec<Citation>> {
3724        let mut stmt = self.conn.prepare(
3725            "SELECT url, title, cited_text FROM citations WHERE run_id = ?1 ORDER BY id",
3726        )?;
3727        let rows = stmt.query_map([run_id], |r| {
3728            Ok(Citation {
3729                url: r.get(0)?,
3730                title: r.get(1)?,
3731                cited_text: r.get(2)?,
3732            })
3733        })?;
3734        let mut out = Vec::new();
3735        for row in rows {
3736            out.push(row?);
3737        }
3738        Ok(out)
3739    }
3740
3741    /// Record the provider-executed calls a completion reported.
3742    ///
3743    /// Both kinds: the ones that worked and the ones that failed inside an
3744    /// otherwise successful response. Keeping the failures is the point — a vendor
3745    /// reports a broken search as an error object rather than an HTTP status, so a
3746    /// trace without these rows cannot tell a search that broke from one that
3747    /// found nothing.
3748    ///
3749    /// ```
3750    /// use io_harness::{ServerToolCall, Store};
3751    ///
3752    /// # fn main() -> io_harness::Result<()> {
3753    /// let store = Store::memory()?;
3754    /// let run = store.start_run("what shipped this week", "anthropic")?;
3755    /// store.record_server_tool_calls(run, 1, &[
3756    ///     ServerToolCall::ok("anthropic", "web_search"),
3757    ///     ServerToolCall::failed("anthropic", "web_search", "max_uses_exceeded"),
3758    /// ])?;
3759    ///
3760    /// let calls = store.server_tool_calls(run)?;
3761    /// assert_eq!(calls.len(), 2);
3762    /// assert!(calls[0].succeeded());
3763    /// assert_eq!(calls[1].error.as_deref(), Some("max_uses_exceeded"));
3764    /// # Ok(())
3765    /// # }
3766    /// ```
3767    pub fn record_server_tool_calls(
3768        &self,
3769        run_id: i64,
3770        step: u32,
3771        calls: &[ServerToolCall],
3772    ) -> Result<()> {
3773        if calls.is_empty() {
3774            return Ok(());
3775        }
3776        let tx = self.conn.unchecked_transaction()?;
3777        {
3778            let mut stmt = tx.prepare(
3779                "INSERT INTO server_tool_calls (run_id, step, provider, tool, error)
3780                 VALUES (?1, ?2, ?3, ?4, ?5)",
3781            )?;
3782            for call in calls {
3783                stmt.execute(rusqlite::params![
3784                    run_id,
3785                    step,
3786                    &call.provider,
3787                    &call.tool,
3788                    &call.error,
3789                ])?;
3790            }
3791        }
3792        tx.commit()?;
3793        Ok(())
3794    }
3795
3796    /// Every provider-executed call this run made, in the order they were made.
3797    ///
3798    /// ```
3799    /// use io_harness::Store;
3800    ///
3801    /// # fn main() -> io_harness::Result<()> {
3802    /// let store = Store::memory()?;
3803    /// let run = store.start_run("a task with no searching in it", "openai")?;
3804    /// assert!(store.server_tool_calls(run)?.is_empty());
3805    /// # Ok(())
3806    /// # }
3807    /// ```
3808    pub fn server_tool_calls(&self, run_id: i64) -> Result<Vec<ServerToolCall>> {
3809        let mut stmt = self.conn.prepare(
3810            "SELECT provider, tool, error FROM server_tool_calls WHERE run_id = ?1 ORDER BY id",
3811        )?;
3812        let rows = stmt.query_map([run_id], |r| {
3813            Ok(ServerToolCall {
3814                provider: r.get(0)?,
3815                tool: r.get(1)?,
3816                error: r.get(2)?,
3817            })
3818        })?;
3819        let mut out = Vec::new();
3820        for row in rows {
3821            out.push(row?);
3822        }
3823        Ok(out)
3824    }
3825
3826    // ---- 0.10.0: durable cross-run memory ----
3827
3828    /// Write or replace `key` for `workspace`, attributed to the run and step
3829    /// that wrote it. A value past [`MEMORY_MAX_ENTRY_CHARS`] is truncated with
3830    /// a visible marker rather than refused. Returns the keys evicted to stay
3831    /// inside the caps, oldest first — the caller records the eviction in the
3832    /// trace; this never writes a trace row itself.
3833    pub fn memory_put(
3834        &self,
3835        workspace: &str,
3836        key: &str,
3837        value: &str,
3838        run_id: i64,
3839        step: u32,
3840    ) -> Result<Vec<String>> {
3841        let value = truncate_memory_value(value);
3842        // An overwrite re-attributes the entry and refreshes `created_at`, so
3843        // recency ordering reflects the latest write rather than the first.
3844        self.conn.execute(
3845            "INSERT INTO memory (workspace, key, value, run_id, step, created_at)
3846             VALUES (?1, ?2, ?3, ?4, ?5, strftime('%Y-%m-%dT%H:%M:%fZ','now'))
3847             ON CONFLICT(workspace, key) DO UPDATE SET
3848                 value      = excluded.value,
3849                 run_id     = excluded.run_id,
3850                 step       = excluded.step,
3851                 created_at = excluded.created_at",
3852            (workspace, key, &value, run_id, step),
3853        )?;
3854        self.enforce_memory_caps(workspace, key)
3855    }
3856
3857    /// Evict this workspace's oldest entries until both caps hold, never the
3858    /// entry `keep` (the one just written — evicting it would make a write a
3859    /// silent no-op). Returns the evicted keys in eviction order.
3860    fn enforce_memory_caps(&self, workspace: &str, keep: &str) -> Result<Vec<String>> {
3861        // LENGTH() on TEXT counts characters, not bytes — the cap is in chars.
3862        let rows: Vec<(String, i64)> = {
3863            let mut stmt = self.conn.prepare(
3864                "SELECT key, LENGTH(value) FROM memory WHERE workspace = ?1
3865                 ORDER BY created_at ASC, id ASC",
3866            )?;
3867            let rows = stmt.query_map([workspace], |r| Ok((r.get(0)?, r.get(1)?)))?;
3868            rows.collect::<std::result::Result<_, _>>()?
3869        };
3870
3871        let mut count = rows.len();
3872        let mut chars: i64 = rows.iter().map(|(_, n)| *n).sum();
3873        let mut evicted = Vec::new();
3874        for (key, n) in &rows {
3875            if count <= MEMORY_MAX_ENTRIES && chars <= MEMORY_MAX_CHARS as i64 {
3876                break;
3877            }
3878            if key == keep {
3879                continue;
3880            }
3881            self.conn.execute(
3882                "DELETE FROM memory WHERE workspace = ?1 AND key = ?2",
3883                (workspace, key),
3884            )?;
3885            count -= 1;
3886            chars -= n;
3887            evicted.push(key.clone());
3888        }
3889        Ok(evicted)
3890    }
3891
3892    /// Every entry for `workspace`, oldest first. Never another workspace's.
3893    pub fn memory_list(&self, workspace: &str) -> Result<Vec<MemoryEntry>> {
3894        let mut stmt = self.conn.prepare(
3895            "SELECT key, value, run_id, step, created_at FROM memory
3896             WHERE workspace = ?1 ORDER BY created_at ASC, id ASC",
3897        )?;
3898        let rows = stmt.query_map([workspace], |r| {
3899            Ok(MemoryEntry {
3900                key: r.get(0)?,
3901                value: r.get(1)?,
3902                run_id: r.get(2)?,
3903                step: r.get::<_, i64>(3)? as u32,
3904                created_at: r.get(4)?,
3905            })
3906        })?;
3907        Ok(rows.collect::<std::result::Result<_, _>>()?)
3908    }
3909
3910    // -----------------------------------------------------------------------
3911    // 0.20.0 — the session tree. The conversation's shape lives here; what a
3912    // turn did lives in the run tables under its `run_id`.
3913    // -----------------------------------------------------------------------
3914
3915    /// Open a new session over `root`. Returns its id, which is all a later
3916    /// process needs to pick the conversation back up.
3917    pub fn create_session(&self, root: &str) -> Result<i64> {
3918        self.conn
3919            .execute("INSERT INTO sessions (root) VALUES (?1)", [root])?;
3920        Ok(self.conn.last_insert_rowid())
3921    }
3922
3923    /// The root a session was opened over, or `None` if no such session exists.
3924    ///
3925    /// A reopen reads the root from here rather than taking it from the caller
3926    /// again: a session whose workspace moved between processes would otherwise
3927    /// carry a conversation about one directory into another.
3928    pub fn session_root(&self, session_id: i64) -> Result<Option<String>> {
3929        Ok(self
3930            .conn
3931            .query_row(
3932                "SELECT root FROM sessions WHERE id = ?1",
3933                [session_id],
3934                |r| r.get(0),
3935            )
3936            .ok())
3937    }
3938
3939    /// Which turn a session is currently answering from.
3940    pub fn session_head(&self, session_id: i64) -> Result<Option<i64>> {
3941        Ok(self
3942            .conn
3943            .query_row(
3944                "SELECT head_turn_id FROM sessions WHERE id = ?1",
3945                [session_id],
3946                |r| r.get::<_, Option<i64>>(0),
3947            )
3948            .ok()
3949            .flatten())
3950    }
3951
3952    /// Move a session's head. Called when a turn is taken and when a caller
3953    /// branches from an earlier one.
3954    pub fn set_session_head(&self, session_id: i64, turn_id: Option<i64>) -> Result<()> {
3955        self.conn.execute(
3956            "UPDATE sessions SET head_turn_id = ?1 WHERE id = ?2",
3957            (turn_id, session_id),
3958        )?;
3959        Ok(())
3960    }
3961
3962    /// Record a turn against a session, under the run that will serve it.
3963    ///
3964    /// Written before the run loop starts, so a turn whose process dies mid-answer
3965    /// is still in the tree with a `run_id` a resume can continue from — the same
3966    /// reason a run row exists before the first completion is billed.
3967    pub fn record_turn(
3968        &self,
3969        session_id: i64,
3970        parent_turn_id: Option<i64>,
3971        run_id: i64,
3972        prompt: &str,
3973    ) -> Result<i64> {
3974        self.conn.execute(
3975            "INSERT INTO session_turns (session_id, parent_turn_id, run_id, prompt)
3976             VALUES (?1, ?2, ?3, ?4)",
3977            (session_id, parent_turn_id, run_id, prompt),
3978        )?;
3979        Ok(self.conn.last_insert_rowid())
3980    }
3981
3982    /// Close a turn with what the agent said and why it stopped. Append-only in
3983    /// spirit: the prompt and the parentage a turn was created with never change.
3984    pub fn finish_turn(&self, turn_id: i64, reply: Option<&str>, outcome: &str) -> Result<()> {
3985        self.conn.execute(
3986            "UPDATE session_turns SET reply = ?1, outcome = ?2 WHERE id = ?3",
3987            (reply, outcome, turn_id),
3988        )?;
3989        Ok(())
3990    }
3991
3992    /// One turn by id, if it exists.
3993    pub fn session_turn(&self, turn_id: i64) -> Result<Option<Turn>> {
3994        Ok(self
3995            .conn
3996            .query_row(
3997                "SELECT id, session_id, parent_turn_id, run_id, prompt, reply, outcome, created_at
3998                 FROM session_turns WHERE id = ?1",
3999                [turn_id],
4000                turn_row,
4001            )
4002            .ok())
4003    }
4004
4005    /// Which turn a run served, if it served one.
4006    ///
4007    /// The seam between the two halves of a turn: the run loop writes the row, and
4008    /// the session reads its id back rather than being handed it — the run id is
4009    /// the only thing both halves are guaranteed to know.
4010    pub fn turn_for_run(&self, run_id: i64) -> Result<Option<i64>> {
4011        Ok(self
4012            .conn
4013            .query_row(
4014                "SELECT id FROM session_turns WHERE run_id = ?1",
4015                [run_id],
4016                |r| r.get(0),
4017            )
4018            .ok())
4019    }
4020
4021    /// Every turn of a session, oldest first — the whole tree, not one path
4022    /// through it. [`crate::Session::history`] is the path.
4023    pub fn session_turns(&self, session_id: i64) -> Result<Vec<Turn>> {
4024        let mut stmt = self.conn.prepare(
4025            "SELECT id, session_id, parent_turn_id, run_id, prompt, reply, outcome, created_at
4026             FROM session_turns WHERE session_id = ?1 ORDER BY id ASC",
4027        )?;
4028        let rows = stmt.query_map([session_id], turn_row)?;
4029        Ok(rows.collect::<std::result::Result<_, _>>()?)
4030    }
4031
4032    /// One entry of `workspace` by key, if it holds one.
4033    pub fn memory_get(&self, workspace: &str, key: &str) -> Result<Option<MemoryEntry>> {
4034        Ok(self
4035            .conn
4036            .query_row(
4037                "SELECT key, value, run_id, step, created_at FROM memory
4038                 WHERE workspace = ?1 AND key = ?2",
4039                (workspace, key),
4040                |r| {
4041                    Ok(MemoryEntry {
4042                        key: r.get(0)?,
4043                        value: r.get(1)?,
4044                        run_id: r.get(2)?,
4045                        step: r.get::<_, i64>(3)? as u32,
4046                        created_at: r.get(4)?,
4047                    })
4048                },
4049            )
4050            .ok())
4051    }
4052
4053    /// Forget one entry of `workspace`. True when an entry was removed.
4054    pub fn memory_delete(&self, workspace: &str, key: &str) -> Result<bool> {
4055        let n = self.conn.execute(
4056            "DELETE FROM memory WHERE workspace = ?1 AND key = ?2",
4057            (workspace, key),
4058        )?;
4059        Ok(n > 0)
4060    }
4061
4062    /// Removes every entry for `workspace`; returns how many. Other workspaces
4063    /// keep theirs.
4064    pub fn memory_clear(&self, workspace: &str) -> Result<usize> {
4065        Ok(self
4066            .conn
4067            .execute("DELETE FROM memory WHERE workspace = ?1", [workspace])?)
4068    }
4069}
4070
4071#[cfg(test)]
4072mod tests {
4073    use super::*;
4074
4075    #[test]
4076    fn refusals_record_action_target_rule_and_layer() {
4077        let store = Store::memory().unwrap();
4078        let run = store.start_run("goal", "root").unwrap();
4079        store
4080            .record_event(
4081                run,
4082                &PolicyEvent::refusal(2, "write", "secrets/key.txt").with_rule("secrets/*", "base"),
4083            )
4084            .unwrap();
4085
4086        let events = store.events(run).unwrap();
4087        assert_eq!(events.len(), 1);
4088        let e = &events[0];
4089        assert_eq!(e.kind, "refusal");
4090        assert_eq!(e.act, "write");
4091        assert_eq!(e.target, "secrets/key.txt");
4092        assert_eq!(e.rule.as_deref(), Some("secrets/*"));
4093        // Attributable to the layer that refused, so a base-layer deny is findable.
4094        assert_eq!(e.layer.as_deref(), Some("base"));
4095    }
4096
4097    #[test]
4098    fn decisions_record_their_value_source_and_any_altered_target() {
4099        let store = Store::memory().unwrap();
4100        let run = store.start_run("goal", "root").unwrap();
4101        store
4102            .record_event(
4103                run,
4104                &PolicyEvent::decision(1, "write", "src/a.rs", "approve", "stdin")
4105                    .with_performed("src/sandbox/a.rs"),
4106            )
4107            .unwrap();
4108        store
4109            .record_event(
4110                run,
4111                &PolicyEvent::decision(2, "write", "src/b.rs", "approve", "remembered"),
4112            )
4113            .unwrap();
4114
4115        let events = store.events(run).unwrap();
4116        assert_eq!(events.len(), 2);
4117        // Requested and performed forms are distinguishable.
4118        assert_eq!(events[0].decision.as_deref(), Some("approve"));
4119        assert_eq!(events[0].target, "src/a.rs");
4120        assert_eq!(events[0].performed.as_deref(), Some("src/sandbox/a.rs"));
4121        // An auto-approval by a remembered rule is not confusable with a fresh one.
4122        assert_eq!(events[1].source.as_deref(), Some("remembered"));
4123        assert_eq!(events[1].performed, None);
4124    }
4125
4126    #[test]
4127    fn a_pre_0_4_database_migrates_in_place_and_keeps_its_rows() {
4128        let dir = tempfile::tempdir().unwrap();
4129        let path = dir.path().join("runs.db");
4130
4131        // A 0.3.0-shaped database: runs + steps only, no policy tables.
4132        {
4133            let conn = rusqlite::Connection::open(&path).unwrap();
4134            conn.execute_batch(
4135                "CREATE TABLE runs (id INTEGER PRIMARY KEY AUTOINCREMENT, goal TEXT NOT NULL,
4136                     file TEXT NOT NULL, outcome TEXT, provider TEXT);
4137                 CREATE TABLE steps (id INTEGER PRIMARY KEY AUTOINCREMENT, run_id INTEGER NOT NULL,
4138                     step INTEGER NOT NULL, decision TEXT NOT NULL, result TEXT NOT NULL,
4139                     prompt TEXT NOT NULL DEFAULT '', tool_call TEXT NOT NULL DEFAULT '',
4140                     tokens INTEGER NOT NULL DEFAULT 0);
4141                 INSERT INTO runs (goal, file) VALUES ('old goal', 'old.txt');",
4142            )
4143            .unwrap();
4144        }
4145
4146        let store = Store::open(&path).unwrap();
4147        // The pre-existing row survives; the new tables are usable.
4148        assert_eq!(store.last_step(1).unwrap(), 0);
4149        store
4150            .record_event(1, &PolicyEvent::refusal(1, "read", ".env"))
4151            .unwrap();
4152        assert_eq!(store.events(1).unwrap().len(), 1);
4153    }
4154
4155    /// NF1 — a 0.19.0 database gains the two session tables on open and keeps
4156    /// everything it had. The integration test cannot write a pre-session schema
4157    /// (`Store::open` always creates them), so the legacy shape is built here.
4158    #[test]
4159    fn a_pre_session_database_gains_the_session_tables_and_keeps_its_rows() {
4160        let dir = tempfile::tempdir().unwrap();
4161        let path = dir.path().join("runs.db");
4162
4163        // A 0.19.0-shaped database: everything except `sessions` and
4164        // `session_turns`, which is what the version before this one wrote.
4165        {
4166            let store = Store::open(&path).unwrap();
4167            let run = store.start_run("an older run", "notes.md").unwrap();
4168            store.finish_run(run, "success").unwrap();
4169            store
4170                .conn
4171                .execute_batch("DROP TABLE sessions; DROP TABLE session_turns;")
4172                .unwrap();
4173            // No format bump means the old file is still resumable by this binary.
4174            let format: i64 = store
4175                .conn
4176                .query_row("PRAGMA user_version", [], |r| r.get(0))
4177                .unwrap();
4178            assert_eq!(format, CHECKPOINT_FORMAT);
4179        }
4180
4181        let store = Store::open(&path).unwrap();
4182        // The run it already had is untouched...
4183        assert_eq!(
4184            store.run_summary(1).unwrap().map(|s| s.outcome),
4185            Some("success".to_string())
4186        );
4187        // ...and a conversation works over the same file.
4188        let session = store.create_session("/repo").unwrap();
4189        let run = store.start_run("a turn", "/repo").unwrap();
4190        let turn = store.record_turn(session, None, run, "hello").unwrap();
4191        assert_eq!(store.turn_for_run(run).unwrap(), Some(turn));
4192        assert_eq!(store.session_turns(session).unwrap().len(), 1);
4193        assert_eq!(
4194            store.session_root(session).unwrap().as_deref(),
4195            Some("/repo")
4196        );
4197    }
4198
4199    /// A branch is two turns with one parent, and reading one path never sees the
4200    /// other's turns. The tree half of F3 at the store level, where the walk
4201    /// [`crate::Session::history`] performs is one query.
4202    #[test]
4203    fn two_turns_may_share_a_parent_and_neither_is_rewritten() {
4204        let store = Store::memory().unwrap();
4205        let session = store.create_session("/repo").unwrap();
4206        let run = |n: &str| store.start_run(n, "/repo").unwrap();
4207
4208        let root = store
4209            .record_turn(session, None, run("t1"), "plan it")
4210            .unwrap();
4211        let left = store
4212            .record_turn(session, Some(root), run("t2"), "plan A")
4213            .unwrap();
4214        let right = store
4215            .record_turn(session, Some(root), run("t3"), "plan B")
4216            .unwrap();
4217
4218        store.finish_turn(left, Some("did A"), "finished").unwrap();
4219        // Closing one branch does not touch the other.
4220        assert_eq!(
4221            store.session_turn(right).unwrap().unwrap().reply,
4222            None,
4223            "closing a sibling turn changed this one"
4224        );
4225        assert_eq!(
4226            store.session_turn(left).unwrap().unwrap().reply.as_deref(),
4227            Some("did A")
4228        );
4229        let all = store.session_turns(session).unwrap();
4230        assert_eq!(all.len(), 3);
4231        assert_eq!(
4232            all.iter()
4233                .filter(|t| t.parent_turn_id == Some(root))
4234                .count(),
4235            2
4236        );
4237    }
4238
4239    #[test]
4240    fn a_pending_approval_survives_the_store_being_reopened() {
4241        let dir = tempfile::tempdir().unwrap();
4242        let path = dir.path().join("runs.db");
4243
4244        let request_id = {
4245            let store = Store::open(&path).unwrap();
4246            let run = store.start_run("goal", "root").unwrap();
4247            store
4248                .put_pending(run, 3, "write", "src/a.rs", Some("fn a() {}"))
4249                .unwrap()
4250        };
4251
4252        // A different Store over the same file — the process that created it is gone.
4253        let store = Store::open(&path).unwrap();
4254        let p = store.pending(request_id).unwrap().expect("still pending");
4255        assert_eq!(p.step, 3);
4256        assert_eq!(p.act, "write");
4257        assert_eq!(p.target, "src/a.rs");
4258        assert_eq!(p.content.as_deref(), Some("fn a() {}"));
4259        assert_eq!(p.resolved, None);
4260
4261        store.resolve_pending(request_id, "approve").unwrap();
4262        let p = store.pending(request_id).unwrap().unwrap();
4263        assert_eq!(p.resolved.as_deref(), Some("approve"));
4264    }
4265
4266    #[test]
4267    fn the_tree_is_reconstructable_from_a_reopened_store() {
4268        let dir = tempfile::tempdir().unwrap();
4269        let path = dir.path().join("runs.db");
4270
4271        // A parent spawns two children (one nests a grandchild) and the tree
4272        // draws against its ceiling, then everything is dropped.
4273        let (root, c1, c2, gc) = {
4274            let store = Store::open(&path).unwrap();
4275            let root = store.start_run("root goal", "ws").unwrap();
4276            let c1 = store.start_child_run("child 1", "ws", root, 1).unwrap();
4277            let c2 = store.start_child_run("child 2", "ws", root, 1).unwrap();
4278            let gc = store.start_child_run("grandchild", "ws", c1, 2).unwrap();
4279            store
4280                .record_agent_event(&AgentEvent::spawn(root, 1, c1, "child 1"))
4281                .unwrap();
4282            store
4283                .record_agent_event(&AgentEvent::spawn(root, 1, c2, "child 2"))
4284                .unwrap();
4285            store
4286                .record_agent_event(&AgentEvent::spawn(c1, 1, gc, "grandchild"))
4287                .unwrap();
4288            store
4289                .record_agent_event(&AgentEvent::spawn_refused(root, 2, "agents"))
4290                .unwrap();
4291            store
4292                .record_agent_event(&AgentEvent::budget_draw(c1, 1, 30, 70))
4293                .unwrap();
4294            (root, c1, c2, gc)
4295        };
4296
4297        // A fresh Store over the same file — the process that built the tree is gone.
4298        let store = Store::open(&path).unwrap();
4299        // The parent/child edges rebuild the graph.
4300        assert_eq!(store.children(root).unwrap(), vec![c1, c2]);
4301        assert_eq!(store.children(c1).unwrap(), vec![gc]);
4302        assert_eq!(store.parent(gc).unwrap(), Some(c1));
4303        assert_eq!(store.parent(root).unwrap(), None);
4304        assert_eq!(store.depth(gc).unwrap(), 2);
4305
4306        // Spawns, the refusal, and the draw are all recorded.
4307        let root_events = store.agent_events(root).unwrap();
4308        assert_eq!(root_events.iter().filter(|e| e.kind == "spawn").count(), 2);
4309        assert_eq!(
4310            root_events
4311                .iter()
4312                .filter(|e| e.kind == "spawn_refused")
4313                .count(),
4314            1
4315        );
4316        let draws = store.agent_events(c1).unwrap();
4317        let draw = draws.iter().find(|e| e.kind == "budget_draw").unwrap();
4318        assert_eq!(draw.tokens, Some(30));
4319        assert_eq!(draw.remaining, Some(70));
4320    }
4321
4322    #[test]
4323    fn a_pre_0_5_database_migrates_and_keeps_its_rows() {
4324        let dir = tempfile::tempdir().unwrap();
4325        let path = dir.path().join("runs.db");
4326
4327        // A 0.4.0-shaped database: runs (no parent_run_id/depth), steps, and the
4328        // policy tables, with a row.
4329        {
4330            let conn = rusqlite::Connection::open(&path).unwrap();
4331            conn.execute_batch(
4332                "CREATE TABLE runs (id INTEGER PRIMARY KEY AUTOINCREMENT, goal TEXT NOT NULL,
4333                     file TEXT NOT NULL, outcome TEXT, provider TEXT);
4334                 CREATE TABLE steps (id INTEGER PRIMARY KEY AUTOINCREMENT, run_id INTEGER NOT NULL,
4335                     step INTEGER NOT NULL, decision TEXT NOT NULL, result TEXT NOT NULL,
4336                     prompt TEXT NOT NULL DEFAULT '', tool_call TEXT NOT NULL DEFAULT '',
4337                     tokens INTEGER NOT NULL DEFAULT 0);
4338                 INSERT INTO runs (goal, file) VALUES ('old', 'old.txt');",
4339            )
4340            .unwrap();
4341        }
4342
4343        let store = Store::open(&path).unwrap();
4344        // The pre-existing row survives and reads as a root at depth 0.
4345        assert_eq!(store.parent(1).unwrap(), None);
4346        assert_eq!(store.depth(1).unwrap(), 0);
4347        // The new table is usable.
4348        let child = store.start_child_run("c", "ws", 1, 1).unwrap();
4349        assert_eq!(store.children(1).unwrap(), vec![child]);
4350    }
4351
4352    #[test]
4353    fn a_pre_0_8_database_migrates_in_place_and_keeps_its_rows() {
4354        let dir = tempfile::tempdir().unwrap();
4355        let path = dir.path().join("runs.db");
4356
4357        // A 0.7.0-shaped database: everything through checkpoints, and no
4358        // mcp_events table.
4359        {
4360            let store = Store::open(&path).unwrap();
4361            let run = store.start_run("old goal", "old.txt").unwrap();
4362            store
4363                .checkpoint_step(run, &StepRecord::new(1, "wrote", "ok"))
4364                .unwrap();
4365            store
4366                .record_event(run, &PolicyEvent::refusal(1, "write", "secrets/k"))
4367                .unwrap();
4368            store
4369                .conn
4370                .execute("DROP TABLE IF EXISTS mcp_events", [])
4371                .unwrap();
4372        }
4373
4374        // Reopening migrates it: the old rows are intact and the new table works.
4375        let store = Store::open(&path).unwrap();
4376        assert_eq!(store.last_step(1).unwrap(), 1);
4377        assert_eq!(store.events(1).unwrap().len(), 1);
4378        assert!(store.mcp_events(1).unwrap().is_empty());
4379        store
4380            .record_mcp(1, &McpEvent::connected("files", "stdio"))
4381            .unwrap();
4382        let events = store.mcp_events(1).unwrap();
4383        assert_eq!(events.len(), 1);
4384        assert_eq!(events[0].detail.as_deref(), Some("stdio"));
4385
4386        // And a 0.7.0 binary, which never queries mcp_events, still reads it —
4387        // nothing it knows about was altered or rewritten.
4388        assert_eq!(store.steps(1).unwrap().len(), 1);
4389        assert_eq!(store.run_status(1).unwrap(), Some(RunStatus::Running));
4390    }
4391
4392    #[test]
4393    fn full_trace_persists_and_reads_back() {
4394        let store = Store::memory().unwrap();
4395        let run = store.start_run("goal", "out.txt").unwrap();
4396        store
4397            .record(
4398                run,
4399                &StepRecord::new(1, "wrote file", "content v1").with_trace(
4400                    "the prompt",
4401                    r#"{"content":"content v1"}"#,
4402                    128,
4403                ),
4404            )
4405            .unwrap();
4406        store
4407            .record(run, &StepRecord::new(2, "verified", "ok"))
4408            .unwrap();
4409        store.finish_run(run, "success").unwrap();
4410
4411        let steps = store.steps(run).unwrap();
4412        assert_eq!(steps.len(), 2);
4413        assert_eq!(steps[0].decision, "wrote file");
4414        assert_eq!(steps[0].prompt, "the prompt");
4415        assert_eq!(steps[0].tokens, 128);
4416        assert_eq!(steps[1].result, "ok");
4417        assert_eq!(store.last_step(run).unwrap(), 2);
4418    }
4419
4420    #[test]
4421    fn migrates_a_0_1_0_steps_table_in_place() {
4422        // A 0.1.0 database: `steps` without the trace columns, with a row.
4423        let conn = Connection::open_in_memory().unwrap();
4424        conn.execute_batch(
4425            "CREATE TABLE runs (id INTEGER PRIMARY KEY AUTOINCREMENT, goal TEXT NOT NULL, file TEXT NOT NULL, outcome TEXT);
4426             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);
4427             INSERT INTO runs (goal, file) VALUES ('g', 'f');
4428             INSERT INTO steps (run_id, step, decision, result) VALUES (1, 1, 'wrote file', 'old');",
4429        )
4430        .unwrap();
4431
4432        // Opening through Store migrates it; the old row survives with defaults.
4433        let store = Store::from_conn(conn).unwrap();
4434        let steps = store.steps(1).unwrap();
4435        assert_eq!(steps.len(), 1);
4436        assert_eq!(steps[0].result, "old");
4437        assert_eq!(steps[0].prompt, "");
4438        assert_eq!(steps[0].tokens, 0);
4439    }
4440
4441    #[test]
4442    fn provider_is_recorded_and_read_back() {
4443        let store = Store::memory().unwrap();
4444        let run = store.start_run("g", "f").unwrap();
4445        assert_eq!(store.provider(run).unwrap(), None);
4446        store.set_provider(run, "anthropic").unwrap();
4447        assert_eq!(store.provider(run).unwrap().as_deref(), Some("anthropic"));
4448    }
4449
4450    #[test]
4451    fn migrates_a_pre_0_3_runs_table_adding_provider() {
4452        // A 0.1/0.2 database: `runs` without the provider column.
4453        let conn = Connection::open_in_memory().unwrap();
4454        conn.execute_batch(
4455            "CREATE TABLE runs (id INTEGER PRIMARY KEY AUTOINCREMENT, goal TEXT NOT NULL, file TEXT NOT NULL, outcome TEXT);
4456             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);
4457             INSERT INTO runs (goal, file) VALUES ('g', 'f');",
4458        )
4459        .unwrap();
4460
4461        // Opening through Store adds the provider column; the old row survives.
4462        let store = Store::from_conn(conn).unwrap();
4463        assert_eq!(store.provider(1).unwrap(), None);
4464        store.set_provider(1, "openai").unwrap();
4465        assert_eq!(store.provider(1).unwrap().as_deref(), Some("openai"));
4466    }
4467
4468    // ---- 0.7.0: durable checkpoint + resume ----
4469
4470    #[test]
4471    fn checkpoint_step_commits_the_step_and_its_event_together() {
4472        let store = Store::memory().unwrap();
4473        let run = store.start_run("goal", "root").unwrap();
4474        store
4475            .checkpoint_step(run, &StepRecord::new(1, "act", "ok"))
4476            .unwrap();
4477        store
4478            .checkpoint_step(run, &StepRecord::new(2, "act", "ok"))
4479            .unwrap();
4480
4481        assert_eq!(store.last_step(run).unwrap(), 2);
4482        assert_eq!(store.steps(run).unwrap().len(), 2);
4483        let cps: Vec<_> = store
4484            .checkpoint_events(run)
4485            .unwrap()
4486            .into_iter()
4487            .filter(|e| e.kind == "checkpoint")
4488            .collect();
4489        assert_eq!(cps.len(), 2);
4490        // NF4: a checkpoint event carries no file content — only step metadata.
4491        assert!(cps.iter().all(|e| e.detail.is_none()));
4492    }
4493
4494    #[test]
4495    fn a_rolled_back_step_leaves_the_prior_checkpoint_intact() {
4496        // The committed checkpoint is the completion marker: a step whose
4497        // transaction never commits (a crash mid-commit) vanishes entirely and
4498        // the prior checkpoint stands — never a torn half recorded as done.
4499        let store = Store::memory().unwrap();
4500        let run = store.start_run("goal", "root").unwrap();
4501        store
4502            .checkpoint_step(run, &StepRecord::new(1, "act", "ok"))
4503            .unwrap();
4504
4505        // Simulate a crash mid-commit: open the step's transaction, write both
4506        // rows, then drop without committing (as a killed process would).
4507        {
4508            let tx = store.conn.unchecked_transaction().unwrap();
4509            tx.execute(
4510                "INSERT INTO steps (run_id, step, decision, result) VALUES (?1, 2, 'act', 'ok')",
4511                [run],
4512            )
4513            .unwrap();
4514            tx.execute(
4515                "INSERT INTO checkpoint_events (run_id, step, kind) VALUES (?1, 2, 'checkpoint')",
4516                [run],
4517            )
4518            .unwrap();
4519            // no tx.commit() — dropped here, rolling back.
4520        }
4521
4522        assert_eq!(
4523            store.last_step(run).unwrap(),
4524            1,
4525            "the torn step must not survive"
4526        );
4527        assert_eq!(store.steps(run).unwrap().len(), 1);
4528    }
4529
4530    #[test]
4531    fn check_resumable_refuses_a_newer_format_and_a_missing_run() {
4532        let store = Store::memory().unwrap();
4533        let run = store.start_run("goal", "root").unwrap();
4534        assert!(store.check_resumable(run).is_ok());
4535
4536        // A run id that does not exist is a typed Resume error, not a panic.
4537        assert!(matches!(
4538            store.check_resumable(9999),
4539            Err(Error::Resume { .. })
4540        ));
4541
4542        // A store written by a newer checkpoint format is refused rather than
4543        // misread.
4544        store
4545            .conn
4546            .execute_batch(&format!("PRAGMA user_version = {}", CHECKPOINT_FORMAT + 1))
4547            .unwrap();
4548        assert!(matches!(
4549            store.check_resumable(run),
4550            Err(Error::Resume { .. })
4551        ));
4552    }
4553
4554    #[test]
4555    fn spent_tokens_and_elapsed_are_durable_reads() {
4556        let store = Store::memory().unwrap();
4557        let run = store.start_run("goal", "root").unwrap();
4558        store
4559            .checkpoint_step(run, &StepRecord::new(1, "a", "ok").with_trace("p", "t", 30))
4560            .unwrap();
4561        store
4562            .checkpoint_step(run, &StepRecord::new(2, "a", "ok").with_trace("p", "t", 12))
4563            .unwrap();
4564        assert_eq!(store.spent_tokens(run).unwrap(), 42);
4565        assert!(store.elapsed_secs(run).unwrap() >= 0.0);
4566    }
4567
4568    #[test]
4569    fn tree_aggregate_reads_span_root_and_descendants() {
4570        let store = Store::memory().unwrap();
4571        let root = store.start_run("goal", "root").unwrap();
4572        let child = store.start_child_run("sub", "root", root, 1).unwrap();
4573        let grandchild = store.start_child_run("subsub", "root", child, 2).unwrap();
4574        store
4575            .checkpoint_step(
4576                root,
4577                &StepRecord::new(1, "a", "ok").with_trace("p", "t", 10),
4578            )
4579            .unwrap();
4580        store
4581            .checkpoint_step(
4582                child,
4583                &StepRecord::new(1, "a", "ok").with_trace("p", "t", 20),
4584            )
4585            .unwrap();
4586        store
4587            .checkpoint_step(
4588                grandchild,
4589                &StepRecord::new(1, "a", "ok").with_trace("p", "t", 5),
4590            )
4591            .unwrap();
4592
4593        assert_eq!(
4594            store.tree_run_ids(root).unwrap(),
4595            vec![root, child, grandchild]
4596        );
4597        assert_eq!(store.spent_tokens_tree(root).unwrap(), 35);
4598        assert_eq!(store.agent_count_tree(root).unwrap(), 3);
4599    }
4600
4601    #[test]
4602    fn status_round_trips_and_a_pre_0_7_database_migrates() {
4603        // A 0.6.0-shaped database: runs without status/started_at.
4604        let conn = Connection::open_in_memory().unwrap();
4605        conn.execute_batch(
4606            "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);
4607             INSERT INTO runs (goal, file) VALUES ('g', 'f');",
4608        )
4609        .unwrap();
4610        let store = Store::from_conn(conn).unwrap();
4611        // The old row gains a default status and no start stamp.
4612        assert_eq!(store.status(1).unwrap().as_deref(), Some("running"));
4613        store.set_status(1, "completed").unwrap();
4614        assert_eq!(store.status(1).unwrap().as_deref(), Some("completed"));
4615    }
4616
4617    // ---- 0.10.0: durable cross-run memory ----
4618
4619    #[test]
4620    fn the_entry_count_cap_evicts_oldest_first_and_never_the_new_entry() {
4621        let store = Store::memory().unwrap();
4622        for i in 0..MEMORY_MAX_ENTRIES {
4623            let evicted = store.memory_put("ws", &format!("k{i}"), "v", 1, 1).unwrap();
4624            assert!(evicted.is_empty(), "no eviction while under the cap");
4625        }
4626        assert_eq!(store.memory_list("ws").unwrap().len(), MEMORY_MAX_ENTRIES);
4627
4628        // Three more writes cost exactly the three oldest keys, in order.
4629        let mut evicted = Vec::new();
4630        for i in 0..3 {
4631            evicted.extend(
4632                store
4633                    .memory_put("ws", &format!("new{i}"), "v", 2, 2)
4634                    .unwrap(),
4635            );
4636        }
4637        assert_eq!(evicted, vec!["k0", "k1", "k2"]);
4638
4639        let keys: Vec<String> = store
4640            .memory_list("ws")
4641            .unwrap()
4642            .into_iter()
4643            .map(|e| e.key)
4644            .collect();
4645        assert_eq!(
4646            keys.len(),
4647            MEMORY_MAX_ENTRIES,
4648            "the cap holds after eviction"
4649        );
4650        assert!(!keys.contains(&"k0".to_string()));
4651        // The entry just written is never the one evicted to make room for it.
4652        for i in 0..3 {
4653            assert!(keys.contains(&format!("new{i}")));
4654        }
4655    }
4656
4657    #[test]
4658    fn the_total_chars_cap_evicts_before_the_count_cap_is_reached() {
4659        let store = Store::memory().unwrap();
4660        let big = "x".repeat(MEMORY_MAX_ENTRY_CHARS);
4661        let mut evicted = Vec::new();
4662        // 10 entries of 2_000 chars = 20_000, past the 16_000 char cap while the
4663        // 64-entry cap is nowhere near.
4664        for i in 0..10 {
4665            evicted.extend(
4666                store
4667                    .memory_put("ws", &format!("k{i}"), &big, 1, 1)
4668                    .unwrap(),
4669            );
4670        }
4671        assert_eq!(
4672            evicted,
4673            vec!["k0", "k1"],
4674            "oldest first, count cap untouched"
4675        );
4676
4677        let entries = store.memory_list("ws").unwrap();
4678        assert!(entries.len() < MEMORY_MAX_ENTRIES);
4679        let total: usize = entries.iter().map(|e| e.value.chars().count()).sum();
4680        assert!(total <= MEMORY_MAX_CHARS, "{total} chars is over the cap");
4681    }
4682
4683    #[test]
4684    fn an_oversized_value_is_truncated_with_a_marker_not_rejected() {
4685        let store = Store::memory().unwrap();
4686        // Multibyte throughout, so a byte-wise cut would not be valid UTF-8.
4687        let huge = "é".repeat(MEMORY_MAX_ENTRY_CHARS * 2);
4688        assert!(store.memory_put("ws", "k", &huge, 1, 1).is_ok());
4689
4690        let stored = store.memory_get("ws", "k").unwrap().unwrap().value;
4691        assert_eq!(stored.chars().count(), MEMORY_MAX_ENTRY_CHARS);
4692        assert!(stored.ends_with(MEMORY_TRUNCATED), "the cut is visible");
4693        // Cut on a char boundary: every kept char is the whole 'é', never a half.
4694        let kept = MEMORY_MAX_ENTRY_CHARS - MEMORY_TRUNCATED.chars().count();
4695        assert!(stored.chars().take(kept).all(|c| c == 'é'));
4696    }
4697
4698    #[test]
4699    fn a_0_9_1_store_opens_unchanged_and_still_resumes() {
4700        let dir = tempfile::tempdir().unwrap();
4701        let path = dir.path().join("runs.db");
4702
4703        // A 0.9.1-shaped database: every table through mcp_events, rows in the
4704        // ones a resume reads, and no `memory` table.
4705        let before_format: i64 = {
4706            let store = Store::open(&path).unwrap();
4707            let run = store.start_run("old goal", "old.txt").unwrap();
4708            store
4709                .checkpoint_step(run, &StepRecord::new(1, "wrote", "ok"))
4710                .unwrap();
4711            store
4712                .record_event(run, &PolicyEvent::refusal(1, "write", "secrets/k"))
4713                .unwrap();
4714            store
4715                .put_pending(run, 1, "write", "src/a.rs", None)
4716                .unwrap();
4717            let child = store.start_child_run("sub", "ws", run, 1).unwrap();
4718            store
4719                .record_agent_event(&AgentEvent::spawn(run, 1, child, "sub"))
4720                .unwrap();
4721            store
4722                .record_sandbox_event(&SandboxEvent::create(run, 1, "proc"))
4723                .unwrap();
4724            store
4725                .record_spawn(run, 1, child, "sub", "out.txt", "ok", None, "[]")
4726                .unwrap();
4727            store
4728                .record_mcp(run, &McpEvent::connected("files", "stdio"))
4729                .unwrap();
4730            store.conn.execute("DROP TABLE memory", []).unwrap();
4731            store
4732                .conn
4733                .query_row("PRAGMA user_version", [], |r| r.get(0))
4734                .unwrap()
4735        };
4736
4737        // Reopening under 0.10.0 adds `memory` and touches nothing else.
4738        let store = Store::open(&path).unwrap();
4739        let after_format: i64 = store
4740            .conn
4741            .query_row("PRAGMA user_version", [], |r| r.get(0))
4742            .unwrap();
4743        assert_eq!(
4744            after_format, before_format,
4745            "the checkpoint format must not move — a 0.9.1 checkpoint still resumes"
4746        );
4747        assert_eq!(after_format, CHECKPOINT_FORMAT);
4748        // The 0.7.0 durability promise: the pre-existing run still resumes.
4749        assert!(store.check_resumable(1).is_ok());
4750
4751        // Every pre-existing table is intact, with its rows.
4752        assert_eq!(store.steps(1).unwrap().len(), 1);
4753        assert_eq!(store.last_step(1).unwrap(), 1);
4754        assert_eq!(store.events(1).unwrap().len(), 1);
4755        assert_eq!(store.pending(1).unwrap().unwrap().act, "write");
4756        assert_eq!(store.checkpoint_events(1).unwrap().len(), 1);
4757        assert_eq!(store.agent_events(1).unwrap().len(), 1);
4758        assert_eq!(store.sandbox_events(1).unwrap().len(), 1);
4759        assert_eq!(store.mcp_events(1).unwrap().len(), 1);
4760        assert_eq!(store.children(1).unwrap(), vec![2]);
4761        assert!(store.find_spawn(1, 1, "sub").is_ok());
4762        assert_eq!(store.run_status(1).unwrap(), Some(RunStatus::Running));
4763        // And the new table is there and usable.
4764        assert!(store.memory_list("ws").unwrap().is_empty());
4765        store.memory_put("ws", "k", "v", 1, 1).unwrap();
4766        assert_eq!(store.memory_get("ws", "k").unwrap().unwrap().value, "v");
4767    }
4768
4769    #[test]
4770    fn a_layered_policy_reads_back_exactly_as_it_was_recorded() {
4771        let store = Store::memory().unwrap();
4772        let run = store.start_run("goal", "root").unwrap();
4773        let policy = Policy::default()
4774            .layer("task")
4775            .deny_write("vendor/**")
4776            .rule(
4777                crate::policy::Act::Exec,
4778                crate::policy::Effect::Allow,
4779                "cargo",
4780            );
4781
4782        store.record_run_policy(run, &policy).unwrap();
4783
4784        // Equal, not merely similar: the layers, their order, and the defaults
4785        // are the boundary, so a lossy round trip is a wrong boundary.
4786        assert_eq!(store.run_policy(run).unwrap(), Some(policy));
4787    }
4788
4789    #[test]
4790    fn a_permissive_policy_reads_back_permissive() {
4791        let store = Store::memory().unwrap();
4792        let run = store.start_run("goal", "root").unwrap();
4793        store.record_run_policy(run, &Policy::permissive()).unwrap();
4794
4795        let back = store.run_policy(run).unwrap().expect("a row was recorded");
4796        assert!(back.is_permissive());
4797    }
4798
4799    #[test]
4800    fn a_run_with_no_recorded_policy_reads_back_none_not_permissive() {
4801        let store = Store::memory().unwrap();
4802        let unrecorded = store.start_run("goal", "root").unwrap();
4803        let permissive = store.start_run("goal", "root").unwrap();
4804        store
4805            .record_run_policy(permissive, &Policy::permissive())
4806            .unwrap();
4807
4808        // The distinction the table exists for: a 0.12.0 run wrote no row, and
4809        // "nobody recorded a policy" must never be read as "the caller chose to
4810        // enforce nothing".
4811        assert_eq!(store.run_policy(unrecorded).unwrap(), None);
4812        assert!(store.run_policy(permissive).unwrap().is_some());
4813    }
4814
4815    #[test]
4816    fn re_recording_a_policy_for_the_same_run_replaces_it() {
4817        let store = Store::memory().unwrap();
4818        let run = store.start_run("goal", "root").unwrap();
4819        store.record_run_policy(run, &Policy::permissive()).unwrap();
4820        store.record_run_policy(run, &Policy::default()).unwrap();
4821
4822        assert_eq!(store.run_policy(run).unwrap(), Some(Policy::default()));
4823        let rows: i64 = store
4824            .conn
4825            .query_row(
4826                "SELECT COUNT(*) FROM run_policies WHERE run_id = ?1",
4827                [run],
4828                |r| r.get(0),
4829            )
4830            .unwrap();
4831        assert_eq!(rows, 1);
4832    }
4833}