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