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