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