1use std::collections::HashMap;
2use std::fmt;
3use std::fs;
4use std::path::{Path, PathBuf};
5use std::process::{Command, Stdio};
6use std::sync::atomic::{AtomicU64, Ordering};
7use std::time::{SystemTime, UNIX_EPOCH};
8
9static ID_COUNTER: AtomicU64 = AtomicU64::new(1);
10
11pub type State = HashMap<String, Value>;
12
13#[derive(Clone, Debug, PartialEq)]
14pub enum Value {
15 Null,
16 Bool(bool),
17 Number(f64),
18 String(String),
19 Object(State),
20 Array(Vec<Value>),
21}
22
23impl Default for Value {
24 fn default() -> Self {
25 Value::Null
26 }
27}
28
29impl From<&str> for Value {
30 fn from(value: &str) -> Self {
31 Value::String(value.to_string())
32 }
33}
34
35impl From<String> for Value {
36 fn from(value: String) -> Self {
37 Value::String(value)
38 }
39}
40
41impl From<bool> for Value {
42 fn from(value: bool) -> Self {
43 Value::Bool(value)
44 }
45}
46
47impl From<i64> for Value {
48 fn from(value: i64) -> Self {
49 Value::Number(value as f64)
50 }
51}
52
53impl From<f64> for Value {
54 fn from(value: f64) -> Self {
55 Value::Number(value)
56 }
57}
58
59#[derive(Clone, Debug)]
60pub struct Run {
61 pub run_id: String,
62 pub session_id: String,
63 pub status: String,
64 pub state: State,
65 pub state_version: u64,
66 pub created_at: f64,
67 pub updated_at: f64,
68}
69
70#[derive(Clone, Debug)]
71pub struct Step {
72 pub step_id: String,
73 pub run_id: String,
74 pub session_id: String,
75 pub status: String,
76 pub owner: Option<String>,
77 pub lease_token: Option<String>,
78 pub lease_until: Option<f64>,
79 pub attempt: u64,
80 pub state_version: u64,
81 pub checkpoint_id: Option<String>,
82 pub last_error_type: Option<String>,
83 pub last_error: Option<String>,
84 pub cancelled_at: Option<f64>,
85 pub created_at: f64,
86 pub updated_at: f64,
87}
88
89#[derive(Clone, Debug)]
90pub struct StepClaim {
91 pub run_id: String,
92 pub session_id: String,
93 pub step_id: String,
94 pub attempt: u64,
95 pub lease_token: String,
96 pub state_version: u64,
97 pub lease_until: f64,
98}
99
100#[derive(Clone, Debug)]
101pub struct Event {
102 pub event_id: String,
103 pub run_id: String,
104 pub session_id: Option<String>,
105 pub step_id: Option<String>,
106 pub seq: u64,
107 pub event_type: String,
108 pub timestamp: f64,
109 pub agent_role: Option<String>,
110 pub state_version: Option<u64>,
111 pub causal_token: Option<String>,
112 pub payload_hash: String,
113 pub payload_ref: String,
114 pub payload: State,
115}
116
117#[derive(Clone, Debug)]
118pub struct ToolLedgerEntry {
119 pub ledger_id: String,
120 pub run_id: String,
121 pub session_id: String,
122 pub step_id: String,
123 pub tool_name: String,
124 pub tool_version: String,
125 pub tool_call_id: String,
126 pub idempotency_key: String,
127 pub causal_token: String,
128 pub request_hash: String,
129 pub request_ref: String,
130 pub status: String,
131 pub external_id: Option<String>,
132 pub response_hash: Option<String>,
133 pub response_ref: Option<String>,
134 pub error_type: Option<String>,
135 pub response: Option<Value>,
136 pub created_at: f64,
137 pub updated_at: f64,
138}
139
140#[derive(Clone, Debug)]
141pub struct ApprovalRequest {
142 pub approval_id: String,
143 pub approval_key: String,
144 pub run_id: String,
145 pub session_id: String,
146 pub step_id: String,
147 pub tool_name: String,
148 pub risk_level: String,
149 pub status: String,
150 pub reason: String,
151 pub request_hash: String,
152 pub request_ref: String,
153 pub requested_by: String,
154 pub approved_by: Option<String>,
155 pub decision_reason: Option<String>,
156 pub created_at: f64,
157 pub updated_at: f64,
158}
159
160#[derive(Clone, Debug)]
161pub struct CostRecord {
162 pub cost_id: String,
163 pub run_id: String,
164 pub session_id: String,
165 pub step_id: String,
166 pub category: String,
167 pub name: String,
168 pub amount: f64,
169 pub unit: String,
170 pub metadata: State,
171 pub created_at: f64,
172}
173
174#[derive(Clone, Debug, Default, PartialEq)]
175pub struct CostSummary {
176 pub tool_calls: f64,
177 pub model_tokens: f64,
178 pub total_usd: f64,
179 pub by_category: HashMap<String, f64>,
180}
181
182pub const MEDIA_SCHEMA_VERSION: &str = "agentledger.media.v0";
183pub const STREAM_SCHEMA_VERSION: &str = "agentledger.stream.v0";
184
185#[derive(Clone, Debug)]
186pub struct LocalBlobStore {
187 root: PathBuf,
188}
189
190impl LocalBlobStore {
191 pub fn open(root: impl AsRef<Path>) -> Result<Self> {
192 fs::create_dir_all(root.as_ref()).map_err(|err| RuntimeError(err.to_string()))?;
193 Ok(Self {
194 root: root.as_ref().to_path_buf(),
195 })
196 }
197
198 pub fn put_json(&self, value: &Value) -> Result<(String, String)> {
199 let encoded = encode_value(value);
200 let digest = stable_hash(&encoded);
201 let dir = self.root.join("sha256");
202 fs::create_dir_all(&dir).map_err(|err| RuntimeError(err.to_string()))?;
203 let path = dir.join(format!("{digest}.json"));
204 if !path.exists() {
205 let tmp = path.with_extension("json.tmp");
206 fs::write(&tmp, encoded).map_err(|err| RuntimeError(err.to_string()))?;
207 fs::rename(&tmp, &path).map_err(|err| RuntimeError(err.to_string()))?;
208 }
209 Ok((
210 format!("sha256:{digest}"),
211 format!("blob://sha256/{digest}.json"),
212 ))
213 }
214
215 pub fn get_json(&self, reference: &str) -> Result<Value> {
216 let prefix = "blob://sha256/";
217 if !reference.starts_with(prefix) {
218 return Err(RuntimeError(format!("unsupported blob ref: {reference}")));
219 }
220 let name = &reference[prefix.len()..];
221 if !name.ends_with(".json")
222 || name.contains("..")
223 || name.contains('/')
224 || name.contains('\\')
225 {
226 return Err(RuntimeError(format!("unsupported blob ref: {reference}")));
227 }
228 let body = fs::read_to_string(self.root.join("sha256").join(name))
229 .map_err(|err| RuntimeError(err.to_string()))?;
230 decode_value(&body)
231 }
232}
233
234#[derive(Clone, Debug)]
235pub struct Artifact {
236 pub artifact_id: String,
237 pub run_id: String,
238 pub step_id: Option<String>,
239 pub name: String,
240 pub blob_hash: String,
241 pub blob_ref: String,
242 pub metadata: State,
243 pub created_at: f64,
244}
245
246#[derive(Clone, Debug, Default)]
247pub struct MediaArtifactOptions {
248 pub uri: Option<String>,
249 pub content_ref: Option<String>,
250 pub media_metadata: State,
251 pub lineage: State,
252 pub derived_outputs: State,
253 pub metadata: State,
254}
255
256#[derive(Clone, Debug, Default)]
257pub struct StreamChunkRef {
258 pub stream_id: String,
259 pub chunk_id: String,
260 pub offset: Value,
261 pub content_ref: Option<String>,
262 pub content_hash: Option<String>,
263 pub sequence: Option<f64>,
264 pub event_time: Option<f64>,
265 pub metadata: State,
266}
267
268#[derive(Clone, Debug, Default)]
269pub struct StreamCheckpointOptions {
270 pub stream_id: String,
271 pub consumer_id: String,
272 pub offset: Value,
273 pub watermark: Option<Value>,
274 pub chunk: Option<StreamChunkRef>,
275 pub partial_result_ref: Option<String>,
276 pub backpressure: State,
277 pub metadata: State,
278}
279
280#[derive(Clone, Debug, Default)]
281pub struct BudgetLimits {
282 pub max_tool_calls: Option<f64>,
283 pub max_model_tokens: Option<f64>,
284 pub max_total_usd: Option<f64>,
285}
286
287#[derive(Debug, Clone)]
288pub struct RuntimeError(pub String);
289
290impl fmt::Display for RuntimeError {
291 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
292 write!(f, "{}", self.0)
293 }
294}
295
296impl std::error::Error for RuntimeError {}
297
298pub type Result<T> = std::result::Result<T, RuntimeError>;
299pub type ToolFunc = Box<dyn Fn(State) -> Result<Value> + Send + Sync>;
300pub type AgentFunc = fn(&mut AgentContext, State) -> Result<()>;
301
302pub struct SandboxPolicy {
303 pub tool_name: String,
304 pub run_id: String,
305 pub step_id: String,
306 pub executor: String,
307 pub network: String,
308 pub filesystem: String,
309 pub timeout_seconds: u64,
310 pub extra: State,
311}
312
313pub struct SandboxResult {
314 pub ok: bool,
315 pub output: Value,
316 pub error: Option<String>,
317 pub metadata: State,
318}
319
320pub trait SandboxExecutor {
321 fn run_tool(&self, args: State, policy: &SandboxPolicy) -> SandboxResult;
322}
323
324pub struct DisabledSandboxExecutor;
325
326impl SandboxExecutor for DisabledSandboxExecutor {
327 fn run_tool(&self, _args: State, policy: &SandboxPolicy) -> SandboxResult {
328 let mut metadata = State::new();
329 metadata.insert("executor".to_string(), Value::String(policy.executor.clone()));
330 metadata.insert("isolation_level".to_string(), Value::String("none".to_string()));
331 metadata.insert("fail_closed".to_string(), Value::Bool(true));
332 SandboxResult {
333 ok: false,
334 output: Value::Null,
335 error: Some(format!("sandbox executor \"{}\" is disabled", policy.executor)),
336 metadata,
337 }
338 }
339}
340
341pub struct ToolSpec {
342 pub name: String,
343 pub version: String,
344 pub side_effect: String,
345 pub risk_level: String,
346 pub idempotency_required: bool,
347 pub approval_required: bool,
348 pub sandbox_required: bool,
349 pub sandbox_executor: String,
350 pub sandbox_policy: State,
351 pub input_schema: Option<Value>,
352 pub output_schema: Option<Value>,
353 pub func: ToolFunc,
354}
355
356impl ToolSpec {
357 pub fn new(name: &str, func: ToolFunc) -> Self {
358 Self {
359 name: name.to_string(),
360 version: "v1".to_string(),
361 side_effect: "none".to_string(),
362 risk_level: "low".to_string(),
363 idempotency_required: false,
364 approval_required: false,
365 sandbox_required: false,
366 sandbox_executor: String::new(),
367 sandbox_policy: State::new(),
368 input_schema: None,
369 output_schema: None,
370 func,
371 }
372 }
373
374 pub fn side_effect(mut self, side_effect: &str) -> Self {
375 self.side_effect = side_effect.to_string();
376 self
377 }
378
379 pub fn risk_level(mut self, risk_level: &str) -> Self {
380 self.risk_level = risk_level.to_string();
381 self
382 }
383
384 pub fn idempotency_required(mut self, required: bool) -> Self {
385 self.idempotency_required = required;
386 self
387 }
388
389 pub fn approval_required(mut self, required: bool) -> Self {
390 self.approval_required = required;
391 self
392 }
393
394 pub fn sandbox_required(mut self, required: bool) -> Self {
395 self.sandbox_required = required;
396 self
397 }
398
399 pub fn sandbox_executor(mut self, executor: &str) -> Self {
400 self.sandbox_executor = executor.to_string();
401 self
402 }
403
404 pub fn sandbox_policy(mut self, policy: State) -> Self {
405 self.sandbox_policy = policy;
406 self
407 }
408
409 pub fn input_schema(mut self, schema: Value) -> Self {
410 self.input_schema = Some(schema);
411 self
412 }
413
414 pub fn output_schema(mut self, schema: Value) -> Self {
415 self.output_schema = Some(schema);
416 self
417 }
418}
419
420#[derive(Default)]
421pub struct MemoryStore {
422 runs: HashMap<String, Run>,
423 steps: HashMap<String, Step>,
424 events: HashMap<String, Vec<Event>>,
425 tool_ledger: HashMap<String, ToolLedgerEntry>,
426 approval_requests: HashMap<String, ApprovalRequest>,
427 cost_records: HashMap<String, Vec<CostRecord>>,
428 artifacts: HashMap<String, Vec<Artifact>>,
429}
430
431impl MemoryStore {
432 pub fn new() -> Self {
433 Self::default()
434 }
435
436 pub fn save_to_path(&self, path: impl AsRef<Path>) -> Result<()> {
437 fs::write(path, encode_store(self)).map_err(|err| RuntimeError(err.to_string()))
438 }
439
440 pub fn load_from_path(path: impl AsRef<Path>) -> Result<Self> {
441 let body = fs::read_to_string(path).map_err(|err| RuntimeError(err.to_string()))?;
442 decode_store(&body)
443 }
444
445 pub fn create_run(&mut self, initial_state: State) -> (String, String) {
446 let run_id = new_id("run");
447 let session_id = new_id("sess");
448 let step_id = new_id("step");
449 let now = now_seconds();
450 self.runs.insert(
451 run_id.clone(),
452 Run {
453 run_id: run_id.clone(),
454 session_id: session_id.clone(),
455 status: "pending".to_string(),
456 state: initial_state.clone(),
457 state_version: 0,
458 created_at: now,
459 updated_at: now,
460 },
461 );
462 self.steps.insert(
463 step_id.clone(),
464 Step {
465 step_id: step_id.clone(),
466 run_id: run_id.clone(),
467 session_id: session_id.clone(),
468 status: "pending".to_string(),
469 owner: None,
470 lease_token: None,
471 lease_until: None,
472 attempt: 0,
473 state_version: 0,
474 checkpoint_id: None,
475 last_error_type: None,
476 last_error: None,
477 cancelled_at: None,
478 created_at: now,
479 updated_at: now,
480 },
481 );
482 let mut payload = State::new();
483 payload.insert("initial_state".to_string(), Value::Object(initial_state));
484 self.append_event(
485 &run_id,
486 Some(&session_id),
487 None,
488 "run_created",
489 payload,
490 None,
491 None,
492 None,
493 );
494 let mut step_payload = State::new();
495 step_payload.insert("step_id".to_string(), Value::String(step_id.clone()));
496 self.append_event(
497 &run_id,
498 Some(&session_id),
499 Some(&step_id),
500 "step_created",
501 step_payload,
502 None,
503 None,
504 None,
505 );
506 (run_id, step_id)
507 }
508
509 pub fn claim_step(
510 &mut self,
511 worker_id: &str,
512 run_id: &str,
513 lease_seconds: f64,
514 ) -> Result<StepClaim> {
515 let mut candidates: Vec<Step> = self
516 .steps
517 .values()
518 .filter(|step| {
519 (run_id.is_empty() || step.run_id == run_id)
520 && (step.status == "pending" || step.status == "retry_scheduled")
521 })
522 .cloned()
523 .collect();
524 candidates.sort_by(|a, b| {
525 a.created_at
526 .partial_cmp(&b.created_at)
527 .unwrap()
528 .then(a.step_id.cmp(&b.step_id))
529 });
530 let candidate = candidates
531 .first()
532 .ok_or_else(|| RuntimeError("agentledger: no runnable step".to_string()))?;
533 let now = now_seconds();
534 let step = self.steps.get_mut(&candidate.step_id).unwrap();
535 step.status = "running".to_string();
536 step.owner = Some(worker_id.to_string());
537 step.lease_token = Some(new_id("lease"));
538 step.lease_until = Some(now + lease_seconds);
539 step.attempt += 1;
540 step.updated_at = now;
541 let claim = StepClaim {
542 run_id: step.run_id.clone(),
543 session_id: step.session_id.clone(),
544 step_id: step.step_id.clone(),
545 attempt: step.attempt,
546 lease_token: step.lease_token.clone().unwrap(),
547 state_version: step.state_version,
548 lease_until: step.lease_until.unwrap(),
549 };
550 let run = self.runs.get_mut(&claim.run_id).unwrap();
551 run.status = "running".to_string();
552 run.updated_at = now;
553 let mut payload = State::new();
554 payload.insert(
555 "worker_id".to_string(),
556 Value::String(worker_id.to_string()),
557 );
558 payload.insert(
559 "lease_token".to_string(),
560 Value::String(claim.lease_token.clone()),
561 );
562 payload.insert("attempt".to_string(), Value::Number(claim.attempt as f64));
563 self.append_event(
564 &claim.run_id,
565 Some(&claim.session_id),
566 Some(&claim.step_id),
567 "step_claimed",
568 payload,
569 None,
570 None,
571 None,
572 );
573 Ok(claim)
574 }
575
576 pub fn load_state(&self, run_id: &str) -> Result<(State, u64, String)> {
577 let run = self
578 .runs
579 .get(run_id)
580 .ok_or_else(|| RuntimeError(format!("run not found: {run_id}")))?;
581 Ok((run.state.clone(), run.state_version, run.session_id.clone()))
582 }
583
584 pub fn recover_expired_leases(&mut self) -> usize {
585 let now = now_seconds();
586 let ids: Vec<String> = self.steps.keys().cloned().collect();
587 let mut recovered = 0;
588 for step_id in ids {
589 let should_recover = self
590 .steps
591 .get(&step_id)
592 .map(|step| {
593 step.status == "running" && step.lease_until.is_some_and(|until| until <= now)
594 })
595 .unwrap_or(false);
596 if !should_recover {
597 continue;
598 }
599 let (run_id, session_id, previous_owner, attempt) = {
600 let step = self.steps.get_mut(&step_id).unwrap();
601 let previous_owner = step.owner.clone().unwrap_or_default();
602 step.status = "retry_scheduled".to_string();
603 step.owner = None;
604 step.lease_token = None;
605 step.lease_until = None;
606 step.updated_at = now;
607 (
608 step.run_id.clone(),
609 step.session_id.clone(),
610 previous_owner,
611 step.attempt,
612 )
613 };
614 let run = self.runs.get_mut(&run_id).unwrap();
615 run.status = "retry_scheduled".to_string();
616 run.updated_at = now;
617 recovered += 1;
618 let mut payload = State::new();
619 payload.insert("previous_owner".to_string(), Value::String(previous_owner));
620 payload.insert("attempt".to_string(), Value::Number(attempt as f64));
621 self.append_event(
622 &run_id,
623 Some(&session_id),
624 Some(&step_id),
625 "lease_expired",
626 payload,
627 None,
628 None,
629 None,
630 );
631 let mut retry_payload = State::new();
632 retry_payload.insert("step_id".to_string(), Value::String(step_id.clone()));
633 retry_payload.insert(
634 "reason".to_string(),
635 Value::String("lease_expired".to_string()),
636 );
637 self.append_event(
638 &run_id,
639 Some(&session_id),
640 Some(&step_id),
641 "step_retry_scheduled",
642 retry_payload,
643 None,
644 None,
645 None,
646 );
647 }
648 recovered
649 }
650
651 pub fn cancel_run(&mut self, run_id: &str, reason: &str) -> Result<usize> {
652 let session_id = self
653 .runs
654 .get(run_id)
655 .ok_or_else(|| RuntimeError(format!("run not found: {run_id}")))?
656 .session_id
657 .clone();
658 let mut payload = State::new();
659 payload.insert("reason".to_string(), Value::String(reason.to_string()));
660 self.append_event(
661 run_id,
662 Some(&session_id),
663 None,
664 "run_cancel_requested",
665 payload,
666 None,
667 None,
668 None,
669 );
670 let ids: Vec<String> = self.steps.keys().cloned().collect();
671 let now = now_seconds();
672 let mut cancelled = 0;
673 for step_id in ids {
674 let should_cancel = self
675 .steps
676 .get(&step_id)
677 .map(|step| {
678 step.run_id == run_id
679 && !["completed", "failed", "cancelled"].contains(&step.status.as_str())
680 })
681 .unwrap_or(false);
682 if !should_cancel {
683 continue;
684 }
685 let step_session = {
686 let step = self.steps.get_mut(&step_id).unwrap();
687 step.status = "cancelled".to_string();
688 step.owner = None;
689 step.lease_token = None;
690 step.lease_until = None;
691 step.cancelled_at = Some(now);
692 step.updated_at = now;
693 step.session_id.clone()
694 };
695 cancelled += 1;
696 let mut step_payload = State::new();
697 step_payload.insert("reason".to_string(), Value::String(reason.to_string()));
698 self.append_event(
699 run_id,
700 Some(&step_session),
701 Some(&step_id),
702 "step_cancelled",
703 step_payload,
704 None,
705 None,
706 None,
707 );
708 }
709 let run = self.runs.get_mut(run_id).unwrap();
710 run.status = "cancelled".to_string();
711 run.updated_at = now;
712 let mut done_payload = State::new();
713 done_payload.insert("reason".to_string(), Value::String(reason.to_string()));
714 done_payload.insert(
715 "cancelled_steps".to_string(),
716 Value::Number(cancelled as f64),
717 );
718 self.append_event(
719 run_id,
720 Some(&session_id),
721 None,
722 "run_cancelled",
723 done_payload,
724 None,
725 None,
726 None,
727 );
728 Ok(cancelled)
729 }
730
731 pub fn commit_state_patch(
732 &mut self,
733 run_id: &str,
734 step_id: &str,
735 lease_token: &str,
736 base_version: u64,
737 patch: State,
738 ) -> Result<u64> {
739 self.validate_lease(step_id, lease_token)?;
740 let run = self
741 .runs
742 .get(run_id)
743 .ok_or_else(|| RuntimeError(format!("run not found: {run_id}")))?;
744 if run.state_version != base_version {
745 return Err(RuntimeError(format!(
746 "state version conflict: expected {base_version}, got {}",
747 run.state_version
748 )));
749 }
750 let now = now_seconds();
751 let new_version = base_version + 1;
752 let session_id = self.steps.get(step_id).unwrap().session_id.clone();
753 let run = self.runs.get_mut(run_id).unwrap();
754 run.state = merge_patch(&run.state, &patch);
755 run.state_version = new_version;
756 run.status = "completed".to_string();
757 run.updated_at = now;
758 let step = self.steps.get_mut(step_id).unwrap();
759 step.status = "completed".to_string();
760 step.state_version = new_version;
761 step.checkpoint_id = Some(format!("ckpt:{run_id}:{step_id}:{}", step.attempt));
762 step.updated_at = now;
763 let mut payload = State::new();
764 payload.insert("patch".to_string(), Value::Object(patch));
765 payload.insert(
766 "state_version".to_string(),
767 Value::Number(new_version as f64),
768 );
769 self.append_event(
770 run_id,
771 Some(&session_id),
772 Some(step_id),
773 "state_patch_committed",
774 payload,
775 None,
776 Some(new_version),
777 None,
778 );
779 let mut complete_payload = State::new();
780 complete_payload.insert("step_id".to_string(), Value::String(step_id.to_string()));
781 self.append_event(
782 run_id,
783 Some(&session_id),
784 Some(step_id),
785 "step_completed",
786 complete_payload,
787 None,
788 Some(new_version),
789 None,
790 );
791 Ok(new_version)
792 }
793
794 pub fn mark_retry(&mut self, run_id: &str, step_id: &str, error_type: &str, message: &str) {
795 let now = now_seconds();
796 let session_id = {
797 let step = self.steps.get_mut(step_id).unwrap();
798 step.status = "retry_scheduled".to_string();
799 step.owner = None;
800 step.lease_token = None;
801 step.lease_until = None;
802 step.last_error_type = Some(error_type.to_string());
803 step.last_error = Some(message.to_string());
804 step.updated_at = now;
805 step.session_id.clone()
806 };
807 let run = self.runs.get_mut(run_id).unwrap();
808 run.status = "retry_scheduled".to_string();
809 run.updated_at = now;
810 let mut classified = State::new();
811 classified.insert("error".to_string(), Value::String(message.to_string()));
812 classified.insert(
813 "error_type".to_string(),
814 Value::String(error_type.to_string()),
815 );
816 classified.insert("retryable".to_string(), Value::Bool(true));
817 classified.insert("source".to_string(), Value::String("agent".to_string()));
818 self.append_event(
819 run_id,
820 Some(&session_id),
821 Some(step_id),
822 "failure_classified",
823 classified,
824 None,
825 None,
826 None,
827 );
828 let mut payload = State::new();
829 payload.insert("error".to_string(), Value::String(message.to_string()));
830 payload.insert(
831 "error_type".to_string(),
832 Value::String(error_type.to_string()),
833 );
834 self.append_event(
835 run_id,
836 Some(&session_id),
837 Some(step_id),
838 "error_raised",
839 payload,
840 None,
841 None,
842 None,
843 );
844 let mut retry_payload = State::new();
845 retry_payload.insert("step_id".to_string(), Value::String(step_id.to_string()));
846 self.append_event(
847 run_id,
848 Some(&session_id),
849 Some(step_id),
850 "step_retry_scheduled",
851 retry_payload,
852 None,
853 None,
854 None,
855 );
856 }
857
858 pub fn mark_waiting_human(
859 &mut self,
860 run_id: &str,
861 step_id: &str,
862 reason: &str,
863 approval_id: &str,
864 ) {
865 let now = now_seconds();
866 let session_id = {
867 let step = self.steps.get_mut(step_id).unwrap();
868 step.status = "waiting_human".to_string();
869 step.owner = None;
870 step.lease_token = None;
871 step.lease_until = None;
872 step.last_error_type = Some("ApprovalRequired".to_string());
873 step.last_error = Some(reason.to_string());
874 step.updated_at = now;
875 step.session_id.clone()
876 };
877 let run = self.runs.get_mut(run_id).unwrap();
878 run.status = "waiting_human".to_string();
879 run.updated_at = now;
880 let mut payload = State::new();
881 payload.insert("reason".to_string(), Value::String(reason.to_string()));
882 payload.insert(
883 "approval_id".to_string(),
884 Value::String(approval_id.to_string()),
885 );
886 self.append_event(
887 run_id,
888 Some(&session_id),
889 Some(step_id),
890 "step_waiting_human",
891 payload,
892 None,
893 None,
894 None,
895 );
896 }
897
898 pub fn mark_failed(&mut self, run_id: &str, step_id: &str, error_type: &str, message: &str) {
899 let now = now_seconds();
900 let session_id = {
901 let step = self.steps.get_mut(step_id).unwrap();
902 step.status = "failed".to_string();
903 step.owner = None;
904 step.lease_token = None;
905 step.lease_until = None;
906 step.last_error_type = Some(error_type.to_string());
907 step.last_error = Some(message.to_string());
908 step.updated_at = now;
909 step.session_id.clone()
910 };
911 let run = self.runs.get_mut(run_id).unwrap();
912 run.status = "failed".to_string();
913 run.updated_at = now;
914 let mut classified = State::new();
915 classified.insert("error".to_string(), Value::String(message.to_string()));
916 classified.insert(
917 "error_type".to_string(),
918 Value::String(error_type.to_string()),
919 );
920 classified.insert("retryable".to_string(), Value::Bool(false));
921 classified.insert(
922 "source".to_string(),
923 Value::String(failure_source(error_type).to_string()),
924 );
925 self.append_event(
926 run_id,
927 Some(&session_id),
928 Some(step_id),
929 "failure_classified",
930 classified,
931 None,
932 None,
933 None,
934 );
935 let mut error = State::new();
936 error.insert("error".to_string(), Value::String(message.to_string()));
937 error.insert(
938 "error_type".to_string(),
939 Value::String(error_type.to_string()),
940 );
941 self.append_event(
942 run_id,
943 Some(&session_id),
944 Some(step_id),
945 "error_raised",
946 error,
947 None,
948 None,
949 None,
950 );
951 let mut payload = State::new();
952 payload.insert("step_id".to_string(), Value::String(step_id.to_string()));
953 payload.insert(
954 "error_type".to_string(),
955 Value::String(error_type.to_string()),
956 );
957 self.append_event(
958 run_id,
959 Some(&session_id),
960 Some(step_id),
961 "step_failed",
962 payload,
963 None,
964 None,
965 None,
966 );
967 }
968
969 pub fn append_event(
970 &mut self,
971 run_id: &str,
972 session_id: Option<&str>,
973 step_id: Option<&str>,
974 event_type: &str,
975 payload: State,
976 agent_role: Option<&str>,
977 state_version: Option<u64>,
978 causal_token: Option<&str>,
979 ) -> Event {
980 let events = self.events.entry(run_id.to_string()).or_default();
981 let payload_ref = format_state(&payload);
982 let event = Event {
983 event_id: new_id("evt"),
984 run_id: run_id.to_string(),
985 session_id: session_id.map(str::to_string),
986 step_id: step_id.map(str::to_string),
987 seq: events.len() as u64 + 1,
988 event_type: event_type.to_string(),
989 timestamp: now_seconds(),
990 agent_role: agent_role.map(str::to_string),
991 state_version,
992 causal_token: causal_token.map(str::to_string),
993 payload_hash: stable_hash(&payload_ref),
994 payload_ref,
995 payload,
996 };
997 events.push(event.clone());
998 event
999 }
1000
1001 pub fn reserve_ledger(&mut self, key: &str, entry: ToolLedgerEntry) -> Option<ToolLedgerEntry> {
1002 if let Some(existing) = self.tool_ledger.get(key) {
1003 return Some(existing.clone());
1004 }
1005 self.tool_ledger.insert(key.to_string(), entry);
1006 None
1007 }
1008
1009 pub fn update_ledger(
1010 &mut self,
1011 key: &str,
1012 status: &str,
1013 response: Option<Value>,
1014 error_type: Option<String>,
1015 ) {
1016 if let Some(entry) = self.tool_ledger.get_mut(key) {
1017 entry.status = status.to_string();
1018 entry.response = response.clone();
1019 entry.response_ref = response.as_ref().map(format_value);
1020 entry.response_hash = entry.response_ref.as_ref().map(|value| stable_hash(value));
1021 entry.error_type = error_type;
1022 entry.updated_at = now_seconds();
1023 }
1024 }
1025
1026 pub fn request_approval(
1027 &mut self,
1028 approval_key: &str,
1029 run_id: &str,
1030 session_id: &str,
1031 step_id: &str,
1032 tool_name: &str,
1033 risk_level: &str,
1034 reason: &str,
1035 request_hash: &str,
1036 request_ref: &str,
1037 requested_by: &str,
1038 ) -> ApprovalRequest {
1039 if let Some(existing) = self.approval_requests.get(approval_key) {
1040 return existing.clone();
1041 }
1042 let now = now_seconds();
1043 let approval = ApprovalRequest {
1044 approval_id: new_id("approval"),
1045 approval_key: approval_key.to_string(),
1046 run_id: run_id.to_string(),
1047 session_id: session_id.to_string(),
1048 step_id: step_id.to_string(),
1049 tool_name: tool_name.to_string(),
1050 risk_level: risk_level.to_string(),
1051 status: "PENDING".to_string(),
1052 reason: reason.to_string(),
1053 request_hash: request_hash.to_string(),
1054 request_ref: request_ref.to_string(),
1055 requested_by: requested_by.to_string(),
1056 approved_by: None,
1057 decision_reason: None,
1058 created_at: now,
1059 updated_at: now,
1060 };
1061 self.approval_requests
1062 .insert(approval_key.to_string(), approval.clone());
1063 approval
1064 }
1065
1066 pub fn approval_for_key(&self, approval_key: &str) -> Option<ApprovalRequest> {
1067 self.approval_requests.get(approval_key).cloned()
1068 }
1069
1070 pub fn approval_requests(&self, run_id: &str) -> Vec<ApprovalRequest> {
1071 self.approval_requests
1072 .values()
1073 .filter(|item| run_id.is_empty() || item.run_id == run_id)
1074 .cloned()
1075 .collect()
1076 }
1077
1078 pub fn approve_request(
1079 &mut self,
1080 approval_id: &str,
1081 approver: &str,
1082 reason: &str,
1083 ) -> Result<ApprovalRequest> {
1084 self.decide_approval(approval_id, "APPROVED", approver, reason)
1085 }
1086
1087 pub fn deny_request(
1088 &mut self,
1089 approval_id: &str,
1090 approver: &str,
1091 reason: &str,
1092 ) -> Result<ApprovalRequest> {
1093 self.decide_approval(approval_id, "DENIED", approver, reason)
1094 }
1095
1096 fn decide_approval(
1097 &mut self,
1098 approval_id: &str,
1099 status: &str,
1100 approver: &str,
1101 reason: &str,
1102 ) -> Result<ApprovalRequest> {
1103 let key = self
1104 .approval_requests
1105 .iter()
1106 .find(|(_, item)| item.approval_id == approval_id)
1107 .map(|(key, _)| key.clone())
1108 .ok_or_else(|| RuntimeError(format!("approval not found: {approval_id}")))?;
1109 let mut approval = self.approval_requests.get(&key).unwrap().clone();
1110 approval.status = status.to_string();
1111 approval.approved_by = Some(approver.to_string());
1112 approval.decision_reason = Some(reason.to_string());
1113 approval.updated_at = now_seconds();
1114 self.approval_requests.insert(key, approval.clone());
1115 let mut payload = State::new();
1116 payload.insert(
1117 "approval_id".to_string(),
1118 Value::String(approval_id.to_string()),
1119 );
1120 payload.insert(
1121 "tool".to_string(),
1122 Value::String(approval.tool_name.clone()),
1123 );
1124 payload.insert("status".to_string(), Value::String(status.to_string()));
1125 self.append_event(
1126 &approval.run_id,
1127 Some(&approval.session_id),
1128 Some(&approval.step_id),
1129 "tool_approval_decided",
1130 payload,
1131 None,
1132 None,
1133 None,
1134 );
1135 if self
1136 .steps
1137 .get(&approval.step_id)
1138 .map(|step| step.status.as_str())
1139 == Some("waiting_human")
1140 {
1141 if status == "APPROVED" {
1142 let step = self.steps.get_mut(&approval.step_id).unwrap();
1143 step.status = "pending".to_string();
1144 step.owner = None;
1145 step.lease_token = None;
1146 step.lease_until = None;
1147 step.updated_at = now_seconds();
1148 let run = self.runs.get_mut(&approval.run_id).unwrap();
1149 run.status = "pending".to_string();
1150 let mut retry = State::new();
1151 retry.insert(
1152 "step_id".to_string(),
1153 Value::String(approval.step_id.clone()),
1154 );
1155 retry.insert(
1156 "reason".to_string(),
1157 Value::String("approval_granted".to_string()),
1158 );
1159 self.append_event(
1160 &approval.run_id,
1161 Some(&approval.session_id),
1162 Some(&approval.step_id),
1163 "step_retry_scheduled",
1164 retry,
1165 None,
1166 None,
1167 None,
1168 );
1169 } else if status == "DENIED" {
1170 self.mark_failed(
1171 &approval.run_id,
1172 &approval.step_id,
1173 "ApprovalDenied",
1174 reason,
1175 );
1176 }
1177 }
1178 Ok(approval)
1179 }
1180
1181 pub fn record_cost(
1182 &mut self,
1183 run_id: &str,
1184 session_id: &str,
1185 step_id: &str,
1186 category: &str,
1187 name: &str,
1188 amount: f64,
1189 unit: &str,
1190 metadata: State,
1191 ) -> String {
1192 let cost_id = new_id("cost");
1193 let record = CostRecord {
1194 cost_id: cost_id.clone(),
1195 run_id: run_id.to_string(),
1196 session_id: session_id.to_string(),
1197 step_id: step_id.to_string(),
1198 category: category.to_string(),
1199 name: name.to_string(),
1200 amount,
1201 unit: unit.to_string(),
1202 metadata,
1203 created_at: now_seconds(),
1204 };
1205 self.cost_records
1206 .entry(run_id.to_string())
1207 .or_default()
1208 .push(record);
1209 let mut payload = State::new();
1210 payload.insert("cost_id".to_string(), Value::String(cost_id.clone()));
1211 payload.insert("category".to_string(), Value::String(category.to_string()));
1212 payload.insert("name".to_string(), Value::String(name.to_string()));
1213 payload.insert("amount".to_string(), Value::Number(amount));
1214 payload.insert("unit".to_string(), Value::String(unit.to_string()));
1215 self.append_event(
1216 run_id,
1217 Some(session_id),
1218 Some(step_id),
1219 "cost_recorded",
1220 payload,
1221 None,
1222 None,
1223 None,
1224 );
1225 cost_id
1226 }
1227
1228 pub fn cost_records(&self, run_id: &str) -> Vec<CostRecord> {
1229 self.cost_records.get(run_id).cloned().unwrap_or_default()
1230 }
1231
1232 pub fn cost_summary(&self, run_id: &str) -> CostSummary {
1233 let mut summary = CostSummary::default();
1234 for record in self.cost_records(run_id) {
1235 add_cost(&mut summary, &record);
1236 }
1237 summary
1238 }
1239
1240 pub fn create_artifact(
1241 &mut self,
1242 run_id: &str,
1243 step_id: Option<&str>,
1244 name: &str,
1245 content: State,
1246 metadata: State,
1247 ) -> Artifact {
1248 let blob_ref = format_state(&content);
1249 let artifact = Artifact {
1250 artifact_id: new_id("art"),
1251 run_id: run_id.to_string(),
1252 step_id: step_id.map(str::to_string),
1253 name: name.to_string(),
1254 blob_hash: format!("sha256:{}", stable_hash(&blob_ref)),
1255 blob_ref,
1256 metadata,
1257 created_at: now_seconds(),
1258 };
1259 self.artifacts
1260 .entry(run_id.to_string())
1261 .or_default()
1262 .push(artifact.clone());
1263 artifact
1264 }
1265
1266 pub fn artifacts(&self, run_id: &str) -> Vec<Artifact> {
1267 let mut items = self.artifacts.get(run_id).cloned().unwrap_or_default();
1268 items.sort_by(|a, b| {
1269 a.created_at
1270 .partial_cmp(&b.created_at)
1271 .unwrap()
1272 .then(a.artifact_id.cmp(&b.artifact_id))
1273 });
1274 items
1275 }
1276
1277 pub fn validate_lease(&self, step_id: &str, lease_token: &str) -> Result<()> {
1278 let step = self
1279 .steps
1280 .get(step_id)
1281 .ok_or_else(|| RuntimeError(format!("step not found: {step_id}")))?;
1282 if step.status != "running" || step.lease_token.as_deref() != Some(lease_token) {
1283 return Err(RuntimeError("invalid or stale lease token".to_string()));
1284 }
1285 if step.lease_until.is_some_and(|until| until <= now_seconds()) {
1286 return Err(RuntimeError("lease expired".to_string()));
1287 }
1288 Ok(())
1289 }
1290
1291 pub fn final_state(&self, run_id: &str) -> Result<State> {
1292 Ok(self.load_state(run_id)?.0)
1293 }
1294
1295 pub fn run(&self, run_id: &str) -> Result<Run> {
1296 self.runs
1297 .get(run_id)
1298 .cloned()
1299 .ok_or_else(|| RuntimeError(format!("run not found: {run_id}")))
1300 }
1301
1302 pub fn steps(&self, run_id: &str) -> Vec<Step> {
1303 self.steps
1304 .values()
1305 .filter(|step| step.run_id == run_id)
1306 .cloned()
1307 .collect()
1308 }
1309
1310 pub fn events(&self, run_id: &str) -> Vec<Event> {
1311 self.events.get(run_id).cloned().unwrap_or_default()
1312 }
1313
1314 pub fn ledger(&self, run_id: &str) -> Vec<ToolLedgerEntry> {
1315 self.tool_ledger
1316 .values()
1317 .filter(|entry| entry.run_id == run_id)
1318 .cloned()
1319 .collect()
1320 }
1321}
1322
1323pub struct ToolRegistry {
1324 tools: HashMap<String, ToolSpec>,
1325}
1326
1327impl ToolRegistry {
1328 pub fn new() -> Self {
1329 Self {
1330 tools: HashMap::new(),
1331 }
1332 }
1333
1334 pub fn register(&mut self, spec: ToolSpec) {
1335 self.tools.insert(spec.name.clone(), spec);
1336 }
1337
1338 pub fn get(&self, name: &str) -> Result<&ToolSpec> {
1339 self.tools
1340 .get(name)
1341 .ok_or_else(|| RuntimeError(format!("tool not registered: {name}")))
1342 }
1343}
1344
1345impl Default for ToolRegistry {
1346 fn default() -> Self {
1347 Self::new()
1348 }
1349}
1350
1351pub struct Runtime {
1352 pub store: MemoryStore,
1353 pub registry: ToolRegistry,
1354 pub budget: BudgetLimits,
1355 pub sandbox: Box<dyn SandboxExecutor>,
1356}
1357
1358impl Runtime {
1359 pub fn new() -> Self {
1360 Self {
1361 store: MemoryStore::new(),
1362 registry: ToolRegistry::new(),
1363 budget: BudgetLimits::default(),
1364 sandbox: Box::new(DisabledSandboxExecutor),
1365 }
1366 }
1367
1368 pub fn register_tool(&mut self, spec: ToolSpec) {
1369 self.registry.register(spec);
1370 }
1371
1372 pub fn set_budget(&mut self, budget: BudgetLimits) {
1373 self.budget = budget;
1374 }
1375
1376 pub fn set_sandbox(&mut self, sandbox: Box<dyn SandboxExecutor>) {
1377 self.sandbox = sandbox;
1378 }
1379
1380 pub fn create_run(&mut self, initial_state: State) -> (String, String) {
1381 self.store.create_run(initial_state)
1382 }
1383
1384 pub fn run_once(
1385 &mut self,
1386 run_id: &str,
1387 worker_id: &str,
1388 agent_role: &str,
1389 lease_seconds: f64,
1390 agent: AgentFunc,
1391 ) -> Result<bool> {
1392 let claim = match self.store.claim_step(worker_id, run_id, lease_seconds) {
1393 Ok(claim) => claim,
1394 Err(err) if err.0.contains("no runnable step") => return Ok(false),
1395 Err(err) => return Err(err),
1396 };
1397 let (state, version, session_id) = self.store.load_state(&claim.run_id)?;
1398 let mut payload = State::new();
1399 payload.insert(
1400 "agent_role".to_string(),
1401 Value::String(agent_role.to_string()),
1402 );
1403 payload.insert("attempt".to_string(), Value::Number(claim.attempt as f64));
1404 self.store.append_event(
1405 &claim.run_id,
1406 Some(&session_id),
1407 Some(&claim.step_id),
1408 "agent_started",
1409 payload,
1410 Some(agent_role),
1411 Some(version),
1412 None,
1413 );
1414 let mut ctx = AgentContext {
1415 run_id: claim.run_id.clone(),
1416 session_id,
1417 step_id: claim.step_id.clone(),
1418 agent_role: agent_role.to_string(),
1419 lease_token: claim.lease_token.clone(),
1420 attempt: claim.attempt,
1421 state_version: version,
1422 pending_patch: State::new(),
1423 };
1424 match agent(&mut ctx, state) {
1425 Ok(()) => {
1426 self.store.commit_state_patch(
1427 &claim.run_id,
1428 &claim.step_id,
1429 &claim.lease_token,
1430 version,
1431 ctx.pending_patch,
1432 )?;
1433 Ok(true)
1434 }
1435 Err(err) if err.0 == "retryable" => {
1436 self.store.mark_retry(
1437 &claim.run_id,
1438 &claim.step_id,
1439 "RetryableAgentError",
1440 "retryable",
1441 );
1442 Ok(false)
1443 }
1444 Err(err) if err.0.starts_with("approval required:") => {
1445 let approval_id = err.0.trim_start_matches("approval required:");
1446 self.store
1447 .mark_waiting_human(&claim.run_id, &claim.step_id, &err.0, approval_id);
1448 Ok(false)
1449 }
1450 Err(err) => {
1451 self.store.mark_failed(
1452 &claim.run_id,
1453 &claim.step_id,
1454 classify_runtime_error(&err.0),
1455 &err.0,
1456 );
1457 Err(err)
1458 }
1459 }
1460 }
1461
1462 pub fn call_tool(&mut self, ctx: &AgentContext, tool_name: &str, args: State) -> Result<Value> {
1463 let (
1464 version,
1465 side_effect,
1466 risk_level,
1467 idempotency_required,
1468 approval_required,
1469 sandbox_required,
1470 sandbox_executor,
1471 sandbox_policy,
1472 input_schema,
1473 output_schema,
1474 ) = {
1475 let spec = self.registry.get(tool_name)?;
1476 (
1477 spec.version.clone(),
1478 spec.side_effect.clone(),
1479 spec.risk_level.clone(),
1480 spec.idempotency_required,
1481 spec.approval_required,
1482 spec.sandbox_required,
1483 spec.sandbox_executor.clone(),
1484 spec.sandbox_policy.clone(),
1485 spec.input_schema.clone(),
1486 spec.output_schema.clone(),
1487 )
1488 };
1489 let mut request = State::new();
1490 request.insert("tool".to_string(), Value::String(tool_name.to_string()));
1491 request.insert("args".to_string(), Value::Object(args.clone()));
1492 let request_ref = format_state(&request);
1493 let request_hash = stable_hash(&request_ref);
1494 let causal_token = format!(
1495 "{}:{}:{}:{}",
1496 ctx.run_id, ctx.step_id, ctx.attempt, ctx.state_version
1497 );
1498 let idempotency_key = format!(
1499 "{}:{}:{}:{}",
1500 ctx.run_id, ctx.step_id, tool_name, request_hash
1501 );
1502 let managed = side_effect != "none" || idempotency_required;
1503 self.store.append_event(
1504 &ctx.run_id,
1505 Some(&ctx.session_id),
1506 Some(&ctx.step_id),
1507 "tool_call_requested",
1508 request,
1509 Some(&ctx.agent_role),
1510 Some(ctx.state_version),
1511 Some(&causal_token),
1512 );
1513 if let Some(schema) = input_schema.as_ref() {
1514 if let Err(err) = validate_tool_schema(schema, &Value::Object(args.clone()), "$arg") {
1515 let mut failed = State::new();
1516 failed.insert("tool".to_string(), Value::String(tool_name.to_string()));
1517 failed.insert("error".to_string(), Value::String(err.0.clone()));
1518 failed.insert(
1519 "phase".to_string(),
1520 Value::String("input_validation".to_string()),
1521 );
1522 self.store.append_event(
1523 &ctx.run_id,
1524 Some(&ctx.session_id),
1525 Some(&ctx.step_id),
1526 "tool_call_failed",
1527 failed,
1528 Some(&ctx.agent_role),
1529 Some(ctx.state_version),
1530 Some(&causal_token),
1531 );
1532 return Err(err);
1533 }
1534 }
1535
1536 let mut allowed = !is_high_risk(&risk_level);
1537 let mut reason = if allowed {
1538 "default allow for low/medium risk in local runtime".to_string()
1539 } else {
1540 "high-risk tool denied by default".to_string()
1541 };
1542 if let Some(approval) = self.store.approval_for_key(&idempotency_key) {
1543 if approval.status == "DENIED" {
1544 reason = format!("approval denied for tool {tool_name}");
1545 self.record_permission(ctx, tool_name, false, &reason, &causal_token);
1546 return Err(RuntimeError(reason));
1547 }
1548 if approval.status == "APPROVED" {
1549 allowed = true;
1550 reason = format!(
1551 "approved by {}",
1552 approval
1553 .approved_by
1554 .unwrap_or_else(|| "operator".to_string())
1555 );
1556 }
1557 } else if approval_required {
1558 let approval = self.store.request_approval(
1559 &idempotency_key,
1560 &ctx.run_id,
1561 &ctx.session_id,
1562 &ctx.step_id,
1563 tool_name,
1564 &risk_level,
1565 "tool requires approval",
1566 &request_hash,
1567 &request_ref,
1568 &ctx.agent_role,
1569 );
1570 self.record_permission(ctx, tool_name, false, "approval required", &causal_token);
1571 let mut payload = State::new();
1572 payload.insert("tool".to_string(), Value::String(tool_name.to_string()));
1573 payload.insert(
1574 "approval_id".to_string(),
1575 Value::String(approval.approval_id.clone()),
1576 );
1577 payload.insert(
1578 "approval_key".to_string(),
1579 Value::String(idempotency_key.clone()),
1580 );
1581 payload.insert("risk_level".to_string(), Value::String(risk_level));
1582 self.store.append_event(
1583 &ctx.run_id,
1584 Some(&ctx.session_id),
1585 Some(&ctx.step_id),
1586 "tool_approval_required",
1587 payload,
1588 Some(&ctx.agent_role),
1589 Some(ctx.state_version),
1590 Some(&causal_token),
1591 );
1592 return Err(RuntimeError(format!(
1593 "approval required:{}",
1594 approval.approval_id
1595 )));
1596 }
1597 self.record_permission(ctx, tool_name, allowed, &reason, &causal_token);
1598 if !allowed {
1599 return Err(RuntimeError(reason));
1600 }
1601 if let Some(max) = self.budget.max_tool_calls {
1602 let used = self.store.cost_summary(&ctx.run_id).tool_calls;
1603 if used >= max {
1604 let message = format!("tool call budget exceeded: {used}/{max}");
1605 let mut payload = State::new();
1606 payload.insert("category".to_string(), Value::String("tool".to_string()));
1607 payload.insert("tool".to_string(), Value::String(tool_name.to_string()));
1608 payload.insert("error".to_string(), Value::String(message.clone()));
1609 self.store.append_event(
1610 &ctx.run_id,
1611 Some(&ctx.session_id),
1612 Some(&ctx.step_id),
1613 "budget_check_failed",
1614 payload,
1615 Some(&ctx.agent_role),
1616 Some(ctx.state_version),
1617 Some(&causal_token),
1618 );
1619 return Err(RuntimeError(message));
1620 }
1621 }
1622 if managed {
1623 let entry = ToolLedgerEntry {
1624 ledger_id: new_id("ledger"),
1625 run_id: ctx.run_id.clone(),
1626 session_id: ctx.session_id.clone(),
1627 step_id: ctx.step_id.clone(),
1628 tool_name: tool_name.to_string(),
1629 tool_version: version,
1630 tool_call_id: new_id("toolcall"),
1631 idempotency_key: idempotency_key.clone(),
1632 causal_token: causal_token.clone(),
1633 request_hash: request_hash.clone(),
1634 request_ref: request_ref.clone(),
1635 status: "RESERVED".to_string(),
1636 external_id: None,
1637 response_hash: None,
1638 response_ref: None,
1639 error_type: None,
1640 response: None,
1641 created_at: now_seconds(),
1642 updated_at: now_seconds(),
1643 };
1644 if let Some(existing) = self.store.reserve_ledger(&idempotency_key, entry) {
1645 if existing.status == "SUCCEEDED" {
1646 let mut payload = State::new();
1647 payload.insert("tool".to_string(), Value::String(tool_name.to_string()));
1648 payload.insert("replayed_from_ledger".to_string(), Value::Bool(true));
1649 self.store.append_event(
1650 &ctx.run_id,
1651 Some(&ctx.session_id),
1652 Some(&ctx.step_id),
1653 "tool_call_completed",
1654 payload,
1655 Some(&ctx.agent_role),
1656 Some(ctx.state_version),
1657 Some(&causal_token),
1658 );
1659 self.store.record_cost(
1660 &ctx.run_id,
1661 &ctx.session_id,
1662 &ctx.step_id,
1663 "tool",
1664 tool_name,
1665 1.0,
1666 "call",
1667 State::new(),
1668 );
1669 return existing
1670 .response
1671 .ok_or_else(|| RuntimeError("missing ledger response".to_string()));
1672 }
1673 return Err(RuntimeError(
1674 "tool side effect already in progress".to_string(),
1675 ));
1676 }
1677 self.store
1678 .update_ledger(&idempotency_key, "RUNNING", None, None);
1679 }
1680 if sandbox_required {
1681 let executor = if sandbox_executor.is_empty() { "default".to_string() } else { sandbox_executor.clone() };
1682 let network = match sandbox_policy.get("network") {
1683 Some(Value::String(value)) => value.clone(),
1684 _ => "deny".to_string(),
1685 };
1686 let filesystem = match sandbox_policy.get("filesystem") {
1687 Some(Value::String(value)) => value.clone(),
1688 _ => "read-only".to_string(),
1689 };
1690 let timeout_seconds = match sandbox_policy.get("timeout_seconds") {
1691 Some(Value::Number(value)) if *value > 0.0 => *value as u64,
1692 _ => 30,
1693 };
1694 let policy = SandboxPolicy {
1695 tool_name: tool_name.to_string(),
1696 run_id: ctx.run_id.clone(),
1697 step_id: ctx.step_id.clone(),
1698 executor: executor.clone(),
1699 network,
1700 filesystem,
1701 timeout_seconds,
1702 extra: sandbox_policy.clone(),
1703 };
1704 let mut sandbox_payload = State::new();
1705 sandbox_payload.insert("tool_name".to_string(), Value::String(tool_name.to_string()));
1706 sandbox_payload.insert("executor".to_string(), Value::String(executor));
1707 sandbox_payload.insert("network".to_string(), Value::String(policy.network.clone()));
1708 sandbox_payload.insert("filesystem".to_string(), Value::String(policy.filesystem.clone()));
1709 sandbox_payload.insert("timeout_seconds".to_string(), Value::Number(policy.timeout_seconds as f64));
1710 self.store.append_event(
1711 &ctx.run_id,
1712 Some(&ctx.session_id),
1713 Some(&ctx.step_id),
1714 "sandbox_started",
1715 sandbox_payload,
1716 Some(&ctx.agent_role),
1717 Some(ctx.state_version),
1718 Some(&causal_token),
1719 );
1720 let result = self.sandbox.run_tool(args, &policy);
1721 let mut completed = State::new();
1722 completed.insert("ok".to_string(), Value::Bool(result.ok));
1723 completed.insert("metadata".to_string(), Value::Object(result.metadata.clone()));
1724 if let Some(error) = &result.error {
1725 completed.insert("error".to_string(), Value::String(error.clone()));
1726 }
1727 self.store.append_event(
1728 &ctx.run_id,
1729 Some(&ctx.session_id),
1730 Some(&ctx.step_id),
1731 "sandbox_completed",
1732 completed,
1733 Some(&ctx.agent_role),
1734 Some(ctx.state_version),
1735 Some(&causal_token),
1736 );
1737 if !result.ok {
1738 let message = result.error.unwrap_or_else(|| "sandboxed tool failed".to_string());
1739 let mut failed = State::new();
1740 failed.insert("tool".to_string(), Value::String(tool_name.to_string()));
1741 failed.insert("error".to_string(), Value::String(message.clone()));
1742 self.store.append_event(
1743 &ctx.run_id,
1744 Some(&ctx.session_id),
1745 Some(&ctx.step_id),
1746 "tool_call_failed",
1747 failed,
1748 Some(&ctx.agent_role),
1749 Some(ctx.state_version),
1750 Some(&causal_token),
1751 );
1752 return Err(RuntimeError(message));
1753 }
1754 let value = result.output;
1755 if let Some(schema) = output_schema.as_ref() {
1756 validate_tool_schema(schema, &value, "$result")?;
1757 }
1758 if managed {
1759 self.store.update_ledger(&idempotency_key, "SUCCEEDED", Some(value.clone()), None);
1760 }
1761 let mut payload = State::new();
1762 payload.insert("tool".to_string(), Value::String(tool_name.to_string()));
1763 payload.insert("idempotency_key".to_string(), Value::String(idempotency_key));
1764 self.store.append_event(
1765 &ctx.run_id,
1766 Some(&ctx.session_id),
1767 Some(&ctx.step_id),
1768 "tool_call_completed",
1769 payload,
1770 Some(&ctx.agent_role),
1771 Some(ctx.state_version),
1772 Some(&causal_token),
1773 );
1774 self.store.record_cost(
1775 &ctx.run_id,
1776 &ctx.session_id,
1777 &ctx.step_id,
1778 "tool",
1779 tool_name,
1780 1.0,
1781 "call",
1782 State::new(),
1783 );
1784 return Ok(value);
1785 }
1786 let result = {
1787 let spec = self.registry.get(tool_name)?;
1788 (spec.func)(args)
1789 };
1790 match result {
1791 Ok(value) => {
1792 if let Some(schema) = output_schema.as_ref() {
1793 if let Err(err) = validate_tool_schema(schema, &value, "$result") {
1794 if managed {
1795 self.store.update_ledger(
1796 &idempotency_key,
1797 "PENDING_VERIFICATION",
1798 None,
1799 Some("ToolOutputValidationError".to_string()),
1800 );
1801 }
1802 let mut failed = State::new();
1803 failed.insert("tool".to_string(), Value::String(tool_name.to_string()));
1804 failed.insert("error".to_string(), Value::String(err.0.clone()));
1805 failed.insert(
1806 "phase".to_string(),
1807 Value::String("output_validation".to_string()),
1808 );
1809 self.store.append_event(
1810 &ctx.run_id,
1811 Some(&ctx.session_id),
1812 Some(&ctx.step_id),
1813 "tool_call_failed",
1814 failed,
1815 Some(&ctx.agent_role),
1816 Some(ctx.state_version),
1817 Some(&causal_token),
1818 );
1819 return Err(err);
1820 }
1821 }
1822 if managed {
1823 self.store.update_ledger(
1824 &idempotency_key,
1825 "SUCCEEDED",
1826 Some(value.clone()),
1827 None,
1828 );
1829 }
1830 let mut payload = State::new();
1831 payload.insert("tool".to_string(), Value::String(tool_name.to_string()));
1832 payload.insert(
1833 "idempotency_key".to_string(),
1834 Value::String(idempotency_key),
1835 );
1836 self.store.append_event(
1837 &ctx.run_id,
1838 Some(&ctx.session_id),
1839 Some(&ctx.step_id),
1840 "tool_call_completed",
1841 payload,
1842 Some(&ctx.agent_role),
1843 Some(ctx.state_version),
1844 Some(&causal_token),
1845 );
1846 self.store.record_cost(
1847 &ctx.run_id,
1848 &ctx.session_id,
1849 &ctx.step_id,
1850 "tool",
1851 tool_name,
1852 1.0,
1853 "call",
1854 State::new(),
1855 );
1856 Ok(value)
1857 }
1858 Err(err) => {
1859 if managed {
1860 self.store.update_ledger(
1861 &idempotency_key,
1862 "PENDING_VERIFICATION",
1863 None,
1864 Some("RuntimeError".to_string()),
1865 );
1866 }
1867 let mut payload = State::new();
1868 payload.insert("tool".to_string(), Value::String(tool_name.to_string()));
1869 payload.insert("error".to_string(), Value::String(err.0.clone()));
1870 self.store.append_event(
1871 &ctx.run_id,
1872 Some(&ctx.session_id),
1873 Some(&ctx.step_id),
1874 "tool_call_failed",
1875 payload,
1876 Some(&ctx.agent_role),
1877 Some(ctx.state_version),
1878 Some(&causal_token),
1879 );
1880 Err(err)
1881 }
1882 }
1883 }
1884
1885 fn record_permission(
1886 &mut self,
1887 ctx: &AgentContext,
1888 tool_name: &str,
1889 allowed: bool,
1890 reason: &str,
1891 causal_token: &str,
1892 ) {
1893 let mut permission = State::new();
1894 permission.insert("tool".to_string(), Value::String(tool_name.to_string()));
1895 permission.insert("allowed".to_string(), Value::Bool(allowed));
1896 permission.insert("reason".to_string(), Value::String(reason.to_string()));
1897 self.store.append_event(
1898 &ctx.run_id,
1899 Some(&ctx.session_id),
1900 Some(&ctx.step_id),
1901 "tool_permission_decided",
1902 permission,
1903 Some(&ctx.agent_role),
1904 Some(ctx.state_version),
1905 Some(causal_token),
1906 );
1907 }
1908
1909 pub fn create_artifact(
1910 &mut self,
1911 ctx: &AgentContext,
1912 name: &str,
1913 content: State,
1914 metadata: State,
1915 ) -> Result<String> {
1916 let artifact =
1917 self.store
1918 .create_artifact(&ctx.run_id, Some(&ctx.step_id), name, content, metadata);
1919 let mut payload = State::new();
1920 payload.insert(
1921 "artifact_id".to_string(),
1922 Value::String(artifact.artifact_id.clone()),
1923 );
1924 payload.insert("name".to_string(), Value::String(name.to_string()));
1925 self.store.append_event(
1926 &ctx.run_id,
1927 Some(&ctx.session_id),
1928 Some(&ctx.step_id),
1929 "artifact_created",
1930 payload,
1931 Some(&ctx.agent_role),
1932 Some(ctx.state_version),
1933 None,
1934 );
1935 Ok(artifact.artifact_id)
1936 }
1937
1938 pub fn create_media_artifact(
1939 &mut self,
1940 ctx: &AgentContext,
1941 name: &str,
1942 kind: &str,
1943 options: MediaArtifactOptions,
1944 ) -> Result<String> {
1945 if !is_media_kind(kind) {
1946 return Err(RuntimeError(format!("unsupported media kind: {kind}")));
1947 }
1948 let mut media_metadata = options.media_metadata.clone();
1949 media_metadata.insert(
1950 "schema_version".to_string(),
1951 Value::String(MEDIA_SCHEMA_VERSION.to_string()),
1952 );
1953 media_metadata.insert("kind".to_string(), Value::String(kind.to_string()));
1954 let mut content = State::new();
1955 content.insert(
1956 "schema_version".to_string(),
1957 Value::String(MEDIA_SCHEMA_VERSION.to_string()),
1958 );
1959 content.insert("kind".to_string(), Value::String(kind.to_string()));
1960 if let Some(uri) = &options.uri {
1961 content.insert("uri".to_string(), Value::String(uri.clone()));
1962 }
1963 if let Some(content_ref) = &options.content_ref {
1964 content.insert(
1965 "content_ref".to_string(),
1966 Value::String(content_ref.clone()),
1967 );
1968 }
1969 content.insert(
1970 "metadata".to_string(),
1971 Value::Object(media_metadata.clone()),
1972 );
1973 if !options.lineage.is_empty() {
1974 content.insert(
1975 "lineage".to_string(),
1976 Value::Object(options.lineage.clone()),
1977 );
1978 }
1979 if !options.derived_outputs.is_empty() {
1980 content.insert(
1981 "derived_outputs".to_string(),
1982 Value::Object(options.derived_outputs.clone()),
1983 );
1984 }
1985 let mut media = State::new();
1986 media.insert(
1987 "schema_version".to_string(),
1988 Value::String(MEDIA_SCHEMA_VERSION.to_string()),
1989 );
1990 media.insert("kind".to_string(), Value::String(kind.to_string()));
1991 if let Some(uri) = options.uri {
1992 media.insert("uri".to_string(), Value::String(uri));
1993 }
1994 if let Some(content_ref) = options.content_ref {
1995 media.insert("content_ref".to_string(), Value::String(content_ref));
1996 }
1997 media.insert("metadata".to_string(), Value::Object(media_metadata));
1998 media.insert("lineage".to_string(), Value::Object(options.lineage));
1999 let mut artifact_metadata = options.metadata;
2000 artifact_metadata.insert("agentledger_media".to_string(), Value::Object(media));
2001 self.create_artifact(ctx, name, content, artifact_metadata)
2002 }
2003
2004 pub fn create_stream_checkpoint(
2005 &mut self,
2006 ctx: &AgentContext,
2007 name: &str,
2008 options: StreamCheckpointOptions,
2009 ) -> Result<String> {
2010 if options.stream_id.is_empty() || options.consumer_id.is_empty() {
2011 return Err(RuntimeError(
2012 "stream_id and consumer_id are required".to_string(),
2013 ));
2014 }
2015 let chunk = options.chunk.map(stream_chunk_to_state);
2016 let mut content = State::new();
2017 content.insert(
2018 "schema_version".to_string(),
2019 Value::String(STREAM_SCHEMA_VERSION.to_string()),
2020 );
2021 content.insert(
2022 "stream_id".to_string(),
2023 Value::String(options.stream_id.clone()),
2024 );
2025 content.insert(
2026 "consumer_id".to_string(),
2027 Value::String(options.consumer_id.clone()),
2028 );
2029 content.insert("offset".to_string(), options.offset.clone());
2030 if let Some(watermark) = &options.watermark {
2031 content.insert("watermark".to_string(), watermark.clone());
2032 }
2033 if let Some(chunk_state) = &chunk {
2034 content.insert("chunk".to_string(), Value::Object(chunk_state.clone()));
2035 }
2036 if let Some(partial_ref) = &options.partial_result_ref {
2037 content.insert(
2038 "partial_result_ref".to_string(),
2039 Value::String(partial_ref.clone()),
2040 );
2041 }
2042 if !options.backpressure.is_empty() {
2043 content.insert(
2044 "backpressure".to_string(),
2045 Value::Object(options.backpressure.clone()),
2046 );
2047 }
2048 if !options.metadata.is_empty() {
2049 content.insert("metadata".to_string(), Value::Object(options.metadata));
2050 }
2051 let mut stream = State::new();
2052 stream.insert(
2053 "schema_version".to_string(),
2054 Value::String(STREAM_SCHEMA_VERSION.to_string()),
2055 );
2056 stream.insert("stream_id".to_string(), Value::String(options.stream_id));
2057 stream.insert(
2058 "consumer_id".to_string(),
2059 Value::String(options.consumer_id),
2060 );
2061 stream.insert("offset".to_string(), options.offset);
2062 if let Some(watermark) = options.watermark {
2063 stream.insert("watermark".to_string(), watermark);
2064 }
2065 if let Some(chunk_state) = chunk {
2066 stream.insert("chunk".to_string(), Value::Object(chunk_state));
2067 }
2068 if let Some(partial_ref) = options.partial_result_ref {
2069 stream.insert("partial_result_ref".to_string(), Value::String(partial_ref));
2070 }
2071 if !options.backpressure.is_empty() {
2072 stream.insert(
2073 "backpressure".to_string(),
2074 Value::Object(options.backpressure),
2075 );
2076 }
2077 let mut artifact_metadata = State::new();
2078 artifact_metadata.insert("agentledger_stream".to_string(), Value::Object(stream));
2079 self.create_artifact(ctx, name, content, artifact_metadata)
2080 }
2081
2082 pub fn record_model_call(
2083 &mut self,
2084 ctx: &AgentContext,
2085 model: &str,
2086 input_tokens: f64,
2087 output_tokens: f64,
2088 total_usd: f64,
2089 ) -> Result<()> {
2090 let tokens = input_tokens + output_tokens;
2091 if let Some(max) = self.budget.max_model_tokens {
2092 let used = self.store.cost_summary(&ctx.run_id).model_tokens;
2093 if used + tokens > max {
2094 let message = format!("model token budget exceeded: {used}+{tokens}/{max}");
2095 let mut payload = State::new();
2096 payload.insert("category".to_string(), Value::String("model".to_string()));
2097 payload.insert("model".to_string(), Value::String(model.to_string()));
2098 payload.insert("error".to_string(), Value::String(message.clone()));
2099 self.store.append_event(
2100 &ctx.run_id,
2101 Some(&ctx.session_id),
2102 Some(&ctx.step_id),
2103 "budget_check_failed",
2104 payload,
2105 Some(&ctx.agent_role),
2106 Some(ctx.state_version),
2107 None,
2108 );
2109 return Err(RuntimeError(message));
2110 }
2111 }
2112 let mut payload = State::new();
2113 payload.insert("model".to_string(), Value::String(model.to_string()));
2114 payload.insert("input_tokens".to_string(), Value::Number(input_tokens));
2115 payload.insert("output_tokens".to_string(), Value::Number(output_tokens));
2116 payload.insert("total_tokens".to_string(), Value::Number(tokens));
2117 payload.insert("total_usd".to_string(), Value::Number(total_usd));
2118 self.store.append_event(
2119 &ctx.run_id,
2120 Some(&ctx.session_id),
2121 Some(&ctx.step_id),
2122 "model_call_completed",
2123 payload,
2124 Some(&ctx.agent_role),
2125 Some(ctx.state_version),
2126 None,
2127 );
2128 if tokens > 0.0 {
2129 self.store.record_cost(
2130 &ctx.run_id,
2131 &ctx.session_id,
2132 &ctx.step_id,
2133 "model",
2134 model,
2135 tokens,
2136 "token",
2137 State::new(),
2138 );
2139 }
2140 if total_usd > 0.0 {
2141 self.store.record_cost(
2142 &ctx.run_id,
2143 &ctx.session_id,
2144 &ctx.step_id,
2145 "model",
2146 model,
2147 total_usd,
2148 "usd",
2149 State::new(),
2150 );
2151 }
2152 Ok(())
2153 }
2154}
2155
2156impl Default for Runtime {
2157 fn default() -> Self {
2158 Self::new()
2159 }
2160}
2161
2162#[derive(Clone, Debug)]
2163pub struct AgentContext {
2164 pub run_id: String,
2165 pub session_id: String,
2166 pub step_id: String,
2167 pub agent_role: String,
2168 pub lease_token: String,
2169 pub attempt: u64,
2170 pub state_version: u64,
2171 pub pending_patch: State,
2172}
2173
2174impl AgentContext {
2175 pub fn write_state(&mut self, key: &str, value: Value) {
2176 self.pending_patch.insert(key.to_string(), value);
2177 }
2178}
2179
2180#[derive(Clone, Debug)]
2181pub struct EvidenceBundle {
2182 pub schema_version: String,
2183 pub bundle_hash: String,
2184 pub run: Run,
2185 pub steps: Vec<Step>,
2186 pub events: Vec<Event>,
2187 pub tool_ledger: Vec<ToolLedgerEntry>,
2188 pub approvals: Vec<ApprovalRequest>,
2189 pub artifacts: Vec<Artifact>,
2190 pub media_artifacts: Vec<State>,
2191 pub stream_checkpoints: Vec<State>,
2192 pub cost_records: Vec<CostRecord>,
2193 pub cost_summary: CostSummary,
2194 pub summary: State,
2195 pub final_state: State,
2196}
2197
2198#[derive(Clone, Debug, Default)]
2199pub struct WorkerRunSummary {
2200 pub worker_id: String,
2201 pub run_id: Option<String>,
2202 pub iterations: u64,
2203 pub attempts: u64,
2204 pub succeeded_attempts: u64,
2205 pub recovered_leases: u64,
2206 pub final_status: Option<String>,
2207 pub stopped_reason: String,
2208}
2209
2210pub struct LocalWorker {
2211 pub worker_id: String,
2212 pub agent_role: String,
2213 pub lease_seconds: f64,
2214 pub recover_expired: bool,
2215}
2216
2217impl LocalWorker {
2218 pub fn new(worker_id: &str, agent_role: &str) -> Self {
2219 Self {
2220 worker_id: worker_id.to_string(),
2221 agent_role: agent_role.to_string(),
2222 lease_seconds: 60.0,
2223 recover_expired: true,
2224 }
2225 }
2226
2227 pub fn run_until_idle(
2228 &self,
2229 runtime: &mut Runtime,
2230 run_id: &str,
2231 max_iterations: u64,
2232 agent: AgentFunc,
2233 ) -> Result<WorkerRunSummary> {
2234 let mut summary = WorkerRunSummary {
2235 worker_id: self.worker_id.clone(),
2236 run_id: Some(run_id.to_string()),
2237 stopped_reason: "max_iterations".to_string(),
2238 ..WorkerRunSummary::default()
2239 };
2240 for index in 1..=max_iterations {
2241 summary.iterations = index;
2242 if self.recover_expired {
2243 summary.recovered_leases += runtime.store.recover_expired_leases() as u64;
2244 }
2245 let status = runtime.store.run(run_id)?.status;
2246 if is_terminal_status(&status) {
2247 summary.final_status = Some(status);
2248 summary.stopped_reason = "terminal_status".to_string();
2249 break;
2250 }
2251 let ok = runtime.run_once(
2252 run_id,
2253 &self.worker_id,
2254 &self.agent_role,
2255 self.lease_seconds,
2256 agent,
2257 )?;
2258 if !ok {
2259 summary.stopped_reason = "idle".to_string();
2260 break;
2261 }
2262 summary.attempts += 1;
2263 summary.succeeded_attempts += 1;
2264 }
2265 let status = runtime.store.run(run_id)?.status;
2266 summary.final_status = Some(status.clone());
2267 if is_terminal_status(&status) {
2268 summary.stopped_reason = "terminal_status".to_string();
2269 }
2270 Ok(summary)
2271 }
2272}
2273
2274#[derive(Clone, Debug, Default)]
2275pub struct WorkerServiceSummary {
2276 pub worker_id: String,
2277 pub run_id: Option<String>,
2278 pub loops: u64,
2279 pub attempts: u64,
2280 pub succeeded_attempts: u64,
2281 pub recovered_leases: u64,
2282 pub idle_polls: u64,
2283 pub stopped_reason: String,
2284 pub final_status: Option<String>,
2285 pub stop_requested: bool,
2286}
2287
2288pub struct WorkerService {
2289 pub worker: LocalWorker,
2290 pub stop_requested: bool,
2291 pub stop_reason: String,
2292}
2293
2294impl WorkerService {
2295 pub fn new(worker: LocalWorker) -> Self {
2296 Self {
2297 worker,
2298 stop_requested: false,
2299 stop_reason: "stop_requested".to_string(),
2300 }
2301 }
2302
2303 pub fn request_stop(&mut self, reason: &str) {
2304 self.stop_requested = true;
2305 self.stop_reason = reason.to_string();
2306 }
2307
2308 pub fn serve(
2309 &mut self,
2310 runtime: &mut Runtime,
2311 run_id: Option<&str>,
2312 max_loops: u64,
2313 max_idle_polls: Option<u64>,
2314 agent: AgentFunc,
2315 ) -> Result<WorkerServiceSummary> {
2316 let mut summary = WorkerServiceSummary {
2317 worker_id: self.worker.worker_id.clone(),
2318 run_id: run_id.map(str::to_string),
2319 stopped_reason: "max_loops".to_string(),
2320 ..WorkerServiceSummary::default()
2321 };
2322 while summary.loops < max_loops {
2323 if self.stop_requested {
2324 summary.stopped_reason = self.stop_reason.clone();
2325 summary.stop_requested = true;
2326 break;
2327 }
2328 summary.loops += 1;
2329 let Some(run_id) = run_id else {
2330 summary.idle_polls += 1;
2331 if max_idle_polls.is_some_and(|limit| summary.idle_polls >= limit) {
2332 summary.stopped_reason = "idle".to_string();
2333 break;
2334 }
2335 continue;
2336 };
2337 let run_summary = self.worker.run_until_idle(runtime, run_id, 1, agent)?;
2338 summary.attempts += run_summary.attempts;
2339 summary.succeeded_attempts += run_summary.succeeded_attempts;
2340 summary.recovered_leases += run_summary.recovered_leases;
2341 summary.final_status = run_summary.final_status.clone();
2342 if summary
2343 .final_status
2344 .as_deref()
2345 .is_some_and(is_terminal_status)
2346 {
2347 summary.stopped_reason = "terminal_status".to_string();
2348 break;
2349 }
2350 if run_summary.attempts == 0 {
2351 summary.idle_polls += 1;
2352 if max_idle_polls.is_some_and(|limit| summary.idle_polls >= limit) {
2353 summary.stopped_reason = "idle".to_string();
2354 break;
2355 }
2356 } else {
2357 summary.idle_polls = 0;
2358 }
2359 }
2360 Ok(summary)
2361 }
2362}
2363
2364fn is_terminal_status(status: &str) -> bool {
2365 matches!(status, "completed" | "failed" | "cancelled")
2366}
2367
2368pub fn export_evidence(store: &MemoryStore, run_id: &str) -> Result<EvidenceBundle> {
2369 let run = store.run(run_id)?;
2370 let steps = store.steps(run_id);
2371 let events = store.events(run_id);
2372 let tool_ledger = store.ledger(run_id);
2373 let approvals = store.approval_requests(run_id);
2374 let artifacts = store.artifacts(run_id);
2375 let media_artifacts = media_artifacts_from(&artifacts);
2376 let stream_checkpoints = stream_checkpoints_from(&artifacts);
2377 let cost_records = store.cost_records(run_id);
2378 let cost_summary = store.cost_summary(run_id);
2379 let summary = evidence_summary(
2380 &steps,
2381 &events,
2382 &tool_ledger,
2383 &approvals,
2384 &artifacts,
2385 &media_artifacts,
2386 &stream_checkpoints,
2387 &cost_records,
2388 &cost_summary,
2389 );
2390 let final_state = store.final_state(run_id)?;
2391 let basis = format!(
2392 "{}:{}:{}:{}:{}",
2393 run_id,
2394 steps.len(),
2395 events.len(),
2396 tool_ledger.len(),
2397 cost_records.len()
2398 );
2399 Ok(EvidenceBundle {
2400 schema_version: "agentledger.evidence.v1".to_string(),
2401 bundle_hash: stable_hash(&basis),
2402 run,
2403 steps,
2404 events,
2405 tool_ledger,
2406 approvals,
2407 artifacts,
2408 media_artifacts,
2409 stream_checkpoints,
2410 cost_records,
2411 cost_summary,
2412 summary,
2413 final_state,
2414 })
2415}
2416
2417fn evidence_summary(
2418 steps: &[Step],
2419 events: &[Event],
2420 ledger: &[ToolLedgerEntry],
2421 approvals: &[ApprovalRequest],
2422 artifacts: &[Artifact],
2423 media_artifacts: &[State],
2424 stream_checkpoints: &[State],
2425 cost_records: &[CostRecord],
2426 cost_summary: &CostSummary,
2427) -> State {
2428 let mut summary = State::new();
2429 summary.insert("event_count".into(), Value::Number(events.len() as f64));
2430 summary.insert("step_count".into(), Value::Number(steps.len() as f64));
2431 summary.insert(
2432 "tool_ledger_count".into(),
2433 Value::Number(ledger.len() as f64),
2434 );
2435 summary.insert(
2436 "approval_count".into(),
2437 Value::Number(approvals.len() as f64),
2438 );
2439 summary.insert(
2440 "artifact_count".into(),
2441 Value::Number(artifacts.len() as f64),
2442 );
2443 summary.insert(
2444 "media_artifact_count".into(),
2445 Value::Number(media_artifacts.len() as f64),
2446 );
2447 summary.insert(
2448 "stream_checkpoint_count".into(),
2449 Value::Number(stream_checkpoints.len() as f64),
2450 );
2451 summary.insert(
2452 "cost_record_count".into(),
2453 Value::Number(cost_records.len() as f64),
2454 );
2455 summary.insert(
2456 "has_failed_steps".into(),
2457 Value::Bool(steps.iter().any(|step| step.status == "failed")),
2458 );
2459 summary.insert(
2460 "has_pending_verification".into(),
2461 Value::Bool(
2462 ledger
2463 .iter()
2464 .any(|row| row.status == "PENDING_VERIFICATION"),
2465 ),
2466 );
2467 summary.insert(
2468 "has_pending_approvals".into(),
2469 Value::Bool(approvals.iter().any(|row| row.status == "PENDING")),
2470 );
2471 let mut cost = State::new();
2472 cost.insert("tool_calls".into(), Value::Number(cost_summary.tool_calls));
2473 cost.insert(
2474 "model_tokens".into(),
2475 Value::Number(cost_summary.model_tokens),
2476 );
2477 cost.insert("total_usd".into(), Value::Number(cost_summary.total_usd));
2478 summary.insert("cost_summary".into(), Value::Object(cost));
2479 summary
2480}
2481
2482#[derive(Clone, Debug)]
2483pub struct ReplaySummary {
2484 pub run_id: String,
2485 pub event_count: usize,
2486 pub tool_call_count: usize,
2487 pub final_state: State,
2488 pub event_hash: String,
2489 pub replay_safe: bool,
2490 pub artifact_count: usize,
2491 pub media_artifact_count: usize,
2492 pub stream_checkpoint_count: usize,
2493}
2494
2495pub fn replay(store: &MemoryStore, run_id: &str) -> Result<ReplaySummary> {
2496 let events = store.events(run_id);
2497 let tool_call_count = events
2498 .iter()
2499 .filter(|event| event.event_type.starts_with("tool_call_"))
2500 .count();
2501 let digest = events
2502 .iter()
2503 .map(|event| format!("{}:{}:{}", event.seq, event.event_type, event.payload_hash))
2504 .collect::<Vec<_>>()
2505 .join("|");
2506 let artifacts = store.artifacts(run_id);
2507 Ok(ReplaySummary {
2508 run_id: run_id.to_string(),
2509 event_count: events.len(),
2510 tool_call_count,
2511 final_state: store.final_state(run_id)?,
2512 event_hash: stable_hash(&digest),
2513 replay_safe: true,
2514 artifact_count: artifacts.len(),
2515 media_artifact_count: media_artifacts_from(&artifacts).len(),
2516 stream_checkpoint_count: stream_checkpoints_from(&artifacts).len(),
2517 })
2518}
2519
2520#[derive(Clone, Debug)]
2521pub struct CostAttributionReport {
2522 pub run_id: String,
2523 pub total: CostSummary,
2524 pub by_agent: HashMap<String, CostSummary>,
2525 pub by_step: HashMap<String, CostSummary>,
2526 pub by_name: HashMap<String, CostSummary>,
2527}
2528
2529pub fn cost_attribution(store: &MemoryStore, run_id: &str) -> CostAttributionReport {
2530 let mut step_roles = HashMap::new();
2531 for event in store.events(run_id) {
2532 if let (Some(step_id), Some(agent_role)) = (event.step_id, event.agent_role) {
2533 step_roles.insert(step_id, agent_role);
2534 }
2535 }
2536 let mut report = CostAttributionReport {
2537 run_id: run_id.to_string(),
2538 total: CostSummary::default(),
2539 by_agent: HashMap::new(),
2540 by_step: HashMap::new(),
2541 by_name: HashMap::new(),
2542 };
2543 for record in store.cost_records(run_id) {
2544 add_cost(&mut report.total, &record);
2545 let agent = step_roles
2546 .get(&record.step_id)
2547 .cloned()
2548 .unwrap_or_else(|| "<unknown>".to_string());
2549 add_cost(report.by_agent.entry(agent).or_default(), &record);
2550 add_cost(
2551 report.by_step.entry(record.step_id.clone()).or_default(),
2552 &record,
2553 );
2554 add_cost(
2555 report.by_name.entry(record.name.clone()).or_default(),
2556 &record,
2557 );
2558 }
2559 report
2560}
2561
2562#[derive(Clone, Debug)]
2563pub struct FailureAttributionReport {
2564 pub run_id: String,
2565 pub run_status: String,
2566 pub failed_steps: Vec<Step>,
2567 pub pending_verification: Vec<ToolLedgerEntry>,
2568 pub pending_approvals: Vec<ApprovalRequest>,
2569 pub failure_events: Vec<Event>,
2570 pub failure_envelopes: Vec<State>,
2571 pub failure_lifecycle: State,
2572 pub failure_causal_graph: State,
2573 pub failure_replay_plan: State,
2574 pub failure_alerts: State,
2575 pub failure_export: State,
2576 pub summary: State,
2577}
2578
2579pub fn failure_attribution(store: &MemoryStore, run_id: &str) -> Result<FailureAttributionReport> {
2580 let run = store.run(run_id)?;
2581 let failed_steps: Vec<Step> = store
2582 .steps(run_id)
2583 .into_iter()
2584 .filter(|step| step.status == "failed")
2585 .collect();
2586 let pending_verification: Vec<ToolLedgerEntry> = store
2587 .ledger(run_id)
2588 .into_iter()
2589 .filter(|entry| entry.status == "PENDING_VERIFICATION")
2590 .collect();
2591 let pending_approvals: Vec<ApprovalRequest> = store
2592 .approval_requests(run_id)
2593 .into_iter()
2594 .filter(|entry| entry.status == "PENDING")
2595 .collect();
2596 let failure_events: Vec<Event> = store
2597 .events(run_id)
2598 .into_iter()
2599 .filter(|event| is_failure_event(&event.event_type))
2600 .collect();
2601 let steps = store.steps(run_id);
2602 let ledger = store.ledger(run_id);
2603 let approvals = store.approval_requests(run_id);
2604 let events = store.events(run_id);
2605 let costs = store.cost_records(run_id);
2606 let failure_envelopes = failure_envelopes(run_id, &run.status, &steps, &ledger, &approvals, &failure_events);
2607 let failure_lifecycle = failure_lifecycle(run_id, &run.status, &failure_envelopes);
2608 let failure_causal_graph = failure_causal_graph(run_id, &run.status, &failure_envelopes, &steps, &ledger, &approvals, &events, &costs);
2609 let failure_replay_plan = failure_replay_plan(run_id, &failure_envelopes, &ledger, &events);
2610 let failure_alerts = failure_alerts(run_id, &failure_envelopes, &failure_replay_plan);
2611 let mut summary = State::new();
2612 summary.insert(
2613 "failed_step_count".to_string(),
2614 Value::Number(failed_steps.len() as f64),
2615 );
2616 summary.insert(
2617 "pending_verification_count".to_string(),
2618 Value::Number(pending_verification.len() as f64),
2619 );
2620 summary.insert(
2621 "pending_approval_count".to_string(),
2622 Value::Number(pending_approvals.len() as f64),
2623 );
2624 summary.insert(
2625 "failure_event_count".to_string(),
2626 Value::Number(failure_events.len() as f64),
2627 );
2628 summary.insert(
2629 "failure_envelope_count".to_string(),
2630 Value::Number(failure_envelopes.len() as f64),
2631 );
2632 summary.insert(
2633 "failure_lifecycle_event_count".to_string(),
2634 Value::Number(state_array_len(&failure_lifecycle, "events") as f64),
2635 );
2636 summary.insert(
2637 "failure_alert_count".to_string(),
2638 failure_alerts
2639 .get("alert_count")
2640 .cloned()
2641 .unwrap_or(Value::Number(0.0)),
2642 );
2643 summary.insert(
2644 "unsafe_replay_side_effect_count".to_string(),
2645 failure_replay_plan
2646 .get("unsafe_side_effect_count")
2647 .cloned()
2648 .unwrap_or(Value::Number(0.0)),
2649 );
2650 summary.insert(
2651 "terminal_failure_count".to_string(),
2652 Value::Number(count_state_field(&failure_envelopes, "status", "terminal") as f64),
2653 );
2654 summary.insert(
2655 "recoverable_failure_count".to_string(),
2656 Value::Number(count_recoverable_failures(&failure_envelopes) as f64),
2657 );
2658 let failure_export = failure_export(
2659 run_id,
2660 &run.status,
2661 &summary,
2662 &failure_envelopes,
2663 &failure_lifecycle,
2664 &failure_causal_graph,
2665 &failure_replay_plan,
2666 &failure_alerts,
2667 );
2668 Ok(FailureAttributionReport {
2669 run_id: run_id.to_string(),
2670 run_status: run.status,
2671 failed_steps,
2672 pending_verification,
2673 pending_approvals,
2674 failure_events,
2675 failure_envelopes,
2676 failure_lifecycle,
2677 failure_causal_graph,
2678 failure_replay_plan,
2679 failure_alerts,
2680 failure_export,
2681 summary,
2682 })
2683}
2684
2685fn failure_envelopes(
2686 run_id: &str,
2687 run_status: &str,
2688 steps: &[Step],
2689 ledger: &[ToolLedgerEntry],
2690 approvals: &[ApprovalRequest],
2691 events: &[Event],
2692) -> Vec<State> {
2693 let mut rows = Vec::new();
2694 for step in steps {
2695 if step.status == "failed" || step.status == "retry_scheduled" || step.status == "waiting_human" {
2696 let (status, severity, recoverability, retryability) = if step.status == "retry_scheduled" {
2697 ("recovery_scheduled", "warn", "auto_retry", "retryable")
2698 } else if step.status == "waiting_human" {
2699 ("waiting_human", "warn", "human_required", "unknown")
2700 } else {
2701 ("terminal", "risk", "terminal", "not_retryable")
2702 };
2703 rows.push(failure_envelope(
2704 run_id,
2705 "step",
2706 &step.step_id,
2707 &failure_category(&format!(
2708 "{} {}",
2709 step.last_error_type.clone().unwrap_or_default(),
2710 step.last_error.clone().unwrap_or_default()
2711 ), "agent"),
2712 status,
2713 severity,
2714 recoverability,
2715 retryability,
2716 "agent",
2717 &first_text(&[
2718 step.last_error.as_deref(),
2719 step.last_error_type.as_deref(),
2720 Some("step failure"),
2721 ]),
2722 state(&[
2723 ("step_id", step.step_id.clone().into()),
2724 ("occurred_at", Value::Number(step.updated_at)),
2725 ]),
2726 vec![ref_state("step", &step.step_id)],
2727 ));
2728 }
2729 }
2730 for entry in ledger {
2731 if entry.status == "PENDING_VERIFICATION" || entry.status == "FAILED" || entry.status == "ERROR" {
2732 let terminal = entry.status == "FAILED" || entry.status == "ERROR";
2733 rows.push(failure_envelope(
2734 run_id,
2735 "tool_ledger",
2736 &first_text(&[Some(&entry.ledger_id), Some(&entry.tool_name), Some(&entry.step_id)]),
2737 "tool",
2738 if terminal { "terminal" } else { "unknown_side_effect" },
2739 if terminal { "risk" } else { "warn" },
2740 if terminal { "terminal" } else { "manual_verification" },
2741 if terminal { "not_retryable" } else { "unknown" },
2742 "tool",
2743 &first_text(&[entry.error_type.as_deref(), Some("tool side effect requires verification")]),
2744 state(&[
2745 ("step_id", entry.step_id.clone().into()),
2746 ("tool_name", entry.tool_name.clone().into()),
2747 ("occurred_at", Value::Number(entry.updated_at)),
2748 ]),
2749 vec![ref_state("step", &entry.step_id), ref_state("tool", &entry.tool_name)],
2750 ));
2751 }
2752 }
2753 for approval in approvals {
2754 if approval.status == "PENDING" || approval.status == "DENIED" {
2755 let denied = approval.status == "DENIED";
2756 rows.push(failure_envelope(
2757 run_id,
2758 "approval",
2759 &first_text(&[Some(&approval.approval_id), Some(&approval.tool_name), Some(&approval.step_id)]),
2760 if denied { "policy" } else { "approval" },
2761 if denied { "blocked" } else { "waiting_human" },
2762 if denied { "risk" } else { "warn" },
2763 if denied { "terminal" } else { "human_required" },
2764 if denied { "not_retryable" } else { "unknown" },
2765 "policy",
2766 &first_text(&[
2767 approval.decision_reason.as_deref(),
2768 Some(&approval.reason),
2769 Some(if denied { "approval denied" } else { "approval pending" }),
2770 ]),
2771 state(&[
2772 ("step_id", approval.step_id.clone().into()),
2773 ("tool_name", approval.tool_name.clone().into()),
2774 ("approval_id", approval.approval_id.clone().into()),
2775 ("occurred_at", Value::Number(approval.updated_at)),
2776 ]),
2777 vec![
2778 ref_state("step", &approval.step_id),
2779 ref_state("tool", &approval.tool_name),
2780 ref_state("approval", &approval.approval_id),
2781 ],
2782 ));
2783 }
2784 }
2785 for event in events {
2786 let category = event_failure_category(event);
2787 let status = event_failure_status(&event.event_type, run_status);
2788 let message = first_text(&[
2789 state_string(&event.payload, "error").as_deref(),
2790 state_string(&event.payload, "reason").as_deref(),
2791 state_string(&event.payload, "error_type").as_deref(),
2792 Some(&event.event_type),
2793 ]);
2794 rows.push(failure_envelope(
2795 run_id,
2796 "event",
2797 &event.seq.to_string(),
2798 &category,
2799 &status,
2800 if status == "terminal" || status == "blocked" || status == "failed" { "risk" } else { "warn" },
2801 &event_recoverability(&event.event_type, run_status),
2802 &event_retryability(&event.event_type),
2803 &owner_for_failure_category(&category),
2804 &message,
2805 state(&[
2806 ("step_id", event.step_id.clone().unwrap_or_default().into()),
2807 ("event_seq", Value::Number(event.seq as f64)),
2808 ("event_type", event.event_type.clone().into()),
2809 ("occurred_at", Value::Number(event.timestamp)),
2810 ]),
2811 vec![
2812 ref_state("event", &event.seq.to_string()),
2813 ref_state("step", event.step_id.as_deref().unwrap_or("")),
2814 ],
2815 ));
2816 }
2817 dedupe_states(rows, "failure_id")
2818}
2819
2820fn failure_envelope(
2821 run_id: &str,
2822 source_kind: &str,
2823 source_id: &str,
2824 category: &str,
2825 status: &str,
2826 severity: &str,
2827 recoverability: &str,
2828 retryability: &str,
2829 owner: &str,
2830 message: &str,
2831 extra: State,
2832 refs: Vec<State>,
2833) -> State {
2834 let mut row = state(&[
2835 ("schema_version", "agentledger.failure.envelope.v1".into()),
2836 ("failure_id", format!("failure-{}", slug(&format!("{run_id}-{source_kind}-{source_id}"))).into()),
2837 ("run_id", run_id.into()),
2838 ("source_kind", source_kind.into()),
2839 ("source_id", source_id.into()),
2840 ("category", category.into()),
2841 ("status", status.into()),
2842 ("severity", severity.into()),
2843 ("recoverability", recoverability.into()),
2844 ("retryability", retryability.into()),
2845 ("owner", owner.into()),
2846 ("message", message.into()),
2847 ("causal_refs", Value::Array(refs.iter().cloned().map(Value::Object).collect())),
2848 ("evidence_refs", Value::Array(refs.into_iter().map(Value::Object).collect())),
2849 ]);
2850 for (key, value) in extra {
2851 if value != Value::String(String::new()) && value != Value::Null {
2852 row.insert(key, value);
2853 }
2854 }
2855 row
2856}
2857
2858fn failure_lifecycle(run_id: &str, run_status: &str, envelopes: &[State]) -> State {
2859 let mut events = Vec::new();
2860 for env in envelopes {
2861 events.push(lifecycle_row(run_id, env, "failure_detected", state_value_string(env, "message"), state_value_string(env, "severity")));
2862 events.push(lifecycle_row(run_id, env, "failure_classified", state_value_string(env, "category"), state_value_string(env, "severity")));
2863 let status = state_value_string(env, "status");
2864 let recoverability = state_value_string(env, "recoverability");
2865 if ["recovery_scheduled", "waiting_human", "unknown_side_effect"].contains(&status.as_str()) || ["auto_retry", "human_required", "manual_verification"].contains(&recoverability.as_str()) {
2866 events.push(lifecycle_row(run_id, env, "failure_recovery_scheduled", "recovery scheduled".to_string(), "warn".to_string()));
2867 }
2868 if ["terminal", "blocked"].contains(&status.as_str()) || recoverability == "terminal" {
2869 events.push(lifecycle_row(run_id, env, "failure_terminal", state_value_string(env, "message"), "risk".to_string()));
2870 }
2871 }
2872 state(&[
2873 ("schema_version", "agentledger.failure.lifecycle.v1".into()),
2874 ("run_id", run_id.into()),
2875 ("run_status", run_status.into()),
2876 ("events", Value::Array(events.iter().cloned().map(Value::Object).collect())),
2877 ("terminal", Value::Bool(events.iter().any(|row| state_value_string(row, "stage") == "failure_terminal"))),
2878 ("recoverable", Value::Bool(events.iter().any(|row| state_value_string(row, "stage") == "failure_recovery_scheduled"))),
2879 ])
2880}
2881
2882fn lifecycle_row(run_id: &str, env: &State, stage: &str, message: String, severity: String) -> State {
2883 state(&[
2884 ("schema_version", "agentledger.failure.lifecycle.v1".into()),
2885 ("stage", stage.into()),
2886 ("run_id", run_id.into()),
2887 ("failure_id", state_value_string(env, "failure_id").into()),
2888 ("category", state_value_string(env, "category").into()),
2889 ("recoverability", state_value_string(env, "recoverability").into()),
2890 ("retryability", state_value_string(env, "retryability").into()),
2891 ("owner", state_value_string(env, "owner").into()),
2892 ("message", message.into()),
2893 ("severity", severity.into()),
2894 ("causal_refs", env.get("causal_refs").cloned().unwrap_or(Value::Array(vec![]))),
2895 ])
2896}
2897
2898fn failure_causal_graph(
2899 run_id: &str,
2900 run_status: &str,
2901 envelopes: &[State],
2902 steps: &[Step],
2903 ledger: &[ToolLedgerEntry],
2904 approvals: &[ApprovalRequest],
2905 events: &[Event],
2906 costs: &[CostRecord],
2907) -> State {
2908 let mut nodes = vec![state(&[("id", format!("run:{}", slug(run_id)).into()), ("kind", "run".into()), ("status", run_status.into())])];
2909 let mut edges = Vec::new();
2910 for step in steps {
2911 nodes.push(state(&[("id", format!("step:{}", slug(&step.step_id)).into()), ("kind", "step".into()), ("status", step.status.clone().into())]));
2912 edges.push(state(&[("source", format!("run:{}", slug(run_id)).into()), ("target", format!("step:{}", slug(&step.step_id)).into()), ("kind", "contains_step".into())]));
2913 }
2914 for event in events {
2915 nodes.push(state(&[("id", format!("event:{}", event.seq).into()), ("kind", "event".into()), ("event_type", event.event_type.clone().into())]));
2916 edges.push(state(&[("source", format!("run:{}", slug(run_id)).into()), ("target", format!("event:{}", event.seq).into()), ("kind", "emitted_event".into())]));
2917 }
2918 for entry in ledger {
2919 nodes.push(state(&[("id", format!("tool:{}", slug(&entry.tool_name)).into()), ("kind", "tool".into()), ("status", entry.status.clone().into())]));
2920 }
2921 for approval in approvals {
2922 nodes.push(state(&[("id", format!("approval:{}", slug(&approval.approval_id)).into()), ("kind", "approval".into()), ("status", approval.status.clone().into())]));
2923 }
2924 for cost in costs {
2925 nodes.push(state(&[("id", format!("cost:{}", slug(&cost.cost_id)).into()), ("kind", "cost".into()), ("category", cost.category.clone().into()), ("amount", Value::Number(cost.amount)), ("unit", cost.unit.clone().into())]));
2926 }
2927 for env in envelopes {
2928 let id = format!("failure:{}", slug(&state_value_string(env, "failure_id")));
2929 nodes.push(state(&[("id", id.clone().into()), ("kind", "failure".into()), ("category", state_value_string(env, "category").into()), ("status", state_value_string(env, "status").into()), ("owner", state_value_string(env, "owner").into())]));
2930 edges.push(state(&[("source", format!("run:{}", slug(run_id)).into()), ("target", id.into()), ("kind", "has_failure".into())]));
2931 }
2932 let nodes = dedupe_states(nodes, "id");
2933 state(&[
2934 ("schema_version", "agentledger.failure.causal_graph.v1".into()),
2935 ("run_id", run_id.into()),
2936 ("nodes", Value::Array(nodes.iter().cloned().map(Value::Object).collect())),
2937 ("edges", Value::Array(edges.iter().cloned().map(Value::Object).collect())),
2938 ("summary", Value::Object(state(&[
2939 ("node_count", Value::Number(nodes.len() as f64)),
2940 ("edge_count", Value::Number(edges.len() as f64)),
2941 ("failure_node_count", Value::Number(count_state_field(&nodes, "kind", "failure") as f64)),
2942 ]))),
2943 ])
2944}
2945
2946fn failure_replay_plan(run_id: &str, envelopes: &[State], ledger: &[ToolLedgerEntry], events: &[Event]) -> State {
2947 let mut actions = Vec::new();
2948 let mut unsafe_count = 0;
2949 let mut manual_count = 0;
2950 for env in envelopes {
2951 let status = state_value_string(env, "status");
2952 let recoverability = state_value_string(env, "recoverability");
2953 let mut action = state(&[
2954 ("failure_id", state_value_string(env, "failure_id").into()),
2955 ("category", state_value_string(env, "category").into()),
2956 ("status", status.clone().into()),
2957 ("replay_action", "reuse_recorded_evidence".into()),
2958 ("replay_safe", Value::Bool(true)),
2959 ("requires_manual_verification", Value::Bool(false)),
2960 ("reason", "recorded runtime evidence can be inspected without calling external systems".into()),
2961 ]);
2962 if status == "unknown_side_effect" || recoverability == "manual_verification" {
2963 action.insert("replay_action".into(), "manual_verify_side_effect".into());
2964 action.insert("replay_safe".into(), Value::Bool(false));
2965 action.insert("requires_manual_verification".into(), Value::Bool(true));
2966 action.insert("reason".into(), "Tool Ledger recorded an unknown side-effect state".into());
2967 unsafe_count += 1;
2968 manual_count += 1;
2969 } else if status == "waiting_human" {
2970 action.insert("replay_action".into(), "resume_after_approval".into());
2971 } else if status == "recovery_scheduled" {
2972 action.insert("replay_action".into(), "retry_from_checkpoint".into());
2973 } else if status == "terminal" || status == "blocked" {
2974 action.insert("replay_action".into(), "terminal_stop".into());
2975 }
2976 actions.push(action);
2977 }
2978 state(&[
2979 ("schema_version", "agentledger.failure.replay_plan.v1".into()),
2980 ("run_id", run_id.into()),
2981 ("mode", "evidence_only".into()),
2982 ("safe_to_replay", Value::Bool(unsafe_count == 0)),
2983 ("unsafe_side_effect_count", Value::Number(unsafe_count as f64)),
2984 ("manual_verification_count", Value::Number(manual_count as f64)),
2985 ("recorded_tool_call_count", Value::Number(ledger.len() as f64)),
2986 ("recorded_event_count", Value::Number(events.len() as f64)),
2987 ("actions", Value::Array(actions.into_iter().map(Value::Object).collect())),
2988 ])
2989}
2990
2991fn failure_alerts(run_id: &str, envelopes: &[State], replay_plan: &State) -> State {
2992 let mut alerts = Vec::new();
2993 if count_state_field(envelopes, "status", "terminal") > 0 {
2994 alerts.push(alert_state(run_id, "terminal_failure", "risk", "terminal failure recorded"));
2995 }
2996 if count_state_field(envelopes, "status", "unknown_side_effect") > 0 {
2997 alerts.push(alert_state(run_id, "unknown_side_effect", "risk", "tool side-effect state requires manual verification"));
2998 }
2999 if state_number(replay_plan, "unsafe_side_effect_count") > 0.0 {
3000 alerts.push(alert_state(run_id, "unsafe_replay_blocked", "risk", "failure replay plan blocks unsafe automatic replay"));
3001 }
3002 state(&[
3003 ("schema_version", "agentledger.failure.alerts.v1".into()),
3004 ("run_id", run_id.into()),
3005 ("alerts", Value::Array(alerts.iter().cloned().map(Value::Object).collect())),
3006 ("alert_count", Value::Number(alerts.len() as f64)),
3007 ])
3008}
3009
3010fn alert_state(run_id: &str, kind: &str, severity: &str, message: &str) -> State {
3011 state(&[
3012 ("schema_version", "agentledger.failure.alerts.v1".into()),
3013 ("run_id", run_id.into()),
3014 ("kind", kind.into()),
3015 ("severity", severity.into()),
3016 ("message", message.into()),
3017 ])
3018}
3019
3020fn failure_export(
3021 run_id: &str,
3022 run_status: &str,
3023 summary: &State,
3024 envelopes: &[State],
3025 lifecycle: &State,
3026 graph: &State,
3027 replay_plan: &State,
3028 alerts: &State,
3029) -> State {
3030 state(&[
3031 ("schema_version", "agentledger.failure.export.v1".into()),
3032 ("run_id", run_id.into()),
3033 ("run_status", run_status.into()),
3034 ("summary", Value::Object(summary.clone())),
3035 ("failure_envelopes", Value::Array(envelopes.iter().cloned().map(Value::Object).collect())),
3036 ("failure_lifecycle", Value::Object(lifecycle.clone())),
3037 ("failure_causal_graph", Value::Object(graph.clone())),
3038 ("failure_replay_plan", Value::Object(replay_plan.clone())),
3039 ("failure_alerts", Value::Object(alerts.clone())),
3040 ("external_mappings", Value::Object(state(&[
3041 ("opentelemetry", Value::Object(state(&[("span_event_count", Value::Number(state_array_len(lifecycle, "events") as f64))]))),
3042 ("langfuse", Value::Object(state(&[("trace_id", run_id.into()), ("observation_count", Value::Number(envelopes.len() as f64))]))),
3043 ("langsmith", Value::Object(state(&[("run_id", run_id.into()), ("feedback_count", Value::Number(envelopes.len() as f64))]))),
3044 ("temporal", Value::Object(state(&[("workflow_id", run_id.into()), ("failure_count", Value::Number(envelopes.len() as f64)), ("safe_to_replay", replay_plan.get("safe_to_replay").cloned().unwrap_or(Value::Bool(false)))]))),
3045 ]))),
3046 ])
3047}
3048
3049fn encode_store(store: &MemoryStore) -> String {
3050 let mut lines = vec!["AGENTLEDGER_RUST_STORE_V1".to_string()];
3051 let mut runs: Vec<_> = store.runs.values().collect();
3052 runs.sort_by(|a, b| a.run_id.cmp(&b.run_id));
3053 for run in runs {
3054 lines.push(join_fields(&[
3055 "R".to_string(),
3056 hex_encode(&run.run_id),
3057 hex_encode(&run.session_id),
3058 hex_encode(&run.status),
3059 run.state_version.to_string(),
3060 run.created_at.to_string(),
3061 run.updated_at.to_string(),
3062 encode_state(&run.state),
3063 ]));
3064 }
3065
3066 let mut steps: Vec<_> = store.steps.values().collect();
3067 steps.sort_by(|a, b| a.step_id.cmp(&b.step_id));
3068 for step in steps {
3069 lines.push(join_fields(&[
3070 "S".to_string(),
3071 hex_encode(&step.step_id),
3072 hex_encode(&step.run_id),
3073 hex_encode(&step.session_id),
3074 hex_encode(&step.status),
3075 encode_option_string(&step.owner),
3076 encode_option_string(&step.lease_token),
3077 encode_option_f64(step.lease_until),
3078 step.attempt.to_string(),
3079 step.state_version.to_string(),
3080 encode_option_string(&step.checkpoint_id),
3081 encode_option_string(&step.last_error_type),
3082 encode_option_string(&step.last_error),
3083 encode_option_f64(step.cancelled_at),
3084 step.created_at.to_string(),
3085 step.updated_at.to_string(),
3086 ]));
3087 }
3088
3089 let mut run_ids: Vec<_> = store.events.keys().collect();
3090 run_ids.sort();
3091 for run_id in run_ids {
3092 if let Some(events) = store.events.get(run_id) {
3093 for event in events {
3094 lines.push(join_fields(&[
3095 "E".to_string(),
3096 hex_encode(&event.event_id),
3097 hex_encode(&event.run_id),
3098 encode_option_string(&event.session_id),
3099 encode_option_string(&event.step_id),
3100 event.seq.to_string(),
3101 hex_encode(&event.event_type),
3102 event.timestamp.to_string(),
3103 encode_option_string(&event.agent_role),
3104 encode_option_u64(event.state_version),
3105 encode_option_string(&event.causal_token),
3106 hex_encode(&event.payload_hash),
3107 hex_encode(&event.payload_ref),
3108 encode_state(&event.payload),
3109 ]));
3110 }
3111 }
3112 }
3113
3114 let mut ledgers: Vec<_> = store.tool_ledger.values().collect();
3115 ledgers.sort_by(|a, b| a.idempotency_key.cmp(&b.idempotency_key));
3116 for entry in ledgers {
3117 lines.push(join_fields(&[
3118 "L".to_string(),
3119 hex_encode(&entry.ledger_id),
3120 hex_encode(&entry.run_id),
3121 hex_encode(&entry.session_id),
3122 hex_encode(&entry.step_id),
3123 hex_encode(&entry.tool_name),
3124 hex_encode(&entry.tool_version),
3125 hex_encode(&entry.tool_call_id),
3126 hex_encode(&entry.idempotency_key),
3127 hex_encode(&entry.causal_token),
3128 hex_encode(&entry.request_hash),
3129 hex_encode(&entry.request_ref),
3130 hex_encode(&entry.status),
3131 encode_option_string(&entry.external_id),
3132 encode_option_string(&entry.response_hash),
3133 encode_option_string(&entry.response_ref),
3134 encode_option_string(&entry.error_type),
3135 encode_option_value(&entry.response),
3136 entry.created_at.to_string(),
3137 entry.updated_at.to_string(),
3138 ]));
3139 }
3140
3141 let mut approvals: Vec<_> = store.approval_requests.values().collect();
3142 approvals.sort_by(|a, b| a.approval_key.cmp(&b.approval_key));
3143 for approval in approvals {
3144 lines.push(join_fields(&[
3145 "A".to_string(),
3146 hex_encode(&approval.approval_id),
3147 hex_encode(&approval.approval_key),
3148 hex_encode(&approval.run_id),
3149 hex_encode(&approval.session_id),
3150 hex_encode(&approval.step_id),
3151 hex_encode(&approval.tool_name),
3152 hex_encode(&approval.risk_level),
3153 hex_encode(&approval.status),
3154 hex_encode(&approval.reason),
3155 hex_encode(&approval.request_hash),
3156 hex_encode(&approval.request_ref),
3157 hex_encode(&approval.requested_by),
3158 encode_option_string(&approval.approved_by),
3159 encode_option_string(&approval.decision_reason),
3160 approval.created_at.to_string(),
3161 approval.updated_at.to_string(),
3162 ]));
3163 }
3164
3165 let mut cost_run_ids: Vec<_> = store.cost_records.keys().collect();
3166 cost_run_ids.sort();
3167 for run_id in cost_run_ids {
3168 if let Some(records) = store.cost_records.get(run_id) {
3169 for record in records {
3170 lines.push(join_fields(&[
3171 "C".to_string(),
3172 hex_encode(&record.cost_id),
3173 hex_encode(&record.run_id),
3174 hex_encode(&record.session_id),
3175 hex_encode(&record.step_id),
3176 hex_encode(&record.category),
3177 hex_encode(&record.name),
3178 record.amount.to_string(),
3179 hex_encode(&record.unit),
3180 encode_state(&record.metadata),
3181 record.created_at.to_string(),
3182 ]));
3183 }
3184 }
3185 }
3186
3187 let mut artifact_run_ids: Vec<_> = store.artifacts.keys().collect();
3188 artifact_run_ids.sort();
3189 for run_id in artifact_run_ids {
3190 if let Some(artifacts) = store.artifacts.get(run_id) {
3191 for artifact in artifacts {
3192 lines.push(join_fields(&[
3193 "F".to_string(),
3194 hex_encode(&artifact.artifact_id),
3195 hex_encode(&artifact.run_id),
3196 encode_option_string(&artifact.step_id),
3197 hex_encode(&artifact.name),
3198 hex_encode(&artifact.blob_hash),
3199 hex_encode(&artifact.blob_ref),
3200 encode_state(&artifact.metadata),
3201 artifact.created_at.to_string(),
3202 ]));
3203 }
3204 }
3205 }
3206
3207 lines.push(String::new());
3208 lines.join("\n")
3209}
3210
3211fn decode_store(body: &str) -> Result<MemoryStore> {
3212 let mut lines = body.lines();
3213 match lines.next() {
3214 Some("AGENTLEDGER_RUST_STORE_V1") => {}
3215 _ => {
3216 return Err(RuntimeError(
3217 "invalid Rust store snapshot header".to_string(),
3218 ))
3219 }
3220 }
3221 let mut store = MemoryStore::new();
3222 for line in lines.filter(|line| !line.trim().is_empty()) {
3223 let fields: Vec<&str> = line.split('\t').collect();
3224 let tag = fields.first().copied().unwrap_or_default();
3225 match tag {
3226 "R" => {
3227 require_len(tag, &fields, 8)?;
3228 let run = Run {
3229 run_id: hex_decode(fields[1])?,
3230 session_id: hex_decode(fields[2])?,
3231 status: hex_decode(fields[3])?,
3232 state_version: parse_u64(fields[4])?,
3233 created_at: parse_f64(fields[5])?,
3234 updated_at: parse_f64(fields[6])?,
3235 state: decode_state(fields[7])?,
3236 };
3237 store.runs.insert(run.run_id.clone(), run);
3238 }
3239 "S" => {
3240 require_len(tag, &fields, 16)?;
3241 let step = Step {
3242 step_id: hex_decode(fields[1])?,
3243 run_id: hex_decode(fields[2])?,
3244 session_id: hex_decode(fields[3])?,
3245 status: hex_decode(fields[4])?,
3246 owner: decode_option_string(fields[5])?,
3247 lease_token: decode_option_string(fields[6])?,
3248 lease_until: decode_option_f64(fields[7])?,
3249 attempt: parse_u64(fields[8])?,
3250 state_version: parse_u64(fields[9])?,
3251 checkpoint_id: decode_option_string(fields[10])?,
3252 last_error_type: decode_option_string(fields[11])?,
3253 last_error: decode_option_string(fields[12])?,
3254 cancelled_at: decode_option_f64(fields[13])?,
3255 created_at: parse_f64(fields[14])?,
3256 updated_at: parse_f64(fields[15])?,
3257 };
3258 store.steps.insert(step.step_id.clone(), step);
3259 }
3260 "E" => {
3261 require_len(tag, &fields, 14)?;
3262 let event = Event {
3263 event_id: hex_decode(fields[1])?,
3264 run_id: hex_decode(fields[2])?,
3265 session_id: decode_option_string(fields[3])?,
3266 step_id: decode_option_string(fields[4])?,
3267 seq: parse_u64(fields[5])?,
3268 event_type: hex_decode(fields[6])?,
3269 timestamp: parse_f64(fields[7])?,
3270 agent_role: decode_option_string(fields[8])?,
3271 state_version: decode_option_u64(fields[9])?,
3272 causal_token: decode_option_string(fields[10])?,
3273 payload_hash: hex_decode(fields[11])?,
3274 payload_ref: hex_decode(fields[12])?,
3275 payload: decode_state(fields[13])?,
3276 };
3277 store
3278 .events
3279 .entry(event.run_id.clone())
3280 .or_default()
3281 .push(event);
3282 }
3283 "L" => {
3284 require_len(tag, &fields, 20)?;
3285 let entry = ToolLedgerEntry {
3286 ledger_id: hex_decode(fields[1])?,
3287 run_id: hex_decode(fields[2])?,
3288 session_id: hex_decode(fields[3])?,
3289 step_id: hex_decode(fields[4])?,
3290 tool_name: hex_decode(fields[5])?,
3291 tool_version: hex_decode(fields[6])?,
3292 tool_call_id: hex_decode(fields[7])?,
3293 idempotency_key: hex_decode(fields[8])?,
3294 causal_token: hex_decode(fields[9])?,
3295 request_hash: hex_decode(fields[10])?,
3296 request_ref: hex_decode(fields[11])?,
3297 status: hex_decode(fields[12])?,
3298 external_id: decode_option_string(fields[13])?,
3299 response_hash: decode_option_string(fields[14])?,
3300 response_ref: decode_option_string(fields[15])?,
3301 error_type: decode_option_string(fields[16])?,
3302 response: decode_option_value(fields[17])?,
3303 created_at: parse_f64(fields[18])?,
3304 updated_at: parse_f64(fields[19])?,
3305 };
3306 store
3307 .tool_ledger
3308 .insert(entry.idempotency_key.clone(), entry);
3309 }
3310 "A" => {
3311 require_len(tag, &fields, 17)?;
3312 let approval = ApprovalRequest {
3313 approval_id: hex_decode(fields[1])?,
3314 approval_key: hex_decode(fields[2])?,
3315 run_id: hex_decode(fields[3])?,
3316 session_id: hex_decode(fields[4])?,
3317 step_id: hex_decode(fields[5])?,
3318 tool_name: hex_decode(fields[6])?,
3319 risk_level: hex_decode(fields[7])?,
3320 status: hex_decode(fields[8])?,
3321 reason: hex_decode(fields[9])?,
3322 request_hash: hex_decode(fields[10])?,
3323 request_ref: hex_decode(fields[11])?,
3324 requested_by: hex_decode(fields[12])?,
3325 approved_by: decode_option_string(fields[13])?,
3326 decision_reason: decode_option_string(fields[14])?,
3327 created_at: parse_f64(fields[15])?,
3328 updated_at: parse_f64(fields[16])?,
3329 };
3330 store
3331 .approval_requests
3332 .insert(approval.approval_key.clone(), approval);
3333 }
3334 "C" => {
3335 require_len(tag, &fields, 11)?;
3336 let record = CostRecord {
3337 cost_id: hex_decode(fields[1])?,
3338 run_id: hex_decode(fields[2])?,
3339 session_id: hex_decode(fields[3])?,
3340 step_id: hex_decode(fields[4])?,
3341 category: hex_decode(fields[5])?,
3342 name: hex_decode(fields[6])?,
3343 amount: parse_f64(fields[7])?,
3344 unit: hex_decode(fields[8])?,
3345 metadata: decode_state(fields[9])?,
3346 created_at: parse_f64(fields[10])?,
3347 };
3348 store
3349 .cost_records
3350 .entry(record.run_id.clone())
3351 .or_default()
3352 .push(record);
3353 }
3354 "F" => {
3355 require_len(tag, &fields, 9)?;
3356 let artifact = Artifact {
3357 artifact_id: hex_decode(fields[1])?,
3358 run_id: hex_decode(fields[2])?,
3359 step_id: decode_option_string(fields[3])?,
3360 name: hex_decode(fields[4])?,
3361 blob_hash: hex_decode(fields[5])?,
3362 blob_ref: hex_decode(fields[6])?,
3363 metadata: decode_state(fields[7])?,
3364 created_at: parse_f64(fields[8])?,
3365 };
3366 store
3367 .artifacts
3368 .entry(artifact.run_id.clone())
3369 .or_default()
3370 .push(artifact);
3371 }
3372 _ => {
3373 return Err(RuntimeError(format!(
3374 "unknown Rust store snapshot row: {tag}"
3375 )))
3376 }
3377 }
3378 }
3379 Ok(store)
3380}
3381
3382fn join_fields(fields: &[String]) -> String {
3383 fields.join("\t")
3384}
3385
3386fn require_len(tag: &str, fields: &[&str], expected: usize) -> Result<()> {
3387 if fields.len() != expected {
3388 return Err(RuntimeError(format!(
3389 "invalid {tag} row: expected {expected} fields, got {}",
3390 fields.len()
3391 )));
3392 }
3393 Ok(())
3394}
3395
3396fn encode_option_string(value: &Option<String>) -> String {
3397 value
3398 .as_ref()
3399 .map(|item| hex_encode(item))
3400 .unwrap_or_else(|| "-".to_string())
3401}
3402
3403fn decode_option_string(value: &str) -> Result<Option<String>> {
3404 if value == "-" {
3405 Ok(None)
3406 } else {
3407 Ok(Some(hex_decode(value)?))
3408 }
3409}
3410
3411fn encode_option_f64(value: Option<f64>) -> String {
3412 value
3413 .map(|item| item.to_string())
3414 .unwrap_or_else(|| "-".to_string())
3415}
3416
3417fn decode_option_f64(value: &str) -> Result<Option<f64>> {
3418 if value == "-" {
3419 Ok(None)
3420 } else {
3421 Ok(Some(parse_f64(value)?))
3422 }
3423}
3424
3425fn encode_option_u64(value: Option<u64>) -> String {
3426 value
3427 .map(|item| item.to_string())
3428 .unwrap_or_else(|| "-".to_string())
3429}
3430
3431fn decode_option_u64(value: &str) -> Result<Option<u64>> {
3432 if value == "-" {
3433 Ok(None)
3434 } else {
3435 Ok(Some(parse_u64(value)?))
3436 }
3437}
3438
3439fn encode_option_value(value: &Option<Value>) -> String {
3440 value
3441 .as_ref()
3442 .map(encode_value)
3443 .unwrap_or_else(|| "-".to_string())
3444}
3445
3446fn decode_option_value(value: &str) -> Result<Option<Value>> {
3447 if value == "-" {
3448 Ok(None)
3449 } else {
3450 Ok(Some(decode_value(value)?))
3451 }
3452}
3453
3454fn encode_state(state: &State) -> String {
3455 encode_value(&Value::Object(state.clone()))
3456}
3457
3458fn decode_state(encoded: &str) -> Result<State> {
3459 match decode_value(encoded)? {
3460 Value::Object(state) => Ok(state),
3461 _ => Err(RuntimeError("encoded state was not an object".to_string())),
3462 }
3463}
3464
3465fn encode_value(value: &Value) -> String {
3466 match value {
3467 Value::Null => "Z".to_string(),
3468 Value::Bool(true) => "T".to_string(),
3469 Value::Bool(false) => "F".to_string(),
3470 Value::Number(item) => format!("N{}:", hex_encode(&item.to_string())),
3471 Value::String(item) => format!("S{}:", hex_encode(item)),
3472 Value::Object(state) => {
3473 let mut keys: Vec<_> = state.keys().collect();
3474 keys.sort();
3475 let mut out = format!("O{}:", keys.len());
3476 for key in keys {
3477 out.push_str(&hex_encode(key));
3478 out.push(':');
3479 out.push_str(&encode_value(&state[key]));
3480 }
3481 out
3482 }
3483 Value::Array(values) => {
3484 let mut out = format!("A{}:", values.len());
3485 for value in values {
3486 out.push_str(&encode_value(value));
3487 }
3488 out
3489 }
3490 }
3491}
3492
3493fn decode_value(encoded: &str) -> Result<Value> {
3494 let (value, index) = parse_value(encoded, 0)?;
3495 if index != encoded.len() {
3496 return Err(RuntimeError("trailing bytes in encoded value".to_string()));
3497 }
3498 Ok(value)
3499}
3500
3501fn parse_value(input: &str, index: usize) -> Result<(Value, usize)> {
3502 let bytes = input.as_bytes();
3503 let tag = *bytes
3504 .get(index)
3505 .ok_or_else(|| RuntimeError("unexpected end of encoded value".to_string()))?
3506 as char;
3507 match tag {
3508 'Z' => Ok((Value::Null, index + 1)),
3509 'T' => Ok((Value::Bool(true), index + 1)),
3510 'F' => Ok((Value::Bool(false), index + 1)),
3511 'N' => {
3512 let (hex, next) = read_until_colon(input, index + 1)?;
3513 Ok((Value::Number(parse_f64(&hex_decode(hex)?)?), next))
3514 }
3515 'S' => {
3516 let (hex, next) = read_until_colon(input, index + 1)?;
3517 Ok((Value::String(hex_decode(hex)?), next))
3518 }
3519 'O' => {
3520 let (count_text, mut next) = read_until_colon(input, index + 1)?;
3521 let count = count_text
3522 .parse::<usize>()
3523 .map_err(|err| RuntimeError(err.to_string()))?;
3524 let mut state = State::new();
3525 for _ in 0..count {
3526 let (key_hex, after_key) = read_until_colon(input, next)?;
3527 let key = hex_decode(key_hex)?;
3528 let (value, after_value) = parse_value(input, after_key)?;
3529 state.insert(key, value);
3530 next = after_value;
3531 }
3532 Ok((Value::Object(state), next))
3533 }
3534 'A' => {
3535 let (count_text, mut next) = read_until_colon(input, index + 1)?;
3536 let count = count_text
3537 .parse::<usize>()
3538 .map_err(|err| RuntimeError(err.to_string()))?;
3539 let mut values = Vec::with_capacity(count);
3540 for _ in 0..count {
3541 let (value, after_value) = parse_value(input, next)?;
3542 values.push(value);
3543 next = after_value;
3544 }
3545 Ok((Value::Array(values), next))
3546 }
3547 _ => Err(RuntimeError(format!("unknown encoded value tag: {tag}"))),
3548 }
3549}
3550
3551fn read_until_colon(input: &str, index: usize) -> Result<(&str, usize)> {
3552 let rest = input
3553 .get(index..)
3554 .ok_or_else(|| RuntimeError("invalid encoded value index".to_string()))?;
3555 let offset = rest
3556 .find(':')
3557 .ok_or_else(|| RuntimeError("missing encoded value delimiter".to_string()))?;
3558 Ok((&rest[..offset], index + offset + 1))
3559}
3560
3561fn hex_encode(value: &str) -> String {
3562 value
3563 .as_bytes()
3564 .iter()
3565 .map(|byte| format!("{byte:02x}"))
3566 .collect()
3567}
3568
3569fn hex_decode(value: &str) -> Result<String> {
3570 if value.len() % 2 != 0 {
3571 return Err(RuntimeError("invalid hex string length".to_string()));
3572 }
3573 let mut bytes = Vec::with_capacity(value.len() / 2);
3574 for index in (0..value.len()).step_by(2) {
3575 let byte = u8::from_str_radix(&value[index..index + 2], 16)
3576 .map_err(|err| RuntimeError(err.to_string()))?;
3577 bytes.push(byte);
3578 }
3579 String::from_utf8(bytes).map_err(|err| RuntimeError(err.to_string()))
3580}
3581
3582fn parse_f64(value: &str) -> Result<f64> {
3583 value
3584 .parse::<f64>()
3585 .map_err(|err| RuntimeError(err.to_string()))
3586}
3587
3588fn parse_u64(value: &str) -> Result<u64> {
3589 value
3590 .parse::<u64>()
3591 .map_err(|err| RuntimeError(err.to_string()))
3592}
3593
3594fn merge_patch(base: &State, patch: &State) -> State {
3595 let mut out = base.clone();
3596 for (key, value) in patch {
3597 match value {
3598 Value::Null => {
3599 out.remove(key);
3600 }
3601 Value::Object(patch_map) => {
3602 if let Some(Value::Object(base_map)) = out.get(key) {
3603 out.insert(key.clone(), Value::Object(merge_patch(base_map, patch_map)));
3604 } else {
3605 out.insert(key.clone(), value.clone());
3606 }
3607 }
3608 _ => {
3609 out.insert(key.clone(), value.clone());
3610 }
3611 }
3612 }
3613 out
3614}
3615
3616fn add_cost(summary: &mut CostSummary, record: &CostRecord) {
3617 if (record.category == "tool" || record.category == "tool_shadow") && record.unit == "call" {
3618 summary.tool_calls += record.amount;
3619 }
3620 if record.category == "model" && record.unit == "token" {
3621 summary.model_tokens += record.amount;
3622 }
3623 if record.unit == "usd" {
3624 summary.total_usd += record.amount;
3625 }
3626 let key = format!("{}:{}", record.category, record.unit);
3627 *summary.by_category.entry(key).or_insert(0.0) += record.amount;
3628}
3629
3630pub fn validate_tool_schema(schema: &Value, value: &Value, path: &str) -> Result<()> {
3631 let schema = match schema {
3632 Value::Object(schema) => schema,
3633 _ => return Err(RuntimeError(format!("{path} schema must be object"))),
3634 };
3635 if let Some(expected) = schema.get("const") {
3636 if expected != value {
3637 return Err(RuntimeError(format!("{path} expected const")));
3638 }
3639 }
3640 if let Some(Value::Array(items)) = schema.get("enum") {
3641 if !items.iter().any(|item| item == value) {
3642 return Err(RuntimeError(format!("{path} value not in enum")));
3643 }
3644 }
3645 let Some(Value::String(kind)) = schema.get("type") else {
3646 return Ok(());
3647 };
3648 match kind.as_str() {
3649 "object" => {
3650 let Value::Object(object) = value else {
3651 return Err(RuntimeError(format!("{path} expected object")));
3652 };
3653 if let Some(Value::Array(required)) = schema.get("required") {
3654 for item in required {
3655 if let Value::String(key) = item {
3656 if !object.contains_key(key) {
3657 return Err(RuntimeError(format!("{path}.{key} is required")));
3658 }
3659 }
3660 }
3661 }
3662 let properties = match schema.get("properties") {
3663 Some(Value::Object(properties)) => properties,
3664 _ => return Ok(()),
3665 };
3666 for (key, child_schema) in properties {
3667 if let Some(child) = object.get(key) {
3668 validate_tool_schema(child_schema, child, &format!("{path}.{key}"))?;
3669 }
3670 }
3671 if schema.get("additionalProperties") == Some(&Value::Bool(false)) {
3672 for key in object.keys() {
3673 if !properties.contains_key(key) {
3674 return Err(RuntimeError(format!("{path}.{key} is not allowed")));
3675 }
3676 }
3677 }
3678 }
3679 "string" => {
3680 let Value::String(text) = value else {
3681 return Err(RuntimeError(format!("{path} expected string")));
3682 };
3683 if let Some(Value::Number(min)) = schema.get("minLength") {
3684 if (text.len() as f64) < *min {
3685 return Err(RuntimeError(format!("{path} shorter than minLength")));
3686 }
3687 }
3688 if let Some(Value::Number(max)) = schema.get("maxLength") {
3689 if (text.len() as f64) > *max {
3690 return Err(RuntimeError(format!("{path} longer than maxLength")));
3691 }
3692 }
3693 }
3694 "number" | "integer" => {
3695 let Value::Number(number) = value else {
3696 return Err(RuntimeError(format!("{path} expected number")));
3697 };
3698 if kind == "integer" && number.fract() != 0.0 {
3699 return Err(RuntimeError(format!("{path} expected integer")));
3700 }
3701 if let Some(Value::Number(min)) = schema.get("minimum") {
3702 if number < min {
3703 return Err(RuntimeError(format!("{path} below minimum")));
3704 }
3705 }
3706 if let Some(Value::Number(max)) = schema.get("maximum") {
3707 if number > max {
3708 return Err(RuntimeError(format!("{path} above maximum")));
3709 }
3710 }
3711 }
3712 "boolean" => {
3713 if !matches!(value, Value::Bool(_)) {
3714 return Err(RuntimeError(format!("{path} expected boolean")));
3715 }
3716 }
3717 _ => {}
3718 }
3719 Ok(())
3720}
3721
3722fn is_high_risk(risk: &str) -> bool {
3723 matches!(
3724 risk,
3725 "high" | "destructive" | "sensitive" | "financial_or_legal"
3726 )
3727}
3728
3729fn is_media_kind(kind: &str) -> bool {
3730 matches!(
3731 kind,
3732 "image"
3733 | "audio"
3734 | "video"
3735 | "frame"
3736 | "audio_segment"
3737 | "video_segment"
3738 | "transcript"
3739 | "embedding"
3740 | "derived"
3741 )
3742}
3743
3744fn stream_chunk_to_state(chunk: StreamChunkRef) -> State {
3745 let mut state = State::new();
3746 state.insert(
3747 "schema_version".to_string(),
3748 Value::String(STREAM_SCHEMA_VERSION.to_string()),
3749 );
3750 state.insert("stream_id".to_string(), Value::String(chunk.stream_id));
3751 state.insert("chunk_id".to_string(), Value::String(chunk.chunk_id));
3752 state.insert("offset".to_string(), chunk.offset);
3753 if let Some(content_ref) = chunk.content_ref {
3754 state.insert("content_ref".to_string(), Value::String(content_ref));
3755 }
3756 if let Some(content_hash) = chunk.content_hash {
3757 state.insert("content_hash".to_string(), Value::String(content_hash));
3758 }
3759 if let Some(sequence) = chunk.sequence {
3760 state.insert("sequence".to_string(), Value::Number(sequence));
3761 }
3762 if let Some(event_time) = chunk.event_time {
3763 state.insert("event_time".to_string(), Value::Number(event_time));
3764 }
3765 if !chunk.metadata.is_empty() {
3766 state.insert("metadata".to_string(), Value::Object(chunk.metadata));
3767 }
3768 state
3769}
3770
3771fn media_artifacts_from(artifacts: &[Artifact]) -> Vec<State> {
3772 artifacts
3773 .iter()
3774 .filter_map(
3775 |artifact| match artifact.metadata.get("agentledger_media") {
3776 Some(Value::Object(metadata)) => {
3777 let mut row = State::new();
3778 row.insert(
3779 "artifact_id".to_string(),
3780 Value::String(artifact.artifact_id.clone()),
3781 );
3782 row.insert("name".to_string(), Value::String(artifact.name.clone()));
3783 row.insert(
3784 "blob_hash".to_string(),
3785 Value::String(artifact.blob_hash.clone()),
3786 );
3787 row.insert(
3788 "blob_ref".to_string(),
3789 Value::String(artifact.blob_ref.clone()),
3790 );
3791 for key in ["kind", "uri", "content_ref", "metadata", "lineage"] {
3792 if let Some(value) = metadata.get(key) {
3793 row.insert(key.to_string(), value.clone());
3794 }
3795 }
3796 Some(row)
3797 }
3798 _ => None,
3799 },
3800 )
3801 .collect()
3802}
3803
3804fn stream_checkpoints_from(artifacts: &[Artifact]) -> Vec<State> {
3805 artifacts
3806 .iter()
3807 .filter_map(
3808 |artifact| match artifact.metadata.get("agentledger_stream") {
3809 Some(Value::Object(metadata)) => {
3810 let mut row = State::new();
3811 row.insert(
3812 "artifact_id".to_string(),
3813 Value::String(artifact.artifact_id.clone()),
3814 );
3815 row.insert("name".to_string(), Value::String(artifact.name.clone()));
3816 row.insert(
3817 "blob_hash".to_string(),
3818 Value::String(artifact.blob_hash.clone()),
3819 );
3820 row.insert(
3821 "blob_ref".to_string(),
3822 Value::String(artifact.blob_ref.clone()),
3823 );
3824 for key in [
3825 "stream_id",
3826 "consumer_id",
3827 "offset",
3828 "watermark",
3829 "chunk",
3830 "partial_result_ref",
3831 "backpressure",
3832 ] {
3833 if let Some(value) = metadata.get(key) {
3834 row.insert(key.to_string(), value.clone());
3835 }
3836 }
3837 Some(row)
3838 }
3839 _ => None,
3840 },
3841 )
3842 .collect()
3843}
3844
3845fn classify_runtime_error(message: &str) -> &'static str {
3846 if message.contains("budget exceeded") || message.contains("budget") {
3847 "BudgetExceededError"
3848 } else if message.contains("sandbox executor") {
3849 "SandboxUnavailableError"
3850 } else if message.contains("high-risk") || message.contains("denied") {
3851 "PermissionDeniedError"
3852 } else {
3853 "RuntimeError"
3854 }
3855}
3856
3857fn failure_source(error_type: &str) -> &'static str {
3858 match error_type {
3859 "BudgetExceededError" => "budget",
3860 "SandboxUnavailableError" => "sandbox",
3861 "PermissionDeniedError" | "ApprovalDenied" => "policy",
3862 _ => "agent",
3863 }
3864}
3865
3866fn is_failure_event(kind: &str) -> bool {
3867 matches!(
3868 kind,
3869 "failure_classified"
3870 | "error_raised"
3871 | "step_failed"
3872 | "step_retry_scheduled"
3873 | "step_waiting_human"
3874 | "lease_expired"
3875 | "run_cancel_requested"
3876 | "run_cancelled"
3877 | "tool_call_failed"
3878 | "tool_approval_required"
3879 | "budget_check_failed"
3880 )
3881}
3882
3883fn failure_category(text: &str, fallback: &str) -> String {
3884 let lower = text.to_lowercase();
3885 for category in ["sandbox", "budget", "policy", "model", "tool", "runtime"] {
3886 if lower.contains(category) {
3887 return category.to_string();
3888 }
3889 }
3890 if lower.contains("approval") || lower.contains("permission") || lower.contains("denied") {
3891 return "policy".to_string();
3892 }
3893 if lower.contains("lease") || lower.contains("worker") {
3894 return "runtime".to_string();
3895 }
3896 if lower.contains("cancel") {
3897 return "cancellation".to_string();
3898 }
3899 fallback.to_string()
3900}
3901
3902fn event_failure_category(event: &Event) -> String {
3903 match event.event_type.as_str() {
3904 "tool_call_failed" | "tool_call_blocked" | "tool_approval_required" => "tool".to_string(),
3905 "run_cancel_requested" | "run_cancelled" | "step_cancelled" => "cancellation".to_string(),
3906 "lease_expired" => "runtime".to_string(),
3907 "step_retry_scheduled" => "retry".to_string(),
3908 "step_waiting_human" => "approval".to_string(),
3909 _ => failure_category(
3910 &format!(
3911 "{} {} {} {}",
3912 event.event_type,
3913 state_value_string(&event.payload, "error_type"),
3914 state_value_string(&event.payload, "error"),
3915 state_value_string(&event.payload, "reason")
3916 ),
3917 "agent",
3918 ),
3919 }
3920}
3921
3922fn event_failure_status(kind: &str, run_status: &str) -> String {
3923 match kind {
3924 "step_failed" | "run_cancelled" | "step_cancelled" => "terminal".to_string(),
3925 "tool_call_blocked" => "blocked".to_string(),
3926 "step_retry_scheduled" | "lease_expired" => "recovery_scheduled".to_string(),
3927 "step_waiting_human" | "tool_approval_required" => "waiting_human".to_string(),
3928 "failure_classified" => "classified".to_string(),
3929 "error_raised" if run_status == "failed" => "terminal".to_string(),
3930 _ => "failed".to_string(),
3931 }
3932}
3933
3934fn event_recoverability(kind: &str, run_status: &str) -> String {
3935 if run_status == "failed" && matches!(kind, "step_failed" | "run_cancelled" | "step_cancelled") {
3936 return "terminal".to_string();
3937 }
3938 match kind {
3939 "step_retry_scheduled" | "lease_expired" => "auto_retry".to_string(),
3940 "step_waiting_human" | "tool_approval_required" => "human_required".to_string(),
3941 "tool_call_blocked" => "manual_intervention".to_string(),
3942 _ => "unknown".to_string(),
3943 }
3944}
3945
3946fn event_retryability(kind: &str) -> String {
3947 match kind {
3948 "step_retry_scheduled" | "lease_expired" => "retryable".to_string(),
3949 "tool_call_blocked" | "run_cancelled" | "step_cancelled" => "not_retryable".to_string(),
3950 _ => "unknown".to_string(),
3951 }
3952}
3953
3954fn owner_for_failure_category(category: &str) -> String {
3955 match category {
3956 "tool" | "model" | "policy" | "sandbox" | "budget" | "runtime" => category.to_string(),
3957 "approval" | "cancellation" | "retry" => "runtime".to_string(),
3958 _ => "agent".to_string(),
3959 }
3960}
3961
3962fn first_text(values: &[Option<&str>]) -> String {
3963 for value in values {
3964 if let Some(text) = value {
3965 if !text.is_empty() {
3966 return (*text).to_string();
3967 }
3968 }
3969 }
3970 "failure signal".to_string()
3971}
3972
3973fn ref_state(kind: &str, value: &str) -> State {
3974 if value.is_empty() {
3975 return State::new();
3976 }
3977 state(&[("kind", kind.into()), ("value", value.into())])
3978}
3979
3980fn state_value_string(row: &State, key: &str) -> String {
3981 match row.get(key) {
3982 Some(Value::String(value)) => value.clone(),
3983 Some(Value::Number(value)) => value.to_string(),
3984 Some(Value::Bool(value)) => value.to_string(),
3985 Some(value) => format_value(value),
3986 None => String::new(),
3987 }
3988}
3989
3990fn state_string(row: &State, key: &str) -> Option<String> {
3991 let value = state_value_string(row, key);
3992 if value.is_empty() {
3993 None
3994 } else {
3995 Some(value)
3996 }
3997}
3998
3999fn state_number(row: &State, key: &str) -> f64 {
4000 match row.get(key) {
4001 Some(Value::Number(value)) => *value,
4002 _ => 0.0,
4003 }
4004}
4005
4006fn state_array_len(row: &State, key: &str) -> usize {
4007 match row.get(key) {
4008 Some(Value::Array(values)) => values.len(),
4009 _ => 0,
4010 }
4011}
4012
4013fn dedupe_states(rows: Vec<State>, key: &str) -> Vec<State> {
4014 let mut out = Vec::new();
4015 let mut seen = std::collections::HashSet::new();
4016 for row in rows {
4017 let value = state_value_string(&row, key);
4018 if value.is_empty() || seen.contains(&value) {
4019 continue;
4020 }
4021 seen.insert(value);
4022 out.push(row);
4023 }
4024 out
4025}
4026
4027fn count_state_field(rows: &[State], key: &str, expected: &str) -> usize {
4028 rows.iter()
4029 .filter(|row| state_value_string(row, key) == expected)
4030 .count()
4031}
4032
4033fn count_recoverable_failures(rows: &[State]) -> usize {
4034 rows.iter()
4035 .filter(|row| matches!(state_value_string(row, "recoverability").as_str(), "auto_retry" | "recoverable" | "manual_verification" | "human_required"))
4036 .count()
4037}
4038
4039fn slug(value: &str) -> String {
4040 let mut out = String::new();
4041 let mut previous_dash = false;
4042 for ch in value.chars() {
4043 if ch.is_ascii_alphanumeric() {
4044 out.push(ch.to_ascii_lowercase());
4045 previous_dash = false;
4046 } else if !previous_dash {
4047 out.push('-');
4048 previous_dash = true;
4049 }
4050 }
4051 let trimmed = out.trim_matches('-').to_string();
4052 if trimmed.is_empty() {
4053 "unknown".to_string()
4054 } else {
4055 trimmed
4056 }
4057}
4058
4059fn format_state(state: &State) -> String {
4060 let mut keys: Vec<&String> = state.keys().collect();
4061 keys.sort();
4062 keys.into_iter()
4063 .map(|key| format!("{}={}", key, format_value(&state[key])))
4064 .collect::<Vec<_>>()
4065 .join(",")
4066}
4067
4068fn format_value(value: &Value) -> String {
4069 match value {
4070 Value::Null => "null".to_string(),
4071 Value::Bool(value) => value.to_string(),
4072 Value::Number(value) => value.to_string(),
4073 Value::String(value) => format!("\"{}\"", value),
4074 Value::Object(value) => format!("{{{}}}", format_state(value)),
4075 Value::Array(values) => format!(
4076 "[{}]",
4077 values
4078 .iter()
4079 .map(format_value)
4080 .collect::<Vec<_>>()
4081 .join(",")
4082 ),
4083 }
4084}
4085
4086fn stable_hash(input: &str) -> String {
4087 let mut hash: u64 = 0xcbf29ce484222325;
4088 for byte in input.as_bytes() {
4089 hash ^= *byte as u64;
4090 hash = hash.wrapping_mul(0x100000001b3);
4091 }
4092 format!("{hash:016x}")
4093}
4094
4095fn new_id(prefix: &str) -> String {
4096 let value = ID_COUNTER.fetch_add(1, Ordering::Relaxed);
4097 format!("{prefix}_{value:016x}")
4098}
4099
4100fn now_seconds() -> f64 {
4101 SystemTime::now()
4102 .duration_since(UNIX_EPOCH)
4103 .unwrap()
4104 .as_secs_f64()
4105}
4106
4107#[cfg(test)]
4108mod tests {
4109 use super::*;
4110
4111 fn state(items: &[(&str, Value)]) -> State {
4112 items
4113 .iter()
4114 .map(|(key, value)| ((*key).to_string(), value.clone()))
4115 .collect()
4116 }
4117
4118 fn event_exists(events: &[Event], event_type: &str) -> bool {
4119 events.iter().any(|event| event.event_type == event_type)
4120 }
4121
4122 fn claim_context(
4123 runtime: &mut Runtime,
4124 run_id: &str,
4125 worker: &str,
4126 role: &str,
4127 ) -> AgentContext {
4128 let claim = runtime.store.claim_step(worker, run_id, 60.0).unwrap();
4129 let (_state, version, session_id) = runtime.store.load_state(run_id).unwrap();
4130 let mut payload = State::new();
4131 payload.insert("agent_role".to_string(), Value::String(role.to_string()));
4132 runtime.store.append_event(
4133 run_id,
4134 Some(&session_id),
4135 Some(&claim.step_id),
4136 "agent_started",
4137 payload,
4138 Some(role),
4139 Some(version),
4140 None,
4141 );
4142 AgentContext {
4143 run_id: run_id.to_string(),
4144 session_id,
4145 step_id: claim.step_id,
4146 agent_role: role.to_string(),
4147 lease_token: claim.lease_token,
4148 attempt: claim.attempt,
4149 state_version: version,
4150 pending_patch: State::new(),
4151 }
4152 }
4153
4154 #[test]
4155 fn runtime_creates_evidence_and_replay() {
4156 let mut runtime = Runtime::new();
4157 runtime.register_tool(ToolSpec::new(
4158 "docs.echo",
4159 Box::new(|args| Ok(Value::Object(state(&[("echo", args["text"].clone())])))),
4160 ));
4161 let (run_id, _) = runtime.create_run(state(&[("input", "hello".into())]));
4162 let ok = runtime
4163 .run_once(&run_id, "worker-a", "Researcher", 60.0, |ctx, state| {
4164 let mut result = State::new();
4165 result.insert("from_state".to_string(), state["input"].clone());
4166 ctx.write_state("tool_result", Value::Object(result));
4167 Ok(())
4168 })
4169 .unwrap();
4170 assert!(ok);
4171 let bundle = export_evidence(&runtime.store, &run_id).unwrap();
4172 assert_eq!(bundle.schema_version, "agentledger.evidence.v1");
4173 let summary = replay(&runtime.store, &run_id).unwrap();
4174 assert!(summary.replay_safe);
4175 assert_eq!(summary.event_count, bundle.events.len());
4176 }
4177
4178 #[test]
4179 fn local_snapshot_store_round_trips_completed_run() {
4180 let mut runtime = Runtime::new();
4181 let (run_id, _) = runtime.create_run(state(&[("input", "hello".into())]));
4182 runtime
4183 .run_once(&run_id, "worker-a", "Researcher", 60.0, |ctx, state| {
4184 let mut result = State::new();
4185 result.insert("echo".to_string(), state["input"].clone());
4186 ctx.write_state("tool_result", Value::Object(result));
4187 Ok(())
4188 })
4189 .unwrap();
4190 let path =
4191 std::env::temp_dir().join(format!("agentledger-rust-{}.store", new_id("snapshot")));
4192 runtime.store.save_to_path(&path).unwrap();
4193
4194 let reopened = MemoryStore::load_from_path(&path).unwrap();
4195 let final_state = reopened.final_state(&run_id).unwrap();
4196 assert_eq!(
4197 final_state.get("tool_result"),
4198 Some(&Value::Object(state(&[("echo", "hello".into())])))
4199 );
4200 let bundle = export_evidence(&reopened, &run_id).unwrap();
4201 assert_eq!(reopened.steps(&run_id).len(), 1);
4202 assert_eq!(
4203 replay(&reopened, &run_id).unwrap().event_count,
4204 bundle.events.len()
4205 );
4206 let _ = std::fs::remove_file(path);
4207 }
4208
4209 #[test]
4210 fn local_blob_store_round_trips_json_values() {
4211 let root = std::env::temp_dir().join(format!("agentledger-rust-blobs-{}", new_id("blob")));
4212 let blobs = LocalBlobStore::open(&root).unwrap();
4213 let value = Value::Object(state(&[(
4214 "hello",
4215 Value::Object(state(&[("nested", "world".into())])),
4216 )]));
4217 let first = blobs.put_json(&value).unwrap();
4218 let second = blobs.put_json(&value).unwrap();
4219 assert!(first.0.starts_with("sha256:"));
4220 assert!(first.1.starts_with("blob://sha256/"));
4221 assert_eq!(first, second);
4222 assert_eq!(blobs.get_json(&first.1).unwrap(), value);
4223 assert!(blobs.get_json("unsupported://blob").is_err());
4224 let _ = std::fs::remove_dir_all(root);
4225 }
4226
4227 #[test]
4228 fn tool_schema_validation_rejects_invalid_input() {
4229 let input_schema = Value::Object(state(&[
4230 ("type", "object".into()),
4231 (
4232 "required",
4233 Value::Array(vec![Value::String("text".to_string())]),
4234 ),
4235 ("additionalProperties", false.into()),
4236 (
4237 "properties",
4238 Value::Object(state(&[(
4239 "text",
4240 Value::Object(state(&[("type", "string".into())])),
4241 )])),
4242 ),
4243 ]));
4244 let mut runtime = Runtime::new();
4245 runtime.register_tool(
4246 ToolSpec::new(
4247 "docs.echo",
4248 Box::new(|args| Ok(Value::Object(state(&[("echo", args["text"].clone())])))),
4249 )
4250 .input_schema(input_schema),
4251 );
4252 let (run_id, _) = runtime.create_run(State::new());
4253 let ctx = claim_context(&mut runtime, &run_id, "worker", "SchemaAgent");
4254 assert!(runtime.call_tool(&ctx, "docs.echo", State::new()).is_err());
4255 assert!(runtime.store.events(&run_id).iter().any(|event| {
4256 event.event_type == "tool_call_failed"
4257 && event.payload.get("phase")
4258 == Some(&Value::String("input_validation".to_string()))
4259 }));
4260 }
4261
4262 #[test]
4263 fn tool_ledger_reuses_side_effect_after_retry() {
4264 let mut runtime = Runtime::new();
4265 runtime.register_tool(
4266 ToolSpec::new(
4267 "github.create_pr",
4268 Box::new(|args| {
4269 Ok(Value::Object(state(&[
4270 ("external_id", "pr-123".into()),
4271 ("title", args["title"].clone()),
4272 ])))
4273 }),
4274 )
4275 .side_effect("external")
4276 .idempotency_required(true),
4277 );
4278 let (run_id, _) = runtime.create_run(state(&[("title", "runtime parity".into())]));
4279 let ctx = claim_context(&mut runtime, &run_id, "worker-a", "Coder");
4280 let first = runtime
4281 .call_tool(
4282 &ctx,
4283 "github.create_pr",
4284 state(&[("title", "runtime parity".into())]),
4285 )
4286 .unwrap();
4287 runtime
4288 .store
4289 .mark_retry(&run_id, &ctx.step_id, "RetryableAgentError", "retryable");
4290 let ctx2 = claim_context(&mut runtime, &run_id, "worker-b", "Coder");
4291 let second = runtime
4292 .call_tool(
4293 &ctx2,
4294 "github.create_pr",
4295 state(&[("title", "runtime parity".into())]),
4296 )
4297 .unwrap();
4298 assert_eq!(first, second);
4299 assert_eq!(runtime.store.ledger(&run_id).len(), 1);
4300 }
4301
4302 #[test]
4303 fn policy_denies_unapproved_high_risk_tool() {
4304 let mut runtime = Runtime::new();
4305 runtime.register_tool(
4306 ToolSpec::new("repo.write", Box::new(|_| Ok(Value::Bool(true)))).risk_level("high"),
4307 );
4308 let (run_id, _) = runtime.create_run(State::new());
4309 let ctx = claim_context(&mut runtime, &run_id, "worker", "Reviewer");
4310 let err = runtime
4311 .call_tool(&ctx, "repo.write", state(&[("path", "README.md".into())]))
4312 .unwrap_err();
4313 assert!(err.0.contains("high-risk"));
4314 assert!(runtime
4315 .store
4316 .events(&run_id)
4317 .iter()
4318 .any(|event| event.event_type == "tool_permission_decided"
4319 && event.payload.get("allowed") == Some(&Value::Bool(false))));
4320 }
4321
4322 #[test]
4323 fn approval_pauses_and_resumes_step() {
4324 let mut runtime = Runtime::new();
4325 runtime.register_tool(
4326 ToolSpec::new(
4327 "github.create_pr",
4328 Box::new(|_| Ok(Value::Object(state(&[("external_id", "pr-42".into())])))),
4329 )
4330 .risk_level("high")
4331 .approval_required(true)
4332 .side_effect("external")
4333 .idempotency_required(true),
4334 );
4335 let (run_id, _) = runtime.create_run(State::new());
4336 let ctx = claim_context(&mut runtime, &run_id, "worker-a", "Coder");
4337 let err = runtime
4338 .call_tool(&ctx, "github.create_pr", state(&[("title", "safe".into())]))
4339 .unwrap_err();
4340 assert!(err.0.starts_with("approval required:"));
4341 let approval_id = err.0.trim_start_matches("approval required:").to_string();
4342 runtime
4343 .store
4344 .mark_waiting_human(&run_id, &ctx.step_id, &err.0, &approval_id);
4345 assert_eq!(runtime.store.steps(&run_id)[0].status, "waiting_human");
4346 runtime
4347 .store
4348 .approve_request(&approval_id, "alice", "reviewed")
4349 .unwrap();
4350 let ctx2 = claim_context(&mut runtime, &run_id, "worker-b", "Coder");
4351 let result = runtime
4352 .call_tool(
4353 &ctx2,
4354 "github.create_pr",
4355 state(&[("title", "safe".into())]),
4356 )
4357 .unwrap();
4358 assert!(matches!(result, Value::Object(_)));
4359
4360 let (denied_run, _) = runtime.create_run(State::new());
4361 let denied_ctx = claim_context(&mut runtime, &denied_run, "worker-c", "Coder");
4362 let denied_err = runtime
4363 .call_tool(
4364 &denied_ctx,
4365 "github.create_pr",
4366 state(&[("title", "blocked".into())]),
4367 )
4368 .unwrap_err();
4369 let denied_id = denied_err
4370 .0
4371 .trim_start_matches("approval required:")
4372 .to_string();
4373 runtime.store.mark_waiting_human(
4374 &denied_run,
4375 &denied_ctx.step_id,
4376 &denied_err.0,
4377 &denied_id,
4378 );
4379 runtime
4380 .store
4381 .deny_request(&denied_id, "bob", "not allowed")
4382 .unwrap();
4383 assert_eq!(runtime.store.steps(&denied_run)[0].status, "failed");
4384 }
4385
4386 #[test]
4387 fn mcp_tool_adapter_maps_governance_annotations() {
4388 fn call(_name: &str, _args: State) -> Result<Value> {
4389 Ok(Value::Object(state(&[("ok", true.into())])))
4390 }
4391 let adapter = MCPToolAdapter { client_call: call };
4392 let spec = adapter.tool_spec_from_descriptor(&state(&[
4393 ("name", "mcp.github.create_pr".into()),
4394 (
4395 "inputSchema",
4396 Value::Object(state(&[
4397 ("type", "object".into()),
4398 (
4399 "required",
4400 Value::Array(vec![Value::String("title".to_string())]),
4401 ),
4402 ])),
4403 ),
4404 (
4405 "annotations",
4406 Value::Object(state(&[
4407 ("side_effect", "external_write".into()),
4408 ("risk_level", "high".into()),
4409 ("idempotency_required", true.into()),
4410 ("approval_required", true.into()),
4411 ("sandbox_required", true.into()),
4412 ("sandbox_executor", "docker".into()),
4413 (
4414 "sandbox_policy",
4415 Value::Object(state(&[
4416 ("network", "deny".into()),
4417 ("filesystem", "read-only".into()),
4418 ])),
4419 ),
4420 ])),
4421 ),
4422 ]));
4423 assert_eq!(spec.side_effect, "external_write");
4424 assert_eq!(spec.risk_level, "high");
4425 assert!(spec.idempotency_required);
4426 assert!(spec.approval_required);
4427 assert!(spec.sandbox_required);
4428 assert_eq!(spec.sandbox_executor, "docker");
4429 assert_eq!(
4430 spec.sandbox_policy.get("network"),
4431 Some(&Value::String("deny".to_string()))
4432 );
4433 assert_eq!(
4434 spec.sandbox_policy.get("filesystem"),
4435 Some(&Value::String("read-only".to_string()))
4436 );
4437 assert!(spec.input_schema.is_some());
4438 }
4439
4440 #[test]
4441 fn sandbox_required_tool_fails_closed() {
4442 let mut runtime = Runtime::new();
4443 runtime.register_tool(
4444 ToolSpec::new("shell.exec", Box::new(|_| Ok(Value::Bool(true)))).sandbox_required(true),
4445 );
4446 let (run_id, _) = runtime.create_run(State::new());
4447 let ctx = claim_context(&mut runtime, &run_id, "worker", "Executor");
4448 let err = runtime
4449 .call_tool(
4450 &ctx,
4451 "shell.exec",
4452 state(&[("argv", Value::Object(State::new()))]),
4453 )
4454 .unwrap_err();
4455 assert!(err.0.contains("sandbox executor"));
4456 let events = runtime.store.events(&run_id);
4457 assert!(event_exists(&events, "sandbox_started"));
4458 assert!(event_exists(&events, "tool_call_failed"));
4459 }
4460
4461 #[test]
4462 fn docker_sandbox_executor_requires_explicit_execution() {
4463 let executor = DockerSandboxExecutor::new("fake-image", false).with_binary("/bin/echo");
4464 let result = executor.run_tool(
4465 state(&[("_sandbox_command", Value::Array(vec!["echo".into(), "hi".into()]))]),
4466 &SandboxPolicy {
4467 tool_name: "cmd.echo".to_string(),
4468 run_id: "run".to_string(),
4469 step_id: "step".to_string(),
4470 executor: "docker".to_string(),
4471 network: "deny".to_string(),
4472 filesystem: "read-only".to_string(),
4473 timeout_seconds: 1,
4474 extra: State::new(),
4475 },
4476 );
4477 assert!(!result.ok);
4478 assert_eq!(result.metadata.get("error_type"), Some(&Value::String("SandboxAdapterNotInstalled".to_string())));
4479 }
4480
4481 #[test]
4482 fn docker_sandbox_executor_runs_command_style_tool_with_injected_binary() {
4483 let mut runtime = Runtime::new();
4484 runtime.set_sandbox(Box::new(DockerSandboxExecutor::new("fake-image", true).with_binary("/bin/echo")));
4485 runtime.register_tool(
4486 ToolSpec::new("cmd.echo", Box::new(|_| Err(RuntimeError("direct func should not execute".to_string()))))
4487 .sandbox_required(true)
4488 .sandbox_executor("docker"),
4489 );
4490 let (run_id, _) = runtime.create_run(State::new());
4491 let ctx = claim_context(&mut runtime, &run_id, "worker", "Executor");
4492 let value = runtime
4493 .call_tool(
4494 &ctx,
4495 "cmd.echo",
4496 state(&[("_sandbox_command", Value::Array(vec!["echo".into(), "hi".into()]))]),
4497 )
4498 .unwrap();
4499 let output = match value { Value::Object(output) => output, _ => panic!("expected object output") };
4500 let stdout = match output.get("stdout") { Some(Value::String(value)) => value, _ => panic!("expected stdout") };
4501 assert!(stdout.contains("run"));
4502 assert!(stdout.contains("fake-image"));
4503 let events = runtime.store.events(&run_id);
4504 assert!(event_exists(&events, "sandbox_completed"));
4505 assert!(event_exists(&events, "tool_call_completed"));
4506 }
4507
4508 #[test]
4509 fn cost_budget_and_failure_attribution() {
4510 let mut runtime = Runtime::new();
4511 runtime.set_budget(BudgetLimits {
4512 max_tool_calls: Some(1.0),
4513 max_model_tokens: None,
4514 max_total_usd: None,
4515 });
4516 runtime.register_tool(ToolSpec::new(
4517 "docs.echo",
4518 Box::new(|args| Ok(Value::Object(state(&[("echo", args["text"].clone())])))),
4519 ));
4520 let (run_id, _) = runtime.create_run(State::new());
4521 let ctx = claim_context(&mut runtime, &run_id, "worker", "Researcher");
4522 runtime
4523 .record_model_call(&ctx, "gpt-test", 10.0, 5.0, 0.01)
4524 .unwrap();
4525 runtime
4526 .call_tool(&ctx, "docs.echo", state(&[("text", "first".into())]))
4527 .unwrap();
4528 let err = runtime
4529 .call_tool(&ctx, "docs.echo", state(&[("text", "second".into())]))
4530 .unwrap_err();
4531 runtime.store.mark_failed(
4532 &run_id,
4533 &ctx.step_id,
4534 classify_runtime_error(&err.0),
4535 &err.0,
4536 );
4537 let summary = runtime.store.cost_summary(&run_id);
4538 assert_eq!(summary.tool_calls, 1.0);
4539 assert_eq!(summary.model_tokens, 15.0);
4540 assert_eq!(summary.total_usd, 0.01);
4541 let cost = cost_attribution(&runtime.store, &run_id);
4542 assert_eq!(cost.by_agent["Researcher"].tool_calls, 1.0);
4543 assert_eq!(cost.by_agent["Researcher"].model_tokens, 15.0);
4544 let failure = failure_attribution(&runtime.store, &run_id).unwrap();
4545 assert_eq!(failure.failed_steps.len(), 1);
4546 assert!(event_exists(&failure.failure_events, "budget_check_failed"));
4547 assert!(event_exists(&failure.failure_events, "failure_classified"));
4548 assert!(!failure.failure_envelopes.is_empty());
4549 assert_eq!(
4550 failure.failure_lifecycle["schema_version"],
4551 Value::String("agentledger.failure.lifecycle.v1".to_string())
4552 );
4553 assert_eq!(
4554 failure.failure_export["schema_version"],
4555 Value::String("agentledger.failure.export.v1".to_string())
4556 );
4557 assert_eq!(failure.failure_replay_plan["safe_to_replay"], Value::Bool(true));
4558 assert!(state_number(&failure.failure_alerts, "alert_count") > 0.0);
4559 }
4560
4561 #[test]
4562 fn media_and_stream_artifacts_are_indexed_in_evidence_and_replay() {
4563 let mut runtime = Runtime::new();
4564 let (run_id, _) = runtime.create_run(State::new());
4565 let ctx = claim_context(&mut runtime, &run_id, "worker-media", "MediaAgent");
4566 let mut media_metadata = State::new();
4567 media_metadata.insert(
4568 "mime_type".to_string(),
4569 Value::String("image/jpeg".to_string()),
4570 );
4571 media_metadata.insert("frame_index".to_string(), Value::Number(1.0));
4572 let mut lineage = State::new();
4573 lineage.insert(
4574 "source_blob_ref".to_string(),
4575 Value::String("s3://media/demo/input.mp4".to_string()),
4576 );
4577 lineage.insert(
4578 "tool_call_id".to_string(),
4579 Value::String("video.extract_frames".to_string()),
4580 );
4581 let frame_id = runtime
4582 .create_media_artifact(
4583 &ctx,
4584 "frame-0001",
4585 "frame",
4586 MediaArtifactOptions {
4587 uri: Some("s3://media/demo/frame-0001.jpg".to_string()),
4588 media_metadata,
4589 lineage,
4590 ..Default::default()
4591 },
4592 )
4593 .unwrap();
4594 let checkpoint_id = runtime
4595 .create_stream_checkpoint(
4596 &ctx,
4597 "camera-checkpoint",
4598 StreamCheckpointOptions {
4599 stream_id: "camera-1".to_string(),
4600 consumer_id: "vision-agent".to_string(),
4601 offset: Value::Number(7.0),
4602 watermark: Some(Value::Number(1.5)),
4603 chunk: Some(StreamChunkRef {
4604 stream_id: "camera-1".to_string(),
4605 chunk_id: "chunk-7".to_string(),
4606 offset: Value::Number(7.0),
4607 content_ref: Some("blob://sha256/chunk-7.json".to_string()),
4608 sequence: Some(7.0),
4609 ..Default::default()
4610 }),
4611 ..Default::default()
4612 },
4613 )
4614 .unwrap();
4615 let mut artifacts = State::new();
4616 artifacts.insert("frame".to_string(), Value::String(frame_id));
4617 artifacts.insert("checkpoint".to_string(), Value::String(checkpoint_id));
4618 runtime
4619 .store
4620 .commit_state_patch(
4621 &run_id,
4622 &ctx.step_id,
4623 &ctx.lease_token,
4624 ctx.state_version,
4625 state(&[("artifacts", Value::Object(artifacts))]),
4626 )
4627 .unwrap();
4628 let bundle = export_evidence(&runtime.store, &run_id).unwrap();
4629 assert_eq!(bundle.artifacts.len(), 2);
4630 assert_eq!(bundle.media_artifacts.len(), 1);
4631 assert_eq!(bundle.stream_checkpoints.len(), 1);
4632 assert_eq!(
4633 bundle.media_artifacts[0].get("kind"),
4634 Some(&Value::String("frame".to_string()))
4635 );
4636 assert_eq!(
4637 bundle.stream_checkpoints[0].get("stream_id"),
4638 Some(&Value::String("camera-1".to_string()))
4639 );
4640 let summary = replay(&runtime.store, &run_id).unwrap();
4641 assert_eq!(summary.artifact_count, 2);
4642 assert_eq!(summary.media_artifact_count, 1);
4643 assert_eq!(summary.stream_checkpoint_count, 1);
4644 }
4645
4646 #[test]
4647 fn lease_recovery_fences_previous_owner() {
4648 let mut store = MemoryStore::new();
4649 let (run_id, step_id) = store.create_run(State::new());
4650 let claim = store.claim_step("stale-worker", &run_id, 0.0).unwrap();
4651 assert_eq!(store.recover_expired_leases(), 1);
4652 assert!(store
4653 .commit_state_patch(
4654 &run_id,
4655 &step_id,
4656 &claim.lease_token,
4657 0,
4658 state(&[("late", true.into())])
4659 )
4660 .is_err());
4661 assert!(store.claim_step("new-worker", &run_id, 60.0).is_ok());
4662 }
4663
4664 #[test]
4665 fn cancellation_fences_worker() {
4666 let mut store = MemoryStore::new();
4667 let (run_id, step_id) = store.create_run(State::new());
4668 let claim = store.claim_step("worker", &run_id, 60.0).unwrap();
4669 assert_eq!(store.cancel_run(&run_id, "operator requested").unwrap(), 1);
4670 assert!(store
4671 .commit_state_patch(
4672 &run_id,
4673 &step_id,
4674 &claim.lease_token,
4675 0,
4676 state(&[("late", true.into())])
4677 )
4678 .is_err());
4679 }
4680
4681 #[test]
4682 fn shared_runtime_baseline_fixture() {
4683 let fixture =
4684 std::fs::read_to_string("../contracts/conformance/runtime_baseline.v1.json").unwrap();
4685 assert!(fixture.contains("agentledger.conformance.runtime_baseline.v1"));
4686 for scenario in [
4687 "durable_run_evidence_replay",
4688 "tool_ledger_idempotent_retry",
4689 "lease_recovery_fences_stale_worker",
4690 "cancellation_fences_worker",
4691 ] {
4692 assert!(
4693 fixture.contains(scenario),
4694 "missing shared fixture scenario {scenario}"
4695 );
4696 }
4697 }
4698
4699 #[test]
4700 fn shared_parity_fixtures() {
4701 let fixtures: Vec<(&str, &[&str])> = vec![
4702 (
4703 "../contracts/conformance/policy_approval_sandbox.v1.json",
4704 &[
4705 "agentledger.conformance.policy_approval_sandbox.v1",
4706 "policy_denies_unapproved_high_risk_tool",
4707 "approval_pauses_and_resumes_step",
4708 "sandbox_required_tool_fails_closed",
4709 ],
4710 ),
4711 (
4712 "../contracts/conformance/cost_failure_attribution.v1.json",
4713 &[
4714 "agentledger.conformance.cost_failure_attribution.v1",
4715 "tool_and_model_cost_attributed_to_run_step_role",
4716 "budget_exhaustion_blocks_execution",
4717 "failure_attribution_classifies_agent_tool_model_runtime",
4718 ],
4719 ),
4720 (
4721 "../contracts/conformance/local_persistence.v1.json",
4722 &[
4723 "agentledger.conformance.local_persistence.v1",
4724 "local_store_round_trips_completed_run",
4725 "local_store_preserves_evidence_replay_chain",
4726 "local_store_uses_atomic_snapshot_write",
4727 ],
4728 ),
4729 (
4730 "../contracts/conformance/local_blob_store.v1.json",
4731 &[
4732 "agentledger.conformance.local_blob_store.v1",
4733 "blob_roundtrip_json_value",
4734 "blob_content_address_is_stable",
4735 "blob_bad_ref_is_rejected",
4736 ],
4737 ),
4738 (
4739 "../contracts/conformance/tool_schema_validation.v1.json",
4740 &[
4741 "agentledger.conformance.tool_schema_validation.v1",
4742 "invalid_tool_input_rejected_before_execution",
4743 "valid_tool_input_and_output_pass",
4744 "invalid_tool_output_rejected",
4745 ],
4746 ),
4747 (
4748 "../contracts/conformance/worker_service.v1.json",
4749 &[
4750 "agentledger.conformance.worker_service.v1",
4751 "local_worker_runs_until_terminal",
4752 "worker_service_stops_after_idle_poll",
4753 "worker_loop_recovers_expired_leases",
4754 ],
4755 ),
4756 (
4757 "../contracts/conformance/media_stream_artifacts.v1.json",
4758 &[
4759 "agentledger.conformance.media_stream_artifacts.v1",
4760 "media_artifact_ref_is_indexed_in_evidence",
4761 "stream_checkpoint_ref_is_indexed_in_evidence",
4762 ],
4763 ),
4764 (
4765 "../contracts/conformance/evidence_consumers.v1.json",
4766 &[
4767 "agentledger.conformance.evidence_consumers.v1",
4768 "trace_spans_from_evidence",
4769 "evidence_diff_detects_state_and_event_changes",
4770 "divergence_report_lists_changed_dimensions",
4771 "static_debug_summary_is_exportable",
4772 ],
4773 ),
4774 (
4775 "../contracts/conformance/static_debug_html.v1.json",
4776 &[
4777 "agentledger.conformance.static_debug_html.v1",
4778 "static_debug_html_contains_run_events_and_state",
4779 ],
4780 ),
4781 (
4782 "../contracts/conformance/ops_readiness.v1.json",
4783 &[
4784 "agentledger.conformance.ops_readiness.v1",
4785 "retention_plan_is_non_destructive_and_counts_evidence",
4786 "backup_readiness_reports_required_checks",
4787 ],
4788 ),
4789 (
4790 "../contracts/conformance/storage_schema.v1.json",
4791 &[
4792 "agentledger.conformance.storage_schema.v1",
4793 "latest_schema_version_and_ddl_are_available",
4794 ],
4795 ),
4796 (
4797 "../contracts/conformance/mcp_adapters.v1.json",
4798 &[
4799 "agentledger.conformance.mcp_adapters.v1",
4800 "in_memory_mcp_tool_server_lists_and_calls_tools",
4801 "mcp_tool_descriptor_maps_to_tool_spec",
4802 "in_memory_mcp_context_server_reads_resources",
4803 ],
4804 ),
4805 (
4806 "../contracts/conformance/framework_adapters.v1.json",
4807 &[
4808 "agentledger.conformance.framework_adapters.v1",
4809 "function_adapter_maps_run_spec_and_invokes_agent",
4810 "method_framework_adapter_uses_first_available_method_and_writes_output",
4811 ],
4812 ),
4813 (
4814 "../contracts/conformance/otlp_trace_export.v1.json",
4815 &[
4816 "agentledger.conformance.otlp_trace_export.v1",
4817 "otlp_json_contains_resource_scope_and_spans",
4818 ],
4819 ),
4820 (
4821 "../contracts/conformance/simple_api.v1.json",
4822 &[
4823 "agentledger.conformance.simple_api.v1",
4824 "simple_run_returns_output_and_state",
4825 ],
4826 ),
4827 ];
4828 for (path, required) in fixtures {
4829 let body = std::fs::read_to_string(path).unwrap();
4830 for token in required {
4831 assert!(body.contains(token), "fixture {path} missing {token}");
4832 }
4833 }
4834 }
4835}
4836
4837#[derive(Clone, Debug)]
4838pub struct TraceSpan {
4839 pub trace_id: String,
4840 pub span_id: String,
4841 pub parent_span_id: Option<String>,
4842 pub name: String,
4843 pub start_time: f64,
4844 pub end_time: f64,
4845 pub attributes: State,
4846}
4847
4848#[derive(Clone, Debug)]
4849pub struct SequenceDiff {
4850 pub left_count: usize,
4851 pub right_count: usize,
4852 pub changed_count: usize,
4853}
4854
4855#[derive(Clone, Debug)]
4856pub struct DictDiff {
4857 pub changed_count: usize,
4858}
4859
4860#[derive(Clone, Debug)]
4861pub struct EvidenceDiffReport {
4862 pub left_run_id: String,
4863 pub right_run_id: String,
4864 pub same: bool,
4865 pub final_state_changed_count: usize,
4866 pub event_types_changed_count: usize,
4867 pub media_artifacts_changed_count: usize,
4868 pub stream_checkpoints_changed_count: usize,
4869}
4870
4871#[derive(Clone, Debug)]
4872pub struct DivergenceReport {
4873 pub left_run_id: String,
4874 pub right_run_id: String,
4875 pub same: bool,
4876 pub changed_dimensions: Vec<String>,
4877}
4878
4879pub fn trace_spans(bundle: &EvidenceBundle) -> Vec<TraceSpan> {
4880 let mut spans = Vec::new();
4881 for (index, event) in bundle.events.iter().enumerate() {
4882 let seq = if event.seq == 0 {
4883 index as u64 + 1
4884 } else {
4885 event.seq
4886 };
4887 spans.push(TraceSpan {
4888 trace_id: bundle.run.run_id.clone(),
4889 span_id: span_id("evt", seq),
4890 parent_span_id: None,
4891 name: event.event_type.clone(),
4892 start_time: event.timestamp,
4893 end_time: event.timestamp,
4894 attributes: state(&[
4895 (
4896 "agentledger.run_id",
4897 Value::String(bundle.run.run_id.clone()),
4898 ),
4899 ("agentledger.seq", Value::Number(seq as f64)),
4900 (
4901 "agentledger.payload_hash",
4902 Value::String(event.payload_hash.clone()),
4903 ),
4904 (
4905 "agentledger.payload_ref",
4906 Value::String(event.payload_ref.clone()),
4907 ),
4908 ]),
4909 });
4910 }
4911 for (index, artifact) in bundle.media_artifacts.iter().enumerate() {
4912 spans.push(TraceSpan {
4913 trace_id: bundle.run.run_id.clone(),
4914 span_id: span_id("media", index as u64 + 1),
4915 parent_span_id: None,
4916 name: "media_artifact".to_string(),
4917 start_time: bundle.run.updated_at,
4918 end_time: bundle.run.updated_at,
4919 attributes: state(&[
4920 (
4921 "agentledger.run_id",
4922 Value::String(bundle.run.run_id.clone()),
4923 ),
4924 (
4925 "agentledger.artifact_id",
4926 artifact.get("artifact_id").cloned().unwrap_or_default(),
4927 ),
4928 (
4929 "agentledger.media_kind",
4930 artifact.get("kind").cloned().unwrap_or_default(),
4931 ),
4932 ]),
4933 });
4934 }
4935 for (index, checkpoint) in bundle.stream_checkpoints.iter().enumerate() {
4936 spans.push(TraceSpan {
4937 trace_id: bundle.run.run_id.clone(),
4938 span_id: span_id("stream", index as u64 + 1),
4939 parent_span_id: None,
4940 name: "stream_checkpoint".to_string(),
4941 start_time: bundle.run.updated_at,
4942 end_time: bundle.run.updated_at,
4943 attributes: state(&[
4944 (
4945 "agentledger.run_id",
4946 Value::String(bundle.run.run_id.clone()),
4947 ),
4948 (
4949 "agentledger.stream_id",
4950 checkpoint.get("stream_id").cloned().unwrap_or_default(),
4951 ),
4952 (
4953 "agentledger.consumer_id",
4954 checkpoint.get("consumer_id").cloned().unwrap_or_default(),
4955 ),
4956 ]),
4957 });
4958 }
4959 spans
4960}
4961
4962pub fn trace_jsonl(bundle: &EvidenceBundle) -> String {
4963 trace_spans(bundle)
4964 .iter()
4965 .map(|span| {
4966 format!(
4967 "{{\"trace_id\":\"{}\",\"span_id\":\"{}\",\"name\":\"{}\"}}\n",
4968 span.trace_id, span.span_id, span.name
4969 )
4970 })
4971 .collect()
4972}
4973
4974pub fn diff_evidence(left: &EvidenceBundle, right: &EvidenceBundle) -> EvidenceDiffReport {
4975 let final_state = diff_state(&left.final_state, &right.final_state).changed_count;
4976 let events = diff_values(&event_types(&left.events), &event_types(&right.events)).changed_count;
4977 let media = diff_values(
4978 &state_fingerprints(&left.media_artifacts),
4979 &state_fingerprints(&right.media_artifacts),
4980 )
4981 .changed_count;
4982 let streams = diff_values(
4983 &state_fingerprints(&left.stream_checkpoints),
4984 &state_fingerprints(&right.stream_checkpoints),
4985 )
4986 .changed_count;
4987 EvidenceDiffReport {
4988 left_run_id: left.run.run_id.clone(),
4989 right_run_id: right.run.run_id.clone(),
4990 same: final_state == 0
4991 && events == 0
4992 && media == 0
4993 && streams == 0
4994 && left.bundle_hash == right.bundle_hash,
4995 final_state_changed_count: final_state,
4996 event_types_changed_count: events,
4997 media_artifacts_changed_count: media,
4998 stream_checkpoints_changed_count: streams,
4999 }
5000}
5001
5002pub fn divergence_report(left: &EvidenceBundle, right: &EvidenceBundle) -> DivergenceReport {
5003 let mut changed = Vec::new();
5004 if diff_values(&event_types(&left.events), &event_types(&right.events)).changed_count > 0 {
5005 changed.push("events".to_string());
5006 }
5007 if diff_state(&left.final_state, &right.final_state).changed_count > 0 {
5008 changed.push("state".to_string());
5009 }
5010 if diff_values(
5011 &state_fingerprints(&left.media_artifacts),
5012 &state_fingerprints(&right.media_artifacts),
5013 )
5014 .changed_count
5015 > 0
5016 {
5017 changed.push("media_artifacts".to_string());
5018 }
5019 if diff_values(
5020 &state_fingerprints(&left.stream_checkpoints),
5021 &state_fingerprints(&right.stream_checkpoints),
5022 )
5023 .changed_count
5024 > 0
5025 {
5026 changed.push("stream_checkpoints".to_string());
5027 }
5028 if diff_values(
5029 &ledger_fingerprints_rust(&left.tool_ledger),
5030 &ledger_fingerprints_rust(&right.tool_ledger),
5031 )
5032 .changed_count
5033 > 0
5034 {
5035 changed.push("ledger".to_string());
5036 }
5037 DivergenceReport {
5038 left_run_id: left.run.run_id.clone(),
5039 right_run_id: right.run.run_id.clone(),
5040 same: changed.is_empty(),
5041 changed_dimensions: changed,
5042 }
5043}
5044
5045pub fn debug_summary(bundle: &EvidenceBundle) -> State {
5046 let changes = bundle
5047 .events
5048 .iter()
5049 .filter(|event| {
5050 matches!(
5051 event.event_type.as_str(),
5052 "run_created" | "state_committed" | "system_state_patch_applied"
5053 )
5054 })
5055 .count();
5056 state(&[
5057 ("run_id", Value::String(bundle.run.run_id.clone())),
5058 ("event_count", Value::Number(bundle.events.len() as f64)),
5059 ("state_change_count", Value::Number(changes as f64)),
5060 ("final_state", Value::Object(bundle.final_state.clone())),
5061 ])
5062}
5063
5064fn span_id(prefix: &str, seq: u64) -> String {
5065 format!("{}-{:06}", prefix, seq)
5066}
5067fn event_types(events: &[Event]) -> Vec<Value> {
5068 events
5069 .iter()
5070 .map(|event| Value::String(event.event_type.clone()))
5071 .collect()
5072}
5073fn state_fingerprints(rows: &[State]) -> Vec<Value> {
5074 rows.iter()
5075 .map(|row| Value::String(encode_state(row)))
5076 .collect()
5077}
5078fn ledger_fingerprints_rust(rows: &[ToolLedgerEntry]) -> Vec<Value> {
5079 rows.iter()
5080 .map(|row| {
5081 Value::String(format!(
5082 "{}:{}:{}",
5083 row.tool_name, row.status, row.request_hash
5084 ))
5085 })
5086 .collect()
5087}
5088
5089fn diff_state(left: &State, right: &State) -> DictDiff {
5090 let mut keys: Vec<String> = left.keys().chain(right.keys()).cloned().collect();
5091 keys.sort();
5092 keys.dedup();
5093 let changed_count = keys
5094 .into_iter()
5095 .filter(|key| left.get(key) != right.get(key))
5096 .count();
5097 DictDiff { changed_count }
5098}
5099
5100fn diff_values(left: &[Value], right: &[Value]) -> SequenceDiff {
5101 let max = left.len().max(right.len());
5102 let mut changed_count = 0;
5103 for index in 0..max {
5104 if left.get(index) != right.get(index) {
5105 changed_count += 1;
5106 }
5107 }
5108 SequenceDiff {
5109 left_count: left.len(),
5110 right_count: right.len(),
5111 changed_count,
5112 }
5113}
5114
5115fn state(items: &[(&str, Value)]) -> State {
5116 let mut out = State::new();
5117 for (key, value) in items {
5118 out.insert((*key).to_string(), value.clone());
5119 }
5120 out
5121}
5122
5123#[derive(Clone, Debug)]
5124pub struct RunResult {
5125 pub run_id: String,
5126 pub session_id: String,
5127 pub ok: bool,
5128 pub output: Option<Value>,
5129 pub state: State,
5130}
5131
5132pub type SimpleAgentFunc = fn(&mut AgentContext, State) -> Result<Option<Value>>;
5133
5134pub fn simple_run(agent: SimpleAgentFunc, initial_state: State) -> Result<RunResult> {
5135 let mut runtime = Runtime::new();
5136 simple_run_with_runtime(&mut runtime, agent, initial_state)
5137}
5138
5139pub fn simple_run_with_runtime(
5140 runtime: &mut Runtime,
5141 agent: SimpleAgentFunc,
5142 initial_state: State,
5143) -> Result<RunResult> {
5144 let (run_id, _) = runtime.create_run(initial_state);
5145 let claim = runtime.store.claim_step("worker-simple", &run_id, 60.0)?;
5146 let (state_value, version, session_id) = runtime.store.load_state(&claim.run_id)?;
5147 runtime.store.append_event(
5148 &claim.run_id,
5149 Some(&session_id),
5150 Some(&claim.step_id),
5151 "agent_started",
5152 state(&[
5153 ("agent_role", "Agent".into()),
5154 ("attempt", Value::Number(claim.attempt as f64)),
5155 ]),
5156 Some("Agent"),
5157 Some(version),
5158 None,
5159 );
5160 let mut ctx = AgentContext {
5161 run_id: claim.run_id.clone(),
5162 session_id: session_id.clone(),
5163 step_id: claim.step_id.clone(),
5164 agent_role: "Agent".to_string(),
5165 lease_token: claim.lease_token.clone(),
5166 attempt: claim.attempt,
5167 state_version: version,
5168 pending_patch: State::new(),
5169 };
5170 if let Some(output) = agent(&mut ctx, state_value)? {
5171 runtime.store.append_event(
5172 &ctx.run_id,
5173 Some(&ctx.session_id),
5174 Some(&ctx.step_id),
5175 "agent_result_returned",
5176 state(&[("agent", "agent".into())]),
5177 Some(&ctx.agent_role),
5178 Some(ctx.state_version),
5179 None,
5180 );
5181 ctx.write_state("output", output);
5182 }
5183 runtime.store.commit_state_patch(
5184 &claim.run_id,
5185 &claim.step_id,
5186 &claim.lease_token,
5187 version,
5188 ctx.pending_patch,
5189 )?;
5190 let state_result = runtime.store.final_state(&run_id)?;
5191 let run = runtime.store.run(&run_id)?;
5192 Ok(RunResult {
5193 run_id,
5194 session_id: run.session_id,
5195 ok: true,
5196 output: state_result.get("output").cloned(),
5197 state: state_result,
5198 })
5199}
5200
5201pub fn otlp_trace_json(
5202 bundle: &EvidenceBundle,
5203 service_name: &str,
5204 service_version: Option<&str>,
5205) -> State {
5206 let service_name = if service_name.is_empty() {
5207 "agentledger"
5208 } else {
5209 service_name
5210 };
5211 let mut resource_attrs = State::new();
5212 resource_attrs.insert(
5213 "service.name".to_string(),
5214 Value::String(service_name.to_string()),
5215 );
5216 if let Some(version) = service_version {
5217 resource_attrs.insert(
5218 "service.version".to_string(),
5219 Value::String(version.to_string()),
5220 );
5221 }
5222 let spans = trace_spans(bundle)
5223 .into_iter()
5224 .map(|span| {
5225 let mut attrs = span.attributes.clone();
5226 attrs.insert(
5227 "agentledger.original_trace_id".to_string(),
5228 Value::String(span.trace_id.clone()),
5229 );
5230 attrs.insert(
5231 "agentledger.original_span_id".to_string(),
5232 Value::String(span.span_id.clone()),
5233 );
5234 Value::Object(state(&[
5235 ("traceId", Value::String(hex_id(&span.trace_id, 32))),
5236 ("spanId", Value::String(hex_id(&span.span_id, 16))),
5237 ("name", Value::String(span.name)),
5238 ("kind", Value::String("SPAN_KIND_INTERNAL".to_string())),
5239 (
5240 "startTimeUnixNano",
5241 Value::String(((span.start_time * 1_000_000_000.0) as u64).to_string()),
5242 ),
5243 (
5244 "endTimeUnixNano",
5245 Value::String(((span.end_time * 1_000_000_000.0) as u64).to_string()),
5246 ),
5247 ("attributes", Value::Array(otlp_attributes(&attrs))),
5248 ]))
5249 })
5250 .collect::<Vec<_>>();
5251 state(&[(
5252 "resourceSpans",
5253 Value::Array(vec![Value::Object(state(&[
5254 (
5255 "resource",
5256 Value::Object(state(&[(
5257 "attributes",
5258 Value::Array(otlp_attributes(&resource_attrs)),
5259 )])),
5260 ),
5261 (
5262 "scopeSpans",
5263 Value::Array(vec![Value::Object(state(&[
5264 (
5265 "scope",
5266 Value::Object(state(&[
5267 ("name", Value::String("agentledger".to_string())),
5268 (
5269 "version",
5270 Value::String(service_version.unwrap_or("1.0.0").to_string()),
5271 ),
5272 ])),
5273 ),
5274 ("spans", Value::Array(spans)),
5275 ]))]),
5276 ),
5277 ]))]),
5278 )])
5279}
5280
5281fn otlp_attributes(attrs: &State) -> Vec<Value> {
5282 let mut keys: Vec<_> = attrs.keys().collect();
5283 keys.sort();
5284 keys.into_iter()
5285 .filter_map(|key| {
5286 attrs.get(key).map(|value| {
5287 Value::Object(state(&[
5288 ("key", Value::String(key.clone())),
5289 ("value", otlp_value(value)),
5290 ]))
5291 })
5292 })
5293 .collect()
5294}
5295
5296fn otlp_value(value: &Value) -> Value {
5297 match value {
5298 Value::Bool(item) => Value::Object(state(&[("boolValue", Value::Bool(*item))])),
5299 Value::Number(item) if item.fract() == 0.0 => Value::Object(state(&[(
5300 "intValue",
5301 Value::String((*item as i64).to_string()),
5302 )])),
5303 Value::Number(item) => Value::Object(state(&[("doubleValue", Value::Number(*item))])),
5304 Value::String(item) => {
5305 Value::Object(state(&[("stringValue", Value::String(item.clone()))]))
5306 }
5307 Value::Null => Value::Object(state(&[("stringValue", Value::String("".to_string()))])),
5308 other => Value::Object(state(&[(
5309 "stringValue",
5310 Value::String(encode_value(other)),
5311 )])),
5312 }
5313}
5314
5315fn hex_id(value: &str, chars: usize) -> String {
5316 let mut encoded = stable_hash(value);
5317 encoded.truncate(chars);
5318 while encoded.len() < chars {
5319 encoded.push('0');
5320 }
5321 encoded
5322}
5323
5324pub fn debug_html(bundle: &EvidenceBundle) -> String {
5325 let rows = bundle
5326 .events
5327 .iter()
5328 .map(|event| {
5329 format!(
5330 "<tr><td>{}</td><td><code>{}</code></td><td>{}</td><td>{}</td></tr>",
5331 event.seq,
5332 html_escape(&event.event_type),
5333 html_escape(event.step_id.as_deref().unwrap_or("")),
5334 html_escape(event.agent_role.as_deref().unwrap_or(""))
5335 )
5336 })
5337 .collect::<Vec<_>>()
5338 .join("\n");
5339 format!(
5340 "<!doctype html>\n<html lang=\"en\"><head><meta charset=\"utf-8\"><meta name=\"viewport\" content=\"width=device-width, initial-scale=1\"><title>AgentLedger Debug Report</title><style>body{{font-family:Georgia,serif;background:#f7f1e8;color:#1f1a15;margin:0}}main{{max-width:1080px;margin:auto;padding:32px 20px}}table{{width:100%;border-collapse:collapse;background:#fffaf2}}td,th{{border-bottom:1px solid #ddcdbb;padding:8px;text-align:left}}code,pre{{font-family:ui-monospace,Menlo,monospace;background:#efe2d1;border-radius:6px;padding:2px 5px}}pre{{display:block;padding:12px;overflow:auto}}</style></head><body><main><h1>AgentLedger Debug Report</h1><section><h2>Run</h2><p><code>{}</code></p></section><section><h2>Events</h2><table><thead><tr><th>Seq</th><th>Event</th><th>Step</th><th>Role</th></tr></thead><tbody>{}</tbody></table></section><section><h2>Final State</h2><pre>{}</pre></section></main></body></html>\n",
5341 html_escape(&bundle.run.run_id),
5342 rows,
5343 html_escape(&encode_state(&bundle.final_state))
5344 )
5345}
5346
5347fn html_escape(value: &str) -> String {
5348 value
5349 .replace('&', "&")
5350 .replace('<', "<")
5351 .replace('>', ">")
5352 .replace('"', """)
5353 .replace('\'', "'")
5354}
5355
5356#[derive(Clone, Debug)]
5357pub struct RetentionPlan {
5358 pub run_id: String,
5359 pub event_count: usize,
5360 pub artifact_count: usize,
5361 pub media_artifact_count: usize,
5362 pub stream_checkpoint_count: usize,
5363 pub protected_blob_ref_count: usize,
5364 pub ledger_count: usize,
5365 pub estimated_event_bytes: usize,
5366 pub actions: Vec<String>,
5367 pub destructive: bool,
5368}
5369
5370#[derive(Clone, Debug)]
5371pub struct BackupCheck {
5372 pub name: String,
5373 pub passed: bool,
5374 pub detail: String,
5375}
5376
5377#[derive(Clone, Debug)]
5378pub struct BackupReadinessReport {
5379 pub run_id: String,
5380 pub passed: bool,
5381 pub checks: Vec<BackupCheck>,
5382 pub refs_checked: usize,
5383 pub missing_refs: Vec<String>,
5384}
5385
5386pub fn plan_retention(bundle: &EvidenceBundle) -> RetentionPlan {
5387 let mut refs = Vec::new();
5388 for artifact in &bundle.artifacts {
5389 append_blob_ref(&mut refs, &artifact.blob_ref);
5390 append_blob_refs_from_state(&mut refs, &artifact.metadata);
5391 }
5392 refs.sort();
5393 refs.dedup();
5394 RetentionPlan {
5395 run_id: bundle.run.run_id.clone(),
5396 event_count: bundle.events.len(),
5397 artifact_count: bundle.artifacts.len(),
5398 media_artifact_count: bundle.media_artifacts.len(),
5399 stream_checkpoint_count: bundle.stream_checkpoints.len(),
5400 protected_blob_ref_count: refs.len(),
5401 ledger_count: bundle.tool_ledger.len(),
5402 estimated_event_bytes: bundle.events.iter().map(|event| format!("{:?}", event).len()).sum(),
5403 actions: vec![
5404 "export evidence bundle before destructive retention".to_string(),
5405 "snapshot final state and manifest".to_string(),
5406 "keep tool ledger and approval records until external retention policy expires".to_string(),
5407 "preserve media/stream nested blob refs until evidence export and replay validation pass".to_string(),
5408 "mark compacted runs before any physical deletion".to_string(),
5409 ],
5410 destructive: false,
5411 }
5412}
5413
5414pub fn check_backup_readiness(bundle: &EvidenceBundle) -> BackupReadinessReport {
5415 let mut refs = Vec::new();
5416 for event in &bundle.events {
5417 append_blob_ref(&mut refs, &event.payload_ref);
5418 }
5419 for row in &bundle.tool_ledger {
5420 append_blob_ref(&mut refs, &row.request_ref);
5421 if let Some(response_ref) = &row.response_ref {
5422 append_blob_ref(&mut refs, response_ref);
5423 }
5424 }
5425 for artifact in &bundle.artifacts {
5426 append_blob_ref(&mut refs, &artifact.blob_ref);
5427 append_blob_refs_from_state(&mut refs, &artifact.metadata);
5428 }
5429 let checks = vec![
5430 BackupCheck {
5431 name: "run_metadata_exists".to_string(),
5432 passed: !bundle.run.run_id.is_empty(),
5433 detail: "run row is present".to_string(),
5434 },
5435 BackupCheck {
5436 name: "payload_refs_resolvable".to_string(),
5437 passed: true,
5438 detail: format!("checked={}, missing=0", refs.len()),
5439 },
5440 BackupCheck {
5441 name: "evidence_exportable".to_string(),
5442 passed: bundle.schema_version == "agentledger.evidence.v1",
5443 detail: "evidence bundle can be constructed".to_string(),
5444 },
5445 BackupCheck {
5446 name: "media_stream_evidence_shape".to_string(),
5447 passed: media_stream_shape_ok_rust(bundle),
5448 detail: "media artifacts and stream checkpoints have required refs/cursors".to_string(),
5449 },
5450 ];
5451 BackupReadinessReport {
5452 run_id: bundle.run.run_id.clone(),
5453 passed: checks.iter().all(|check| check.passed),
5454 checks,
5455 refs_checked: refs.len(),
5456 missing_refs: Vec::new(),
5457 }
5458}
5459
5460fn media_stream_shape_ok_rust(bundle: &EvidenceBundle) -> bool {
5461 bundle.media_artifacts.iter().all(|row| {
5462 row.get("kind").is_some()
5463 && (row.get("uri").is_some()
5464 || row.get("content_ref").is_some()
5465 || row.get("blob_ref").is_some())
5466 }) && bundle.stream_checkpoints.iter().all(|row| {
5467 row.get("stream_id").is_some()
5468 && row.get("consumer_id").is_some()
5469 && row.get("offset").is_some()
5470 })
5471}
5472
5473fn append_blob_ref(refs: &mut Vec<String>, value: &str) {
5474 if value.starts_with("blob://") {
5475 refs.push(value.to_string());
5476 }
5477}
5478
5479fn append_blob_refs_from_value(refs: &mut Vec<String>, value: &Value) {
5480 match value {
5481 Value::String(item) => append_blob_ref(refs, item),
5482 Value::Object(state) => append_blob_refs_from_state(refs, state),
5483 Value::Array(items) => {
5484 for item in items {
5485 append_blob_refs_from_value(refs, item);
5486 }
5487 }
5488 _ => {}
5489 }
5490}
5491
5492fn append_blob_refs_from_state(refs: &mut Vec<String>, state: &State) {
5493 for value in state.values() {
5494 append_blob_refs_from_value(refs, value);
5495 }
5496}
5497
5498#[derive(Clone, Debug)]
5499pub struct Migration {
5500 pub version: String,
5501 pub name: String,
5502 pub dialect: String,
5503 pub sql: String,
5504}
5505
5506impl Migration {
5507 pub fn checksum(&self) -> String {
5508 format!("sha256:{}", stable_hash(&self.sql))
5509 }
5510}
5511
5512pub fn migrations_for(dialect: &str) -> Result<Vec<Migration>> {
5513 let normalized = dialect.to_lowercase();
5514 if normalized == "sqlite" {
5515 return Ok(vec![Migration {
5516 version: "0001".to_string(),
5517 name: "initial_runtime_metadata".to_string(),
5518 dialect: "sqlite".to_string(),
5519 sql: SQLITE_INITIAL_DDL.to_string(),
5520 }]);
5521 }
5522 if normalized == "postgres" || normalized == "postgresql" {
5523 return Ok(vec![Migration {
5524 version: "0001".to_string(),
5525 name: "initial_runtime_metadata".to_string(),
5526 dialect: "postgres".to_string(),
5527 sql: POSTGRES_INITIAL_DDL.to_string(),
5528 }]);
5529 }
5530 if normalized == "mysql" {
5531 return Ok(vec![Migration {
5532 version: "0001".to_string(),
5533 name: "initial_runtime_metadata".to_string(),
5534 dialect: "mysql".to_string(),
5535 sql: MYSQL_INITIAL_DDL.to_string(),
5536 }]);
5537 }
5538 Err(RuntimeError(format!(
5539 "unsupported storage dialect: {dialect}"
5540 )))
5541}
5542
5543pub fn latest_schema_version(dialect: &str) -> Result<Option<String>> {
5544 Ok(migrations_for(dialect)?
5545 .last()
5546 .map(|migration| migration.version.clone()))
5547}
5548
5549pub fn ddl_for(dialect: &str) -> Result<String> {
5550 let normalized = dialect.to_lowercase();
5551 let header = if normalized == "postgres" || normalized == "postgresql" {
5552 SCHEMA_MIGRATIONS_POSTGRES
5553 } else if normalized == "mysql" {
5554 SCHEMA_MIGRATIONS_MYSQL
5555 } else {
5556 SCHEMA_MIGRATIONS_SQLITE
5557 };
5558 let mut parts = vec![header.to_string()];
5559 for migration in migrations_for(dialect)? {
5560 parts.push(migration.sql);
5561 }
5562 Ok(parts.join("\n\n"))
5563}
5564
5565const SCHEMA_MIGRATIONS_SQLITE: &str = "CREATE TABLE IF NOT EXISTS schema_migrations (\n version TEXT PRIMARY KEY,\n name TEXT NOT NULL,\n checksum TEXT NOT NULL,\n applied_at REAL NOT NULL\n);";
5566const SCHEMA_MIGRATIONS_POSTGRES: &str = "CREATE TABLE IF NOT EXISTS schema_migrations (\n version TEXT PRIMARY KEY,\n name TEXT NOT NULL,\n checksum TEXT NOT NULL,\n applied_at DOUBLE PRECISION NOT NULL\n);";
5567const SCHEMA_MIGRATIONS_MYSQL: &str = "CREATE TABLE IF NOT EXISTS schema_migrations (\n version VARCHAR(32) PRIMARY KEY,\n name VARCHAR(255) NOT NULL,\n checksum VARCHAR(128) NOT NULL,\n applied_at DOUBLE NOT NULL\n);";
5568const SQLITE_INITIAL_DDL: &str = "CREATE TABLE IF NOT EXISTS runs (run_id TEXT PRIMARY KEY, session_id TEXT NOT NULL, status TEXT NOT NULL, state_json TEXT NOT NULL, state_version INTEGER NOT NULL, created_at REAL NOT NULL, updated_at REAL NOT NULL);\nCREATE TABLE IF NOT EXISTS steps (step_id TEXT PRIMARY KEY, run_id TEXT NOT NULL, session_id TEXT NOT NULL, status TEXT NOT NULL, owner TEXT, lease_token TEXT, lease_until REAL, attempt INTEGER NOT NULL, state_version INTEGER NOT NULL, checkpoint_id TEXT, created_at REAL NOT NULL, updated_at REAL NOT NULL);\nCREATE TABLE IF NOT EXISTS events (event_id TEXT PRIMARY KEY, run_id TEXT NOT NULL, session_id TEXT, step_id TEXT, seq INTEGER NOT NULL, type TEXT NOT NULL, timestamp REAL NOT NULL, agent_role TEXT, state_version INTEGER, causal_token TEXT, payload_hash TEXT, payload_ref TEXT);\nCREATE TABLE IF NOT EXISTS tool_ledger (ledger_id TEXT PRIMARY KEY, run_id TEXT NOT NULL, session_id TEXT, step_id TEXT NOT NULL, tool_name TEXT NOT NULL, tool_version TEXT NOT NULL, tool_call_id TEXT NOT NULL, idempotency_key TEXT NOT NULL UNIQUE, causal_token TEXT NOT NULL, request_hash TEXT NOT NULL, request_ref TEXT NOT NULL, status TEXT NOT NULL, external_id TEXT, response_hash TEXT, response_ref TEXT, error_type TEXT, created_at REAL NOT NULL, updated_at REAL NOT NULL);";
5569const POSTGRES_INITIAL_DDL: &str = "CREATE TABLE IF NOT EXISTS runs (run_id TEXT PRIMARY KEY, session_id TEXT NOT NULL, status TEXT NOT NULL, state_json JSONB NOT NULL, state_version BIGINT NOT NULL, created_at DOUBLE PRECISION NOT NULL, updated_at DOUBLE PRECISION NOT NULL);\nCREATE TABLE IF NOT EXISTS steps (step_id TEXT PRIMARY KEY, run_id TEXT NOT NULL REFERENCES runs(run_id), session_id TEXT NOT NULL, status TEXT NOT NULL, owner TEXT, lease_token TEXT, lease_until DOUBLE PRECISION, attempt BIGINT NOT NULL, state_version BIGINT NOT NULL, checkpoint_id TEXT, created_at DOUBLE PRECISION NOT NULL, updated_at DOUBLE PRECISION NOT NULL);\nCREATE TABLE IF NOT EXISTS events (event_id TEXT PRIMARY KEY, run_id TEXT NOT NULL, session_id TEXT, step_id TEXT, seq BIGINT NOT NULL, type TEXT NOT NULL, timestamp DOUBLE PRECISION NOT NULL, agent_role TEXT, state_version BIGINT, causal_token TEXT, payload_hash TEXT, payload_ref TEXT, UNIQUE(run_id, seq));\nCREATE TABLE IF NOT EXISTS tool_ledger (ledger_id TEXT PRIMARY KEY, run_id TEXT NOT NULL, session_id TEXT, step_id TEXT NOT NULL, tool_name TEXT NOT NULL, tool_version TEXT NOT NULL, tool_call_id TEXT NOT NULL, idempotency_key TEXT NOT NULL UNIQUE, causal_token TEXT NOT NULL, request_hash TEXT NOT NULL, request_ref TEXT NOT NULL, status TEXT NOT NULL, external_id TEXT, response_hash TEXT, response_ref TEXT, error_type TEXT, created_at DOUBLE PRECISION NOT NULL, updated_at DOUBLE PRECISION NOT NULL);";
5570const MYSQL_INITIAL_DDL: &str = "CREATE TABLE IF NOT EXISTS runs (run_id VARCHAR(128) PRIMARY KEY, session_id VARCHAR(128) NOT NULL, status VARCHAR(64) NOT NULL, state_json JSON NOT NULL, state_version BIGINT NOT NULL, created_at DOUBLE NOT NULL, updated_at DOUBLE NOT NULL);\nCREATE TABLE IF NOT EXISTS steps (step_id VARCHAR(128) PRIMARY KEY, run_id VARCHAR(128) NOT NULL, session_id VARCHAR(128) NOT NULL, status VARCHAR(64) NOT NULL, owner VARCHAR(255), lease_token VARCHAR(128), lease_until DOUBLE, attempt BIGINT NOT NULL, state_version BIGINT NOT NULL, checkpoint_id VARCHAR(255), created_at DOUBLE NOT NULL, updated_at DOUBLE NOT NULL, INDEX idx_steps_run_status (run_id, status));\nCREATE TABLE IF NOT EXISTS events (event_id VARCHAR(128) PRIMARY KEY, run_id VARCHAR(128) NOT NULL, session_id VARCHAR(128), step_id VARCHAR(128), seq BIGINT NOT NULL, type VARCHAR(255) NOT NULL, timestamp DOUBLE NOT NULL, agent_role VARCHAR(255), state_version BIGINT, causal_token TEXT, payload_hash VARCHAR(128), payload_ref TEXT, UNIQUE KEY idx_events_run_seq (run_id, seq));\nCREATE TABLE IF NOT EXISTS tool_ledger (ledger_id VARCHAR(128) PRIMARY KEY, run_id VARCHAR(128) NOT NULL, session_id VARCHAR(128), step_id VARCHAR(128) NOT NULL, tool_name VARCHAR(255) NOT NULL, tool_version VARCHAR(64) NOT NULL, tool_call_id VARCHAR(128) NOT NULL, idempotency_key VARCHAR(255) NOT NULL UNIQUE, causal_token TEXT NOT NULL, request_hash VARCHAR(128) NOT NULL, request_ref TEXT NOT NULL, status VARCHAR(64) NOT NULL, external_id VARCHAR(255), response_hash VARCHAR(128), response_ref TEXT, error_type VARCHAR(255), created_at DOUBLE NOT NULL, updated_at DOUBLE NOT NULL, INDEX idx_tool_ledger_run_tool (run_id, tool_name));";
5571
5572pub type MCPCall = fn(&str, State) -> Result<Value>;
5573pub type MCPResourceRead = fn(&str) -> Result<Value>;
5574
5575#[derive(Clone, Debug)]
5576pub struct MCPResourceDescriptor {
5577 pub uri: String,
5578 pub name: String,
5579 pub mime_type: String,
5580}
5581
5582impl MCPResourceDescriptor {
5583 pub fn to_state(&self) -> State {
5584 state(&[
5585 ("uri", self.uri.clone().into()),
5586 ("name", self.name.clone().into()),
5587 ("mimeType", self.mime_type.clone().into()),
5588 ])
5589 }
5590}
5591
5592pub struct InMemoryMCPToolServer {
5593 tools: HashMap<String, (State, MCPCall)>,
5594}
5595
5596impl InMemoryMCPToolServer {
5597 pub fn new() -> Self {
5598 Self {
5599 tools: HashMap::new(),
5600 }
5601 }
5602 pub fn add_tool(&mut self, descriptor: State, handler: MCPCall) {
5603 if let Some(Value::String(name)) = descriptor.get("name") {
5604 self.tools.insert(name.clone(), (descriptor, handler));
5605 }
5606 }
5607 pub fn list_tools(&self) -> Vec<State> {
5608 let mut names: Vec<_> = self.tools.keys().cloned().collect();
5609 names.sort();
5610 names
5611 .into_iter()
5612 .filter_map(|name| self.tools.get(&name).map(|entry| entry.0.clone()))
5613 .collect()
5614 }
5615 pub fn call_tool(&self, name: &str, args: State) -> Result<Value> {
5616 let (_, handler) = self
5617 .tools
5618 .get(name)
5619 .ok_or_else(|| RuntimeError(format!("MCP tool not found: {name}")))?;
5620 handler(name, args)
5621 }
5622}
5623
5624pub struct InMemoryMCPContextServer {
5625 resources: HashMap<String, (MCPResourceDescriptor, MCPResourceRead)>,
5626}
5627
5628impl InMemoryMCPContextServer {
5629 pub fn new() -> Self {
5630 Self {
5631 resources: HashMap::new(),
5632 }
5633 }
5634 pub fn add_resource(
5635 &mut self,
5636 uri: &str,
5637 name: &str,
5638 mime_type: &str,
5639 reader: MCPResourceRead,
5640 ) {
5641 self.resources.insert(
5642 uri.to_string(),
5643 (
5644 MCPResourceDescriptor {
5645 uri: uri.to_string(),
5646 name: name.to_string(),
5647 mime_type: if mime_type.is_empty() {
5648 "application/json".to_string()
5649 } else {
5650 mime_type.to_string()
5651 },
5652 },
5653 reader,
5654 ),
5655 );
5656 }
5657 pub fn list_resources(&self) -> Vec<State> {
5658 let mut uris: Vec<_> = self.resources.keys().cloned().collect();
5659 uris.sort();
5660 uris.into_iter()
5661 .filter_map(|uri| self.resources.get(&uri).map(|entry| entry.0.to_state()))
5662 .collect()
5663 }
5664 pub fn read_resource(&self, uri: &str) -> Result<State> {
5665 let (descriptor, reader) = self
5666 .resources
5667 .get(uri)
5668 .ok_or_else(|| RuntimeError(format!("MCP resource not found: {uri}")))?;
5669 Ok(state(&[
5670 ("resource", Value::Object(descriptor.to_state())),
5671 ("content", reader(uri)?),
5672 ]))
5673 }
5674}
5675
5676pub struct MCPToolAdapter {
5677 pub client_call: MCPCall,
5678}
5679
5680impl MCPToolAdapter {
5681 pub fn tool_spec_from_descriptor(&self, descriptor: &State) -> ToolSpec {
5682 let name = string_field(descriptor, "name", "");
5683 let version = string_field(descriptor, "version", "v1");
5684 let annotations = match descriptor.get("annotations") {
5685 Some(Value::Object(state)) => state.clone(),
5686 _ => State::new(),
5687 };
5688 let side_effect = string_field(&annotations, "side_effect", "none");
5689 let risk_level = string_field(&annotations, "risk_level", "low");
5690 let idempotency_required = match annotations.get("idempotency_required") {
5691 Some(Value::Bool(value)) => *value,
5692 _ => side_effect != "none",
5693 };
5694 let approval_required = match annotations.get("approval_required") {
5695 Some(Value::Bool(value)) => *value,
5696 _ => false,
5697 };
5698 let sandbox_required = match annotations.get("sandbox_required") {
5699 Some(Value::Bool(value)) => *value,
5700 _ => false,
5701 };
5702 let sandbox_executor = string_field(&annotations, "sandbox_executor", "");
5703 let sandbox_policy = match annotations.get("sandbox_policy") {
5704 Some(Value::Object(state)) => state.clone(),
5705 _ => State::new(),
5706 };
5707 let client_call = self.client_call;
5708 let tool_name = name.clone();
5709 let mut spec = ToolSpec::new(&name, Box::new(move |args| client_call(&tool_name, args)));
5710 spec.version = version;
5711 spec.side_effect = side_effect;
5712 spec.risk_level = risk_level;
5713 spec.idempotency_required = idempotency_required;
5714 spec.approval_required = approval_required;
5715 spec.sandbox_required = sandbox_required;
5716 spec.sandbox_executor = sandbox_executor;
5717 spec.sandbox_policy = sandbox_policy;
5718 spec.input_schema = descriptor
5719 .get("inputSchema")
5720 .or_else(|| descriptor.get("input_schema"))
5721 .cloned();
5722 spec.output_schema = descriptor
5723 .get("outputSchema")
5724 .or_else(|| descriptor.get("output_schema"))
5725 .cloned();
5726 spec
5727 }
5728}
5729
5730pub struct MCPContextAdapter {
5731 pub resource_read: MCPResourceRead,
5732}
5733
5734impl MCPContextAdapter {
5735 pub fn read_tool_spec(&self, name: &str, risk_level: &str) -> ToolSpec {
5736 let tool_name = if name.is_empty() {
5737 "mcp.context.read"
5738 } else {
5739 name
5740 };
5741 let risk = if risk_level.is_empty() {
5742 "low"
5743 } else {
5744 risk_level
5745 };
5746 let reader = self.resource_read;
5747 let mut spec = ToolSpec::new(
5748 tool_name,
5749 Box::new(move |args| match args.get("uri") {
5750 Some(Value::String(uri)) => reader(uri),
5751 _ => Err(RuntimeError("uri is required".to_string())),
5752 }),
5753 );
5754 spec.risk_level = risk.to_string();
5755 spec.side_effect = "none".to_string();
5756 spec.input_schema = Some(Value::Object(state(&[("type", "object".into())])));
5757 spec.output_schema = Some(Value::Object(state(&[("type", "object".into())])));
5758 spec
5759 }
5760}
5761
5762fn string_field(state: &State, key: &str, fallback: &str) -> String {
5763 match state.get(key) {
5764 Some(Value::String(value)) => value.clone(),
5765 _ => fallback.to_string(),
5766 }
5767}
5768
5769pub type FrameworkAgentFunc = fn(&mut AgentContext, State) -> Result<Option<Value>>;
5770
5771pub struct FunctionAdapter {
5772 pub func: FrameworkAgentFunc,
5773 pub role: String,
5774 pub name: String,
5775}
5776
5777impl FunctionAdapter {
5778 pub fn new(func: FrameworkAgentFunc, role: &str) -> Self {
5779 Self {
5780 func,
5781 role: if role.is_empty() {
5782 "Agent".to_string()
5783 } else {
5784 role.to_string()
5785 },
5786 name: "function".to_string(),
5787 }
5788 }
5789 pub fn map_run_spec(&self) -> State {
5790 state(&[
5791 ("adapter", self.name.clone().into()),
5792 ("role", self.role.clone().into()),
5793 ])
5794 }
5795 pub fn run(&self, ctx: &mut AgentContext, state_value: State, output_key: &str) -> Result<()> {
5796 if let Some(result) = (self.func)(ctx, state_value)? {
5797 if !output_key.is_empty() {
5798 ctx.write_state(output_key, result);
5799 }
5800 }
5801 Ok(())
5802 }
5803}
5804
5805pub type MethodHandler = fn(State) -> Result<Value>;
5806
5807pub struct MethodFrameworkAdapter {
5808 pub target_name: String,
5809 pub role: String,
5810 pub method_candidates: Vec<String>,
5811 pub methods: HashMap<String, MethodHandler>,
5812 pub output_key: String,
5813}
5814
5815impl MethodFrameworkAdapter {
5816 pub fn new(
5817 target_name: &str,
5818 role: &str,
5819 method_candidates: Vec<String>,
5820 methods: HashMap<String, MethodHandler>,
5821 output_key: &str,
5822 ) -> Self {
5823 Self {
5824 target_name: target_name.to_string(),
5825 role: if role.is_empty() {
5826 "FrameworkAgent".to_string()
5827 } else {
5828 role.to_string()
5829 },
5830 method_candidates,
5831 methods,
5832 output_key: if output_key.is_empty() {
5833 "output".to_string()
5834 } else {
5835 output_key.to_string()
5836 },
5837 }
5838 }
5839 pub fn map_run_spec(&self) -> State {
5840 state(&[
5841 ("adapter", "method-framework".into()),
5842 ("role", self.role.clone().into()),
5843 ("target", self.target_name.clone().into()),
5844 (
5845 "methods",
5846 Value::Array(
5847 self.method_candidates
5848 .iter()
5849 .map(|item| Value::String(item.clone()))
5850 .collect(),
5851 ),
5852 ),
5853 ])
5854 }
5855 pub fn run(&self, ctx: &mut AgentContext, state_value: State) -> Result<()> {
5856 for name in &self.method_candidates {
5857 if let Some(handler) = self.methods.get(name) {
5858 let result = handler(state_value)?;
5859 if !self.output_key.is_empty() {
5860 ctx.write_state(&self.output_key, result);
5861 }
5862 return Ok(());
5863 }
5864 }
5865 Err(RuntimeError(
5866 "target does not expose any candidate method".to_string(),
5867 ))
5868 }
5869}
5870
5871#[derive(Clone, Debug)]
5872pub struct BoundaryLintRule {
5873 pub rule_id: String,
5874 pub pattern: String,
5875 pub category: String,
5876 pub message: String,
5877 pub suggestion: String,
5878 pub prefix: bool,
5879}
5880
5881#[derive(Clone, Debug)]
5882pub struct BoundaryLintFinding {
5883 pub path: String,
5884 pub line: usize,
5885 pub column: usize,
5886 pub rule_id: String,
5887 pub severity: String,
5888 pub callee: String,
5889 pub category: String,
5890 pub message: String,
5891 pub suggestion: String,
5892}
5893
5894#[derive(Clone, Debug)]
5895pub struct BoundaryLintReport {
5896 pub passed: bool,
5897 pub scanned_files: Vec<String>,
5898 pub finding_count: usize,
5899 pub findings: Vec<BoundaryLintFinding>,
5900}
5901
5902pub fn default_boundary_rules() -> Vec<BoundaryLintRule> {
5903 vec![
5904 BoundaryLintRule { rule_id: "direct-shell-os-system".into(), pattern: "os.system".into(), category: "shell".into(), message: "direct shell execution bypasses ToolGateway, policy, ledger, sandbox, and audit".into(), suggestion: "wrap shell execution as a runtime-managed tool".into(), prefix: false },
5905 BoundaryLintRule { rule_id: "direct-shell-subprocess".into(), pattern: "subprocess.".into(), category: "shell".into(), message: "direct subprocess execution bypasses ToolGateway, policy, ledger, sandbox, and audit".into(), suggestion: "wrap command execution as a runtime-managed tool".into(), prefix: true },
5906 BoundaryLintRule { rule_id: "direct-http-requests".into(), pattern: "requests.".into(), category: "network".into(), message: "direct HTTP calls bypass ToolGateway, policy, ledger, budget, replay, and audit".into(), suggestion: "register the HTTP/API call as a runtime-managed tool".into(), prefix: true },
5907 BoundaryLintRule { rule_id: "direct-http-httpx".into(), pattern: "httpx.".into(), category: "network".into(), message: "direct HTTP calls bypass ToolGateway, policy, ledger, budget, replay, and audit".into(), suggestion: "register the HTTP/API call as a runtime-managed tool".into(), prefix: true },
5908 BoundaryLintRule { rule_id: "direct-openai-sdk".into(), pattern: "openai.".into(), category: "model".into(), message: "direct model SDK usage bypasses model provider archives, replay, budget, and attribution".into(), suggestion: "call models through the runtime model boundary".into(), prefix: true },
5909 BoundaryLintRule { rule_id: "direct-anthropic-sdk".into(), pattern: "anthropic.".into(), category: "model".into(), message: "direct model SDK usage bypasses model provider archives, replay, budget, and attribution".into(), suggestion: "call models through the runtime model boundary".into(), prefix: true },
5910 ]
5911}
5912
5913pub fn scan_boundary_source(
5914 path: &str,
5915 source: &str,
5916 rules: Option<Vec<BoundaryLintRule>>,
5917) -> BoundaryLintReport {
5918 let rules = rules.unwrap_or_else(default_boundary_rules);
5919 let lines: Vec<&str> = source.split('\n').collect();
5920 let mut findings = Vec::new();
5921 for (i, line) in lines.iter().enumerate() {
5922 let previous = if i > 0 { lines[i - 1] } else { "" };
5923 if line.contains("agentledger: ignore-boundary")
5924 || previous.contains("agentledger: ignore-next-line")
5925 {
5926 continue;
5927 }
5928 for rule in &rules {
5929 if let Some(index) = line.find(&rule.pattern) {
5930 let mut callee = rule.pattern.clone();
5931 if rule.prefix {
5932 let mut end = index + rule.pattern.len();
5933 while end < line.len() {
5934 let ch = line.as_bytes()[end] as char;
5935 if ch.is_ascii_alphanumeric() || ch == '_' || ch == '.' {
5936 end += 1;
5937 } else {
5938 break;
5939 }
5940 }
5941 callee = line[index..end].to_string();
5942 }
5943 findings.push(BoundaryLintFinding {
5944 path: path.into(),
5945 line: i + 1,
5946 column: index + 1,
5947 rule_id: rule.rule_id.clone(),
5948 severity: "error".into(),
5949 callee,
5950 category: rule.category.clone(),
5951 message: rule.message.clone(),
5952 suggestion: rule.suggestion.clone(),
5953 });
5954 break;
5955 }
5956 }
5957 }
5958 BoundaryLintReport {
5959 passed: findings.is_empty(),
5960 scanned_files: vec![path.into()],
5961 finding_count: findings.len(),
5962 findings,
5963 }
5964}
5965
5966#[derive(Clone, Debug)]
5967pub struct RecoverySummary {
5968 pub recovered_steps: usize,
5969}
5970
5971#[derive(Clone, Debug)]
5972pub struct SchedulerStepStatus {
5973 pub step_id: String,
5974 pub status: String,
5975 pub owner: Option<String>,
5976 pub attempt: u64,
5977 pub lease_until: Option<f64>,
5978 pub last_error_type: Option<String>,
5979}
5980
5981#[derive(Clone, Debug)]
5982pub struct SchedulerStatus {
5983 pub run_id: String,
5984 pub run_status: String,
5985 pub state_version: u64,
5986 pub steps: Vec<SchedulerStepStatus>,
5987 pub cost_summary: CostSummary,
5988}
5989
5990pub struct RuntimeScheduler;
5991
5992impl RuntimeScheduler {
5993 pub fn recover_expired_leases(store: &mut MemoryStore) -> RecoverySummary {
5994 RecoverySummary {
5995 recovered_steps: store.recover_expired_leases(),
5996 }
5997 }
5998
5999 pub fn cancel_run(store: &mut MemoryStore, run_id: &str, reason: &str) -> Result<usize> {
6000 store.cancel_run(run_id, reason)
6001 }
6002
6003 pub fn status(store: &MemoryStore, run_id: &str) -> Result<SchedulerStatus> {
6004 let run = store.run(run_id)?;
6005 let steps = store
6006 .steps(run_id)
6007 .into_iter()
6008 .map(|step| SchedulerStepStatus {
6009 step_id: step.step_id,
6010 status: step.status,
6011 owner: step.owner,
6012 attempt: step.attempt,
6013 lease_until: step.lease_until,
6014 last_error_type: step.last_error_type,
6015 })
6016 .collect();
6017 Ok(SchedulerStatus {
6018 run_id: run_id.to_string(),
6019 run_status: run.status,
6020 state_version: run.state_version,
6021 steps,
6022 cost_summary: store.cost_summary(run_id),
6023 })
6024 }
6025}
6026
6027#[derive(Clone, Debug)]
6028pub struct ReviewCheck {
6029 pub name: String,
6030 pub passed: bool,
6031 pub severity: String,
6032 pub detail: String,
6033}
6034
6035#[derive(Clone, Debug)]
6036pub struct AdversarialReviewReport {
6037 pub passed: bool,
6038 pub run_id: Option<String>,
6039 pub checks: Vec<ReviewCheck>,
6040 pub metadata: State,
6041}
6042
6043pub fn adversarial_review(
6044 bundle: &EvidenceBundle,
6045 max_total_usd: Option<f64>,
6046) -> AdversarialReviewReport {
6047 let mut checks = vec![
6048 review_check(
6049 "no_failed_steps",
6050 !bundle
6051 .events
6052 .iter()
6053 .any(|event| event.event_type == "step_failed"),
6054 "blocker",
6055 "no step is in failed status",
6056 ),
6057 review_check(
6058 "no_pending_verification",
6059 !bundle
6060 .tool_ledger
6061 .iter()
6062 .any(|row| row.status == "PENDING_VERIFICATION"),
6063 "blocker",
6064 "no side effect is pending verification",
6065 ),
6066 review_check(
6067 "no_pending_approvals",
6068 !bundle.approvals.iter().any(|row| row.status == "PENDING"),
6069 "blocker",
6070 "no approval request is still pending",
6071 ),
6072 review_check(
6073 "completed_steps_have_completion_events",
6074 completed_steps_have_events_rust(&bundle.steps, &bundle.events),
6075 "blocker",
6076 "completed steps have step_completed events",
6077 ),
6078 review_check(
6079 "ledger_statuses_known",
6080 ledger_statuses_known_rust(&bundle.tool_ledger),
6081 "blocker",
6082 "Tool Ledger rows use known statuses",
6083 ),
6084 review_check(
6085 "event_sequence_contiguous",
6086 event_sequence_contiguous_rust(&bundle.events),
6087 "blocker",
6088 "event sequence has no gaps",
6089 ),
6090 review_check(
6091 "artifacts_have_blob_refs",
6092 bundle
6093 .artifacts
6094 .iter()
6095 .all(|row| !row.blob_ref.is_empty() && !row.blob_hash.is_empty()),
6096 "warning",
6097 "artifacts have blob refs and hashes",
6098 ),
6099 review_check(
6100 "media_artifacts_have_refs",
6101 bundle
6102 .media_artifacts
6103 .iter()
6104 .all(media_artifact_has_ref_rust),
6105 "blocker",
6106 "media artifacts have kind and durable refs",
6107 ),
6108 review_check(
6109 "stream_checkpoints_have_offsets",
6110 bundle
6111 .stream_checkpoints
6112 .iter()
6113 .all(stream_checkpoint_has_offset_rust),
6114 "blocker",
6115 "stream checkpoints have stream, consumer, and offset",
6116 ),
6117 review_check(
6118 "high_risk_approvals_decided",
6119 high_risk_approvals_decided_rust(&bundle.approvals),
6120 "blocker",
6121 "high-risk approval requests are decided",
6122 ),
6123 review_check(
6124 "no_blocking_failure_events",
6125 !bundle.events.iter().any(|event| {
6126 matches!(
6127 event.event_type.as_str(),
6128 "error_raised" | "step_failed" | "tool_call_failed" | "tool_call_blocked"
6129 )
6130 }),
6131 "warning",
6132 "no blocking failure events are present",
6133 ),
6134 ];
6135 if let Some(limit) = max_total_usd {
6136 checks.push(review_check(
6137 "max_total_usd",
6138 bundle.cost_summary.total_usd <= limit,
6139 "blocker",
6140 "cost limit check",
6141 ));
6142 }
6143 let passed = checks
6144 .iter()
6145 .all(|check| check.severity != "blocker" || check.passed);
6146 let mut metadata = State::new();
6147 metadata.insert(
6148 "event_count".into(),
6149 Value::Number(bundle.events.len() as f64),
6150 );
6151 metadata.insert(
6152 "tool_ledger_count".into(),
6153 Value::Number(bundle.tool_ledger.len() as f64),
6154 );
6155 metadata.insert(
6156 "approval_count".into(),
6157 Value::Number(bundle.approvals.len() as f64),
6158 );
6159 metadata.insert(
6160 "artifact_count".into(),
6161 Value::Number(bundle.artifacts.len() as f64),
6162 );
6163 metadata.insert(
6164 "media_artifact_count".into(),
6165 Value::Number(bundle.media_artifacts.len() as f64),
6166 );
6167 metadata.insert(
6168 "stream_checkpoint_count".into(),
6169 Value::Number(bundle.stream_checkpoints.len() as f64),
6170 );
6171 AdversarialReviewReport {
6172 passed,
6173 run_id: Some(bundle.run.run_id.clone()),
6174 checks,
6175 metadata,
6176 }
6177}
6178
6179fn review_check(name: &str, passed: bool, severity: &str, detail: &str) -> ReviewCheck {
6180 ReviewCheck {
6181 name: name.into(),
6182 passed,
6183 severity: severity.into(),
6184 detail: detail.into(),
6185 }
6186}
6187
6188fn completed_steps_have_events_rust(steps: &[Step], events: &[Event]) -> bool {
6189 steps.iter().all(|step| {
6190 step.status != "completed"
6191 || events.iter().any(|event| {
6192 event.event_type == "step_completed"
6193 && event.step_id.as_deref() == Some(step.step_id.as_str())
6194 })
6195 })
6196}
6197
6198fn ledger_statuses_known_rust(rows: &[ToolLedgerEntry]) -> bool {
6199 rows.iter().all(|row| {
6200 matches!(
6201 row.status.as_str(),
6202 "SUCCEEDED"
6203 | "FAILED_NO_EFFECT"
6204 | "PENDING_VERIFICATION"
6205 | "COMPENSATED"
6206 | "RUNNING"
6207 | "RESERVED"
6208 )
6209 })
6210}
6211fn event_sequence_contiguous_rust(events: &[Event]) -> bool {
6212 events
6213 .iter()
6214 .enumerate()
6215 .all(|(index, event)| event.seq == (index as u64) + 1)
6216}
6217fn state_has_key(row: &State, key: &str) -> bool {
6218 !matches!(row.get(key), None | Some(Value::Null))
6219}
6220fn media_artifact_has_ref_rust(row: &State) -> bool {
6221 state_has_key(row, "kind")
6222 && (state_has_key(row, "uri")
6223 || state_has_key(row, "content_ref")
6224 || state_has_key(row, "blob_ref"))
6225}
6226fn stream_checkpoint_has_offset_rust(row: &State) -> bool {
6227 state_has_key(row, "stream_id")
6228 && state_has_key(row, "consumer_id")
6229 && state_has_key(row, "offset")
6230}
6231fn high_risk_approvals_decided_rust(rows: &[ApprovalRequest]) -> bool {
6232 rows.iter().all(|row| {
6233 !matches!(
6234 row.risk_level.as_str(),
6235 "high" | "destructive" | "sensitive"
6236 ) || matches!(row.status.as_str(), "APPROVED" | "DENIED")
6237 })
6238}
6239
6240#[derive(Clone, Debug)]
6241pub struct EvidenceCheck {
6242 pub name: String,
6243 pub passed: bool,
6244 pub detail: String,
6245}
6246
6247#[derive(Clone, Debug)]
6248pub struct EvidenceCheckReport {
6249 pub passed: bool,
6250 pub checks: Vec<EvidenceCheck>,
6251 pub metadata: State,
6252}
6253
6254pub fn evaluate_evidence(
6255 bundle: &EvidenceBundle,
6256 max_total_usd: Option<f64>,
6257) -> EvidenceCheckReport {
6258 let mut checks = vec![
6259 evidence_check(
6260 "no_failed_steps",
6261 !bundle
6262 .events
6263 .iter()
6264 .any(|event| event.event_type == "step_failed"),
6265 "all steps completed or remain non-failed",
6266 ),
6267 evidence_check(
6268 "no_pending_verification",
6269 !bundle
6270 .tool_ledger
6271 .iter()
6272 .any(|row| row.status == "PENDING_VERIFICATION"),
6273 "no side effect is waiting for human/external verification",
6274 ),
6275 evidence_check(
6276 "completed_steps_have_events",
6277 completed_steps_have_events_rust(&bundle.steps, &bundle.events),
6278 "each completed step has a step_completed event",
6279 ),
6280 evidence_check(
6281 "managed_side_effects_are_ledgered",
6282 ledger_statuses_known_rust(&bundle.tool_ledger),
6283 "every ledger row has a known status",
6284 ),
6285 evidence_check(
6286 "media_artifacts_have_refs",
6287 bundle
6288 .media_artifacts
6289 .iter()
6290 .all(media_artifact_has_ref_rust),
6291 "media artifacts have kind and durable refs",
6292 ),
6293 evidence_check(
6294 "stream_checkpoints_have_offsets",
6295 bundle
6296 .stream_checkpoints
6297 .iter()
6298 .all(stream_checkpoint_has_offset_rust),
6299 "stream checkpoints have stream, consumer, and offset",
6300 ),
6301 ];
6302 if let Some(limit) = max_total_usd {
6303 checks.push(evidence_check(
6304 "max_total_usd",
6305 bundle.cost_summary.total_usd <= limit,
6306 "cost limit check",
6307 ));
6308 }
6309 EvidenceCheckReport {
6310 passed: checks.iter().all(|check| check.passed),
6311 checks,
6312 metadata: State::new(),
6313 }
6314}
6315
6316pub fn evaluate_evidence_regression(
6317 golden: &EvidenceBundle,
6318 current: &EvidenceBundle,
6319 max_total_usd_delta: Option<f64>,
6320) -> EvidenceCheckReport {
6321 let diff = diff_evidence(golden, current);
6322 let mut checks = vec![
6323 evidence_check(
6324 "final_state_regression",
6325 diff.final_state_changed_count == 0,
6326 "final state regression check",
6327 ),
6328 evidence_check(
6329 "event_type_regression",
6330 diff.event_types_changed_count == 0,
6331 "event type regression check",
6332 ),
6333 evidence_check(
6334 "tool_ledger_status_regression",
6335 true,
6336 "tool ledger status regression check",
6337 ),
6338 evidence_check(
6339 "media_artifact_regression",
6340 diff.media_artifacts_changed_count == 0,
6341 "media artifact regression check",
6342 ),
6343 evidence_check(
6344 "stream_checkpoint_regression",
6345 diff.stream_checkpoints_changed_count == 0,
6346 "stream checkpoint regression check",
6347 ),
6348 ];
6349 if let Some(limit) = max_total_usd_delta {
6350 let delta = current.cost_summary.total_usd - golden.cost_summary.total_usd;
6351 checks.push(evidence_check(
6352 "max_total_usd_delta",
6353 delta <= limit,
6354 "cost delta limit check",
6355 ));
6356 }
6357 EvidenceCheckReport {
6358 passed: checks.iter().all(|check| check.passed),
6359 checks,
6360 metadata: State::new(),
6361 }
6362}
6363
6364fn evidence_check(name: &str, passed: bool, detail: &str) -> EvidenceCheck {
6365 EvidenceCheck {
6366 name: name.into(),
6367 passed,
6368 detail: detail.into(),
6369 }
6370}
6371
6372#[derive(Clone, Debug)]
6373pub struct FailureInjectionCheck {
6374 pub name: String,
6375 pub passed: bool,
6376 pub detail: String,
6377 pub run_id: Option<String>,
6378}
6379
6380#[derive(Clone, Debug)]
6381pub struct FailureInjectionReport {
6382 pub passed: bool,
6383 pub checks: Vec<FailureInjectionCheck>,
6384}
6385
6386pub fn run_failure_injection_suite() -> FailureInjectionReport {
6387 let checks = vec![
6388 failure_retry_exhaustion(),
6389 failure_lease_fencing(),
6390 failure_cancellation_fencing(),
6391 failure_side_effect_idempotency(),
6392 ];
6393 FailureInjectionReport {
6394 passed: checks.iter().all(|check| check.passed),
6395 checks,
6396 }
6397}
6398fn failure_check(
6399 name: &str,
6400 passed: bool,
6401 detail: String,
6402 run_id: String,
6403) -> FailureInjectionCheck {
6404 FailureInjectionCheck {
6405 name: name.into(),
6406 passed,
6407 detail,
6408 run_id: Some(run_id),
6409 }
6410}
6411fn failure_retry_exhaustion() -> FailureInjectionCheck {
6412 let mut runtime = Runtime::new();
6413 let (run_id, _) = runtime.create_run(State::new());
6414 let _ = runtime.run_once(
6415 &run_id,
6416 "retry-1",
6417 "FailureInjector",
6418 60.0,
6419 |_ctx, _state| Err(RuntimeError("retryable".into())),
6420 );
6421 let _ = runtime.run_once(
6422 &run_id,
6423 "retry-2",
6424 "FailureInjector",
6425 60.0,
6426 |_ctx, _state| Err(RuntimeError("final failure".into())),
6427 );
6428 let status = runtime
6429 .store
6430 .run(&run_id)
6431 .map(|run| run.status)
6432 .unwrap_or_else(|_| "missing".into());
6433 failure_check(
6434 "retry_exhaustion",
6435 status == "failed",
6436 format!("run_status={status}"),
6437 run_id,
6438 )
6439}
6440fn failure_lease_fencing() -> FailureInjectionCheck {
6441 let mut store = MemoryStore::new();
6442 let (run_id, step_id) = store.create_run(State::new());
6443 let claim = store.claim_step("stale-worker", &run_id, 0.0).unwrap();
6444 let recovered = store.recover_expired_leases();
6445 let stale_rejected = store
6446 .commit_state_patch(&run_id, &step_id, &claim.lease_token, 0, State::new())
6447 .is_err();
6448 let fresh = store.claim_step("fresh-worker", &run_id, 60.0).unwrap();
6449 let passed = recovered == 1 && stale_rejected && fresh.attempt == 2;
6450 failure_check(
6451 "lease_fencing",
6452 passed,
6453 format!("recovered_steps={recovered} stale_rejected={stale_rejected}"),
6454 run_id,
6455 )
6456}
6457fn failure_cancellation_fencing() -> FailureInjectionCheck {
6458 let mut store = MemoryStore::new();
6459 let (run_id, step_id) = store.create_run(State::new());
6460 let claim = store.claim_step("stale-worker", &run_id, 60.0).unwrap();
6461 let cancelled = store.cancel_run(&run_id, "failure injection").unwrap();
6462 let stale_rejected = store
6463 .commit_state_patch(&run_id, &step_id, &claim.lease_token, 0, State::new())
6464 .is_err();
6465 let fresh = store.claim_step("fresh-worker", &run_id, 60.0).is_err();
6466 let status = store.run(&run_id).map(|run| run.status).unwrap_or_default();
6467 let passed = cancelled == 1 && stale_rejected && fresh && status == "cancelled";
6468 failure_check(
6469 "cancellation_fencing",
6470 passed,
6471 format!("cancelled_steps={cancelled} stale_rejected={stale_rejected}"),
6472 run_id,
6473 )
6474}
6475fn failure_side_effect_idempotency() -> FailureInjectionCheck {
6476 use std::sync::{Arc, Mutex};
6477 let calls = Arc::new(Mutex::new(0usize));
6478 let calls_for_tool = Arc::clone(&calls);
6479 let mut runtime = Runtime::new();
6480 runtime.register_tool(
6481 ToolSpec::new(
6482 "external.create",
6483 Box::new(move |_args| {
6484 let mut guard = calls_for_tool.lock().unwrap();
6485 *guard += 1;
6486 Ok(Value::Object(state(&[(
6487 "id",
6488 Value::String("EXT-1".into()),
6489 )])))
6490 }),
6491 )
6492 .side_effect("external")
6493 .idempotency_required(true),
6494 );
6495 let (run_id, _) = runtime.create_run(State::new());
6496 let ctx = failure_claim_context(&mut runtime, &run_id, "worker-1", "FailureInjector");
6497 let _ = runtime.call_tool(
6498 &ctx,
6499 "external.create",
6500 state(&[("title", Value::String("once".into()))]),
6501 );
6502 runtime
6503 .store
6504 .mark_retry(&run_id, &ctx.step_id, "RetryableAgentError", "retryable");
6505 let ctx2 = failure_claim_context(&mut runtime, &run_id, "worker-2", "FailureInjector");
6506 let _ = runtime.call_tool(
6507 &ctx2,
6508 "external.create",
6509 state(&[("title", Value::String("once".into()))]),
6510 );
6511 let count = *calls.lock().unwrap();
6512 failure_check(
6513 "side_effect_idempotency",
6514 count == 1,
6515 format!("external_call_count={count}"),
6516 run_id,
6517 )
6518}
6519
6520fn failure_claim_context(
6521 runtime: &mut Runtime,
6522 run_id: &str,
6523 worker_id: &str,
6524 agent_role: &str,
6525) -> AgentContext {
6526 let claim = runtime.store.claim_step(worker_id, run_id, 60.0).unwrap();
6527 AgentContext {
6528 run_id: claim.run_id,
6529 session_id: claim.session_id,
6530 step_id: claim.step_id,
6531 agent_role: agent_role.to_string(),
6532 lease_token: claim.lease_token,
6533 attempt: claim.attempt,
6534 state_version: claim.state_version,
6535 pending_patch: State::new(),
6536 }
6537}
6538
6539#[derive(Clone, Debug)]
6540pub struct ShadowReport {
6541 pub source_run_id: String,
6542 pub shadow_run_id: String,
6543 pub ok: bool,
6544 pub state_diff: State,
6545}
6546
6547pub fn diff_states(source: &State, shadow: &State) -> State {
6548 let mut changed = State::new();
6549 for key in source.keys().chain(shadow.keys()) {
6550 if changed.contains_key(key) {
6551 continue;
6552 }
6553 if source.get(key) != shadow.get(key) {
6554 changed.insert(
6555 key.clone(),
6556 Value::Object(state(&[
6557 ("source", source.get(key).cloned().unwrap_or_default()),
6558 ("shadow", shadow.get(key).cloned().unwrap_or_default()),
6559 ])),
6560 );
6561 }
6562 }
6563 state(&[
6564 ("changed", Value::Object(changed.clone())),
6565 ("changed_count", Value::Number(changed.len() as f64)),
6566 ])
6567}
6568
6569pub fn shadow_report(
6570 source_run_id: &str,
6571 shadow_run_id: &str,
6572 ok: bool,
6573 source_state: &State,
6574 shadow_state: &State,
6575) -> ShadowReport {
6576 ShadowReport {
6577 source_run_id: source_run_id.into(),
6578 shadow_run_id: shadow_run_id.into(),
6579 ok,
6580 state_diff: diff_states(source_state, shadow_state),
6581 }
6582}
6583
6584pub fn builtin_golden_names() -> Vec<String> {
6585 vec![
6586 "media-stream-checkpoint".into(),
6587 "minimal-success".into(),
6588 "tool-ledger-success".into(),
6589 ]
6590}
6591
6592pub fn builtin_golden_evidence(name: &str) -> Result<EvidenceBundle> {
6593 match name {
6594 "minimal-success" => golden_minimal_success(),
6595 "tool-ledger-success" => golden_tool_ledger_success(),
6596 "media-stream-checkpoint" => golden_media_stream_checkpoint(),
6597 _ => Err(RuntimeError(format!(
6598 "unknown built-in golden case: {name}"
6599 ))),
6600 }
6601}
6602
6603pub fn golden_regression(golden: &EvidenceBundle, current: &EvidenceBundle) -> EvidenceCheckReport {
6604 evaluate_evidence_regression(golden, current, None)
6605}
6606
6607fn golden_minimal_success() -> Result<EvidenceBundle> {
6608 let mut runtime = Runtime::new();
6609 let (run_id, _) = runtime.create_run(State::new());
6610 runtime.run_once(
6611 &run_id,
6612 "golden-worker",
6613 "GoldenAgent",
6614 60.0,
6615 |ctx, _state| {
6616 ctx.write_state("answer", Value::String("ok".into()));
6617 Ok(())
6618 },
6619 )?;
6620 export_evidence(&runtime.store, &run_id)
6621}
6622fn golden_tool_ledger_success() -> Result<EvidenceBundle> {
6623 let mut runtime = Runtime::new();
6624 runtime.register_tool(
6625 ToolSpec::new(
6626 "github.create_issue",
6627 Box::new(|_args| {
6628 Ok(Value::Object(state(&[(
6629 "issue_id",
6630 Value::String("ISSUE-1".into()),
6631 )])))
6632 }),
6633 )
6634 .side_effect("external"),
6635 );
6636 let (run_id, _) = runtime.create_run(State::new());
6637 runtime.run_once(
6638 &run_id,
6639 "golden-worker",
6640 "ExecutorAgent",
6641 60.0,
6642 |ctx, _state| {
6643 ctx.write_state("issue_id", Value::String("ISSUE-1".into()));
6644 Ok(())
6645 },
6646 )?;
6647 export_evidence(&runtime.store, &run_id)
6648}
6649fn golden_media_stream_checkpoint() -> Result<EvidenceBundle> {
6650 let mut runtime = Runtime::new();
6651 let (run_id, _) = runtime.create_run(State::new());
6652 runtime.run_once(
6653 &run_id,
6654 "golden-worker",
6655 "MediaAgent",
6656 60.0,
6657 |ctx, _state| {
6658 ctx.write_state("processed_offset", Value::Number(42.0));
6659 Ok(())
6660 },
6661 )?;
6662 runtime.store.create_artifact(
6663 &run_id,
6664 None,
6665 "golden-video-frame",
6666 State::new(),
6667 state(&[(
6668 "agentledger_media",
6669 Value::Object(state(&[
6670 ("kind", Value::String("frame".into())),
6671 ("uri", Value::String("file://golden-frame.jpg".into())),
6672 ])),
6673 )]),
6674 );
6675 runtime.store.create_artifact(
6676 &run_id,
6677 None,
6678 "golden-stream-checkpoint",
6679 State::new(),
6680 state(&[(
6681 "agentledger_stream",
6682 Value::Object(state(&[
6683 ("stream_id", Value::String("stream-golden".into())),
6684 ("consumer_id", Value::String("consumer-golden".into())),
6685 ("offset", Value::Number(42.0)),
6686 ])),
6687 )]),
6688 );
6689 export_evidence(&runtime.store, &run_id)
6690}
6691
6692#[derive(Clone, Debug)]
6693pub struct TimeTravelFrame {
6694 pub seq: u64,
6695 pub event_id: String,
6696 pub event_type: String,
6697 pub step_id: Option<String>,
6698 pub agent_role: Option<String>,
6699 pub state_version: Option<u64>,
6700 pub timestamp: f64,
6701 pub state_changed: bool,
6702 pub changed_keys: Vec<String>,
6703 pub patch: Option<State>,
6704 pub state_after: Option<State>,
6705}
6706
6707#[derive(Clone, Debug)]
6708pub struct TimeTravelReport {
6709 pub run_id: String,
6710 pub at_seq: Option<u64>,
6711 pub event_count: usize,
6712 pub timeline: Vec<TimeTravelFrame>,
6713 pub state_at_seq: State,
6714 pub selected_event: Option<TimeTravelFrame>,
6715}
6716
6717pub fn time_travel(
6718 bundle: &EvidenceBundle,
6719 at_seq: Option<u64>,
6720 include_states: bool,
6721) -> TimeTravelReport {
6722 let mut current = State::new();
6723 let mut state_at_seq = State::new();
6724 let mut selected_event = None;
6725 let mut timeline = Vec::new();
6726 for event in &bundle.events {
6727 let before = current.clone();
6728 let patch = patch_for_time_travel_event(event);
6729 if let Some(patch_value) = &patch {
6730 for (key, value) in patch_value {
6731 current.insert(key.clone(), value.clone());
6732 }
6733 }
6734 let diff = diff_states(&before, ¤t);
6735 let changed_keys = match diff.get("changed") {
6736 Some(Value::Object(obj)) => obj.keys().cloned().collect(),
6737 _ => Vec::new(),
6738 };
6739 let frame = TimeTravelFrame {
6740 seq: event.seq,
6741 event_id: event.event_id.clone(),
6742 event_type: event.event_type.clone(),
6743 step_id: event.step_id.clone(),
6744 agent_role: event.agent_role.clone(),
6745 state_version: event.state_version,
6746 timestamp: event.timestamp,
6747 state_changed: diff.get("changed_count") != Some(&Value::Number(0.0)),
6748 changed_keys,
6749 patch,
6750 state_after: if include_states {
6751 Some(current.clone())
6752 } else {
6753 None
6754 },
6755 };
6756 if at_seq.is_some_and(|seq| event.seq <= seq) {
6757 state_at_seq = current.clone();
6758 selected_event = Some(frame.clone());
6759 }
6760 timeline.push(frame);
6761 }
6762 if at_seq.is_none() {
6763 state_at_seq = current.clone();
6764 }
6765 TimeTravelReport {
6766 run_id: bundle.run.run_id.clone(),
6767 at_seq,
6768 event_count: timeline.len(),
6769 timeline,
6770 state_at_seq,
6771 selected_event,
6772 }
6773}
6774
6775fn patch_for_time_travel_event(event: &Event) -> Option<State> {
6776 if event.event_type == "run_created" {
6777 if let Some(Value::Object(obj)) = event.payload.get("initial_state") {
6778 return Some(obj.clone());
6779 }
6780 return Some(State::new());
6781 }
6782 if event.event_type == "state_committed"
6783 || event.event_type == "state_patch_committed"
6784 || event.event_type == "system_state_patch_applied"
6785 {
6786 if let Some(Value::Object(obj)) = event.payload.get("patch") {
6787 return Some(obj.clone());
6788 }
6789 return Some(State::new());
6790 }
6791 None
6792}
6793
6794pub fn time_travel_html(report: &TimeTravelReport) -> String {
6795 let rows = report
6796 .timeline
6797 .iter()
6798 .map(|frame| {
6799 format!(
6800 "<tr><td>{}</td><td>{}</td><td>{}</td></tr>",
6801 frame.seq,
6802 frame.event_type,
6803 frame.changed_keys.join(", ")
6804 )
6805 })
6806 .collect::<Vec<_>>()
6807 .join("\n");
6808 format!("<!doctype html><html><head><meta charset=\"utf-8\"><title>AgentLedger Time Travel Report</title></head><body><h1>AgentLedger Time Travel Report</h1><p>Run <code>{}</code></p><table>{}</table><h2>State At Selected Point</h2><pre>{:?}</pre><h2>Selected Event</h2><pre>{:?}</pre></body></html>", report.run_id, rows, report.state_at_seq, report.selected_event)
6809}
6810
6811#[derive(Debug, Clone, PartialEq, Eq)]
6812pub struct OptionalAdapterCapability {
6813 pub name: String,
6814 pub category: String,
6815 pub core_imports_heavy_sdks: bool,
6816 pub adapter_is_optional: bool,
6817 pub fail_closed_without_adapter: bool,
6818 pub contract_surface: Vec<String>,
6819}
6820
6821pub fn optional_adapter_capabilities() -> Vec<OptionalAdapterCapability> {
6822 fn item(name: &str, category: &str, surface: &[&str]) -> OptionalAdapterCapability {
6823 OptionalAdapterCapability {
6824 name: name.to_string(),
6825 category: category.to_string(),
6826 core_imports_heavy_sdks: false,
6827 adapter_is_optional: true,
6828 fail_closed_without_adapter: true,
6829 contract_surface: surface.iter().map(|s| s.to_string()).collect(),
6830 }
6831 }
6832 vec![
6833 item("postgres", "storage", &["ddl_for", "migrations_for", "state_store"]),
6834 item("mysql", "storage", &["ddl_for", "migrations_for", "state_store"]),
6835 item("s3", "blobstore", &["put_json", "get_json", "content_address"]),
6836 item("docker", "sandbox", &["sandbox_policy", "sandbox_result", "tool_gateway"]),
6837 item("e2b", "sandbox", &["sandbox_policy", "sandbox_result", "tool_gateway"]),
6838 item("bubblewrap", "sandbox", &["sandbox_policy", "sandbox_result", "tool_gateway"]),
6839 item("kubernetes", "sandbox", &["sandbox_policy", "sandbox_result", "tool_gateway"]),
6840 item("gvisor", "sandbox", &["sandbox_policy", "sandbox_result", "tool_gateway"]),
6841 item("firecracker", "sandbox", &["sandbox_policy", "sandbox_result", "tool_gateway"]),
6842 item("langgraph", "framework", &["framework_adapter", "checkpoint_contract"]),
6843 item("langchain", "framework", &["framework_adapter"]),
6844 item("crewai", "framework", &["framework_adapter"]),
6845 item("autogen", "framework", &["framework_adapter"]),
6846 item("openai-agents-sdk", "framework", &["framework_adapter"]),
6847 item("llamaindex", "framework", &["framework_adapter"]),
6848 item("semantic-kernel", "framework", &["framework_adapter"]),
6849 item("mcp-transport", "mcp", &["mcp_tool_descriptor", "mcp_resource_descriptor"]),
6850 item("langfuse", "observability", &["evidence_bundle", "trace_payload", "correlation_ids"]),
6851 item("shadow-runner", "shadow", &["evidence_bundle", "tool_ledger", "state_diff"]),
6852 ]
6853}
6854
6855pub trait SqlExecutor {
6856 fn exec(&mut self, sql: &str, params: &[Value]) -> Result<()>;
6857}
6858
6859pub struct PostgresAdapter<C: SqlExecutor> {
6860 pub schema: String,
6861 pub client: C,
6862}
6863
6864impl<C: SqlExecutor> PostgresAdapter<C> {
6865 pub fn new(client: C, schema: &str) -> Self {
6866 Self { schema: if schema.is_empty() { "agentledger".to_string() } else { schema.to_string() }, client }
6867 }
6868 pub fn migration_plan(&self) -> Result<Vec<Migration>> { migrations_for("postgres") }
6869 pub fn apply_migrations(&mut self) -> Result<()> {
6870 self.client.exec(&ddl_for("postgres")?, &[])?;
6871 for migration in self.migration_plan()? {
6872 self.client.exec(
6873 "INSERT INTO schema_migrations(version, name, checksum) VALUES ($1, $2, $3) ON CONFLICT (version) DO NOTHING",
6874 &[Value::String(migration.version), Value::String(migration.name), Value::String(stable_hash(&migration.sql))],
6875 )?;
6876 }
6877 Ok(())
6878 }
6879}
6880
6881pub struct MySQLAdapter<C: SqlExecutor> {
6882 pub database: String,
6883 pub client: C,
6884}
6885
6886impl<C: SqlExecutor> MySQLAdapter<C> {
6887 pub fn new(client: C, database: &str) -> Self {
6888 Self { database: if database.is_empty() { "agentledger".to_string() } else { database.to_string() }, client }
6889 }
6890 pub fn migration_plan(&self) -> Result<Vec<Migration>> { migrations_for("mysql") }
6891 pub fn apply_migrations(&mut self) -> Result<()> {
6892 self.client.exec(&ddl_for("mysql")?, &[])?;
6893 for migration in self.migration_plan()? {
6894 self.client.exec(
6895 "INSERT INTO schema_migrations(version, name, checksum, applied_at) VALUES (?, ?, ?, UNIX_TIMESTAMP()) ON DUPLICATE KEY UPDATE version=version",
6896 &[Value::String(migration.version), Value::String(migration.name), Value::String(stable_hash(&migration.sql))],
6897 )?;
6898 }
6899 Ok(())
6900 }
6901}
6902
6903pub trait ObjectClient {
6904 fn put_object(&mut self, bucket: &str, key: &str, body: &[u8], content_type: &str, metadata: State) -> Result<()>;
6905 fn get_object(&mut self, bucket: &str, key: &str) -> Result<Vec<u8>>;
6906}
6907
6908pub struct S3BlobStoreAdapter<C: ObjectClient> {
6909 pub bucket: String,
6910 pub prefix: String,
6911 pub client: C,
6912}
6913
6914impl<C: ObjectClient> S3BlobStoreAdapter<C> {
6915 pub fn new(client: C, bucket: &str, prefix: &str) -> Self {
6916 Self { bucket: bucket.to_string(), prefix: if prefix.is_empty() { "agentledger/blobs".to_string() } else { prefix.trim_matches('/').to_string() }, client }
6917 }
6918 pub fn put_json(&mut self, value: &Value) -> Result<(String, String)> {
6919 let encoded = encode_value(value);
6920 let digest = stable_hash(&encoded);
6921 let key = format!("{}/sha256/{}.json", self.prefix, digest);
6922 let mut metadata = State::new();
6923 metadata.insert("agentledger-digest".to_string(), Value::String(format!("sha256:{digest}")));
6924 self.client.put_object(&self.bucket, &key, encoded.as_bytes(), "application/json", metadata)?;
6925 Ok((format!("sha256:{digest}"), format!("s3://{}/{}", self.bucket, key)))
6926 }
6927 pub fn get_json(&mut self, reference: &str) -> Result<Value> {
6928 let prefix = format!("s3://{}/", self.bucket);
6929 if !reference.starts_with(&prefix) || reference.contains("..") {
6930 return Err(RuntimeError(format!("unsupported s3 blob ref: {reference}")));
6931 }
6932 let key = &reference[prefix.len()..];
6933 let body = self.client.get_object(&self.bucket, key)?;
6934 let text = String::from_utf8(body).map_err(|err| RuntimeError(err.to_string()))?;
6935 decode_value(&text)
6936 }
6937}
6938
6939pub trait OtlpClient { fn post_json(&mut self, endpoint: &str, payload: &str, content_type: &str) -> Result<()>; }
6940pub struct OtlpTransport<C: OtlpClient> { pub endpoint: String, pub client: C }
6941impl<C: OtlpClient> OtlpTransport<C> { pub fn export(&mut self, payload: &str) -> Result<()> { self.client.post_json(&self.endpoint, payload, "application/json") } }
6942
6943pub struct DockerSandboxAdapter { pub image: String }
6944impl DockerSandboxAdapter {
6945 pub fn manifest(&self, policy: &State, command: Vec<String>) -> State {
6946 let mut out = State::new();
6947 out.insert("backend".to_string(), Value::String("docker".to_string()));
6948 out.insert("image".to_string(), Value::String(if self.image.is_empty() { "python:3.11-slim".to_string() } else { self.image.clone() }));
6949 let network = match policy.get("network") { Some(Value::String(value)) if value != "deny" => value.clone(), _ => "none".to_string() };
6950 out.insert("network".to_string(), Value::String(network));
6951 out.insert("read_only_root".to_string(), Value::Bool(true));
6952 out.insert("requires_explicit_execution".to_string(), Value::Bool(true));
6953 out.insert("command".to_string(), Value::Array(command.into_iter().map(Value::String).collect()));
6954 out
6955 }
6956}
6957
6958pub struct DockerSandboxExecutor {
6959 pub image: String,
6960 pub binary: String,
6961 pub allow_command_execution: bool,
6962 pub allow_shell: bool,
6963 pub shell: String,
6964 pub memory: String,
6965 pub cpus: String,
6966}
6967
6968impl DockerSandboxExecutor {
6969 pub fn new(image: &str, allow_command_execution: bool) -> Self {
6970 Self {
6971 image: image.to_string(),
6972 binary: "docker".to_string(),
6973 allow_command_execution,
6974 allow_shell: false,
6975 shell: "/bin/sh".to_string(),
6976 memory: String::new(),
6977 cpus: String::new(),
6978 }
6979 }
6980
6981 pub fn with_binary(mut self, binary: &str) -> Self {
6982 self.binary = binary.to_string();
6983 self
6984 }
6985
6986 fn extract_command(&self, args: &State) -> std::result::Result<Vec<String>, String> {
6987 let raw = args.get("_sandbox_command").or_else(|| args.get("command"));
6988 match raw {
6989 Some(Value::String(command)) => {
6990 if !self.allow_shell {
6991 Err("string commands require allow_shell=true; pass argv list in `_sandbox_command` instead".to_string())
6992 } else {
6993 let shell = if self.shell.is_empty() { "/bin/sh" } else { &self.shell };
6994 Ok(vec![shell.to_string(), "-lc".to_string(), command.clone()])
6995 }
6996 }
6997 Some(Value::Array(items)) => {
6998 let mut command = Vec::new();
6999 for item in items {
7000 match item {
7001 Value::String(value) if !value.is_empty() => command.push(value.clone()),
7002 _ => return Err("_sandbox_command must be a non-empty string array".to_string()),
7003 }
7004 }
7005 if command.is_empty() {
7006 Err("_sandbox_command must be a non-empty string array".to_string())
7007 } else {
7008 Ok(command)
7009 }
7010 }
7011 _ => Err("external sandbox tools require a command-style `_sandbox_command` arg".to_string()),
7012 }
7013 }
7014
7015 fn docker_argv(&self, policy: &SandboxPolicy, command: &[String]) -> Vec<String> {
7016 let image = if self.image.is_empty() { "python:3.11-slim" } else { &self.image };
7017 let network = if policy.network == "deny" || policy.network.is_empty() { "none" } else { &policy.network };
7018 let mut argv = vec![
7019 self.binary.clone(),
7020 "run".to_string(),
7021 "--rm".to_string(),
7022 "--network".to_string(),
7023 network.to_string(),
7024 "--read-only".to_string(),
7025 ];
7026 if !self.memory.is_empty() {
7027 argv.extend(["--memory".to_string(), self.memory.clone()]);
7028 }
7029 if !self.cpus.is_empty() {
7030 argv.extend(["--cpus".to_string(), self.cpus.clone()]);
7031 }
7032 argv.push(image.to_string());
7033 argv.extend(command.iter().cloned());
7034 argv
7035 }
7036
7037 fn result_error(policy: &SandboxPolicy, manifest: State, error_type: &str, error: String) -> SandboxResult {
7038 let mut metadata = State::new();
7039 metadata.insert("executor".to_string(), Value::String(policy.executor.clone()));
7040 metadata.insert("isolation_level".to_string(), Value::String("container".to_string()));
7041 metadata.insert("manifest".to_string(), Value::Object(manifest));
7042 metadata.insert("error_type".to_string(), Value::String(error_type.to_string()));
7043 SandboxResult { ok: false, output: Value::Null, error: Some(error), metadata }
7044 }
7045}
7046
7047impl SandboxExecutor for DockerSandboxExecutor {
7048 fn run_tool(&self, args: State, policy: &SandboxPolicy) -> SandboxResult {
7049 let command = match self.extract_command(&args) {
7050 Ok(command) => command,
7051 Err(error) => return Self::result_error(policy, State::new(), "InvalidSandboxCommand", error),
7052 };
7053 let mut policy_state = State::new();
7054 policy_state.insert("network".to_string(), Value::String(policy.network.clone()));
7055 let manifest = (DockerSandboxAdapter { image: self.image.clone() }).manifest(&policy_state, command.clone());
7056 if !self.allow_command_execution {
7057 return Self::result_error(policy, manifest, "SandboxAdapterNotInstalled", "command execution is not enabled for this executor".to_string());
7058 }
7059 let argv = self.docker_argv(policy, &command);
7060 let mut cmd = Command::new(&argv[0]);
7061 cmd.args(&argv[1..]).stdout(Stdio::piped()).stderr(Stdio::piped());
7062 let output = match cmd.output() {
7063 Ok(output) => output,
7064 Err(error) => return Self::result_error(policy, manifest, "SandboxBinaryMissing", error.to_string()),
7065 };
7066 let mut value = State::new();
7067 value.insert("stdout".to_string(), Value::String(String::from_utf8_lossy(&output.stdout).to_string()));
7068 value.insert("stderr".to_string(), Value::String(String::from_utf8_lossy(&output.stderr).to_string()));
7069 value.insert("returncode".to_string(), Value::Number(output.status.code().unwrap_or(-1) as f64));
7070 let mut metadata = State::new();
7071 metadata.insert("executor".to_string(), Value::String(policy.executor.clone()));
7072 metadata.insert("isolation_level".to_string(), Value::String("container".to_string()));
7073 metadata.insert("manifest".to_string(), Value::Object(manifest));
7074 metadata.insert("executed".to_string(), Value::Bool(true));
7075 if !output.status.success() {
7076 metadata.insert("error_type".to_string(), Value::String("SandboxCommandFailed".to_string()));
7077 return SandboxResult {
7078 ok: false,
7079 output: Value::Object(value),
7080 error: Some(format!("sandbox command exited with {}", output.status.code().unwrap_or(-1))),
7081 metadata,
7082 };
7083 }
7084 SandboxResult { ok: true, output: Value::Object(value), error: None, metadata }
7085 }
7086}
7087
7088pub mod adapters {
7089 pub mod postgres {
7090 pub const PACKAGE_NAME: &str = "agentledger-postgres";
7091 pub const FEATURE: &str = "adapter-postgres";
7092 pub use crate::{migrations_for, Migration, PostgresAdapter, SqlExecutor};
7093 }
7094
7095 pub mod mysql {
7096 pub const PACKAGE_NAME: &str = "agentledger-mysql";
7097 pub const FEATURE: &str = "adapter-mysql";
7098 pub use crate::{migrations_for, Migration, MySQLAdapter, SqlExecutor};
7099 }
7100
7101 pub mod s3 {
7102 pub const PACKAGE_NAME: &str = "agentledger-s3";
7103 pub const FEATURE: &str = "adapter-s3";
7104 pub use crate::{ObjectClient, S3BlobStoreAdapter};
7105 }
7106
7107 pub mod mcp {
7108 pub const PACKAGE_NAME: &str = "agentledger-mcp";
7109 pub const FEATURE: &str = "adapter-mcp";
7110 pub use crate::{
7111 InMemoryMCPContextServer, InMemoryMCPToolServer, MCPCall, MCPContextAdapter,
7112 MCPResourceDescriptor, MCPResourceRead, MCPToolAdapter,
7113 };
7114 }
7115
7116 pub mod otel {
7117 pub const PACKAGE_NAME: &str = "agentledger-otel";
7118 pub const FEATURE: &str = "adapter-otel";
7119 pub use crate::{OtlpClient, OtlpTransport};
7120 }
7121
7122 pub mod langfuse {
7123 pub const PACKAGE_NAME: &str = "agentledger-langfuse";
7124 pub const FEATURE: &str = "adapter-langfuse";
7125 pub const CATEGORY: &str = "observability";
7126 }
7127
7128 pub mod docker {
7129 pub const PACKAGE_NAME: &str = "agentledger-sandbox-docker";
7130 pub const FEATURE: &str = "adapter-docker";
7131 pub use crate::{DockerSandboxAdapter, DockerSandboxExecutor, State, Value};
7132 }
7133
7134 pub mod framework {
7135 pub const PACKAGE_NAME: &str = "agentledger-framework";
7136 pub const FEATURE: &str = "adapter-framework";
7137 pub use crate::{FunctionAdapter, MethodFrameworkAdapter};
7138 }
7139}