lean_ctx/core/execution_ledger/
mod.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Append-only, task-aware execution ledger.
3//!
4//! The execution ledger is an additive projection beside the Savings Ledger.  It
5//! shares the Savings Ledger's SHA-256 chain primitive, but stores execution
6//! identity and lifecycle observations keyed by task and trace IDs.
7
8pub mod canonical_receipt;
9pub mod event;
10pub mod host;
11pub mod migration;
12pub mod producer;
13pub mod projection;
14pub(crate) mod receipt_inputs;
15pub mod store;
16pub mod verify;
17
18pub use canonical_receipt::{PublishedCanonicalReceipt, publish_canonical_receipt};
19pub use event::{ContextBalance, ContextBalanceV1, ExecutionEvent, TriState};
20pub use migration::{migrate_from_savings_ledger, migrate_from_savings_ledger_at};
21pub use producer::{
22    CanonicalReceiptRecordV1, ReceiptSignerAdmissionV1, record_canonical_engine_receipt,
23};
24pub use projection::{
25    CanonicalReceiptProjectionV1, ExecutionProjection, LegacyReceiptProjectionV1, TaskCostSummary,
26    canonical_receipt_for_task_from_store, canonical_receipt_for_task_verified_from_store,
27    legacy_receipt_for_task, legacy_receipt_for_task_from_store, task_cost_summary,
28    task_cost_summary_from_store,
29};
30pub use store::{ExecutionLedgerStore, default_path};
31pub use verify::{GENESIS, hash_event, verify_events};
32
33use std::io;
34
35use serde::{Deserialize, Serialize};
36
37/// Errors returned while reading, serializing, or validating the execution ledger.
38#[derive(Debug, thiserror::Error)]
39pub enum ExecutionLedgerError {
40    #[error("execution ledger I/O failed: {0}")]
41    Io(#[from] io::Error),
42    #[error("execution ledger serialization failed: {0}")]
43    Serialization(#[from] serde_json::Error),
44    #[error("execution ledger chain is invalid: {0}")]
45    InvalidChain(String),
46    #[error("execution ledger record is invalid: {0}")]
47    InvalidRecord(String),
48}
49
50/// Result type shared by execution-ledger operations.
51pub type Result<T> = std::result::Result<T, ExecutionLedgerError>;
52
53/// Compatibility view consumed by the existing `ledger execution` CLI.
54///
55/// The event store remains the source of truth; this view groups the current
56/// task projection without introducing a second writable ledger format. Legacy
57/// receipt evidence and its estimates remain explicitly named and never occupy
58/// canonical receipt or accepted-cost fields.
59#[derive(Debug, Clone, Serialize, Deserialize)]
60pub struct ExecutionLedgerEntryV1 {
61    pub task_id: String,
62    #[serde(alias = "receipt")]
63    pub legacy_receipt: Option<LegacyReceiptProjectionV1>,
64    #[serde(alias = "actual_cost_micros")]
65    pub legacy_estimated_actual_cost_micros: Option<u64>,
66    #[serde(alias = "baseline_cost_micros")]
67    pub legacy_estimated_baseline_cost_micros: Option<u64>,
68    pub predicted_cost_micros: Option<u64>,
69    pub actual_etpao: Option<crate::core::etpao::EtpaoResult>,
70    pub baseline_etpao: Option<crate::core::etpao::EtpaoResult>,
71    pub predicted_etpao: Option<crate::core::etpao::EtpaoResult>,
72}
73
74/// Compatibility aggregate for the existing execution-ledger CLI commands.
75#[derive(Debug, Clone, Default, Serialize, Deserialize)]
76pub struct ExecutionLedger {
77    pub entries: Vec<ExecutionLedgerEntryV1>,
78}
79
80#[derive(Debug, Clone, Default)]
81pub struct ExecutionLedgerVerifyResult {
82    pub valid: bool,
83    pub total_entries: usize,
84    pub first_invalid_at: Option<usize>,
85    pub error: Option<String>,
86}
87
88impl ExecutionLedger {
89    /// Loads the compatibility projection from the default event store.
90    pub fn load() -> Result<Self> {
91        let store = ExecutionLedgerStore::from_default()?;
92        let events = store.load_verified()?;
93        let mut task_ids = Vec::new();
94        for event in events {
95            if !task_ids.iter().any(|known| known == event.task_id()) {
96                task_ids.push(event.task_id().to_owned());
97            }
98        }
99
100        let entries = task_ids
101            .into_iter()
102            .map(|task_id| {
103                let legacy_receipt = legacy_receipt_for_task_from_store(&store, &task_id);
104                ExecutionLedgerEntryV1 {
105                    task_id,
106                    legacy_estimated_actual_cost_micros: legacy_receipt
107                        .as_ref()
108                        .map(|value| value.legacy_receipt().actual_cost_micros),
109                    legacy_estimated_baseline_cost_micros: legacy_receipt
110                        .as_ref()
111                        .map(|value| value.legacy_receipt().baseline_cost_micros),
112                    predicted_cost_micros: None,
113                    actual_etpao: None,
114                    baseline_etpao: None,
115                    predicted_etpao: None,
116                    legacy_receipt,
117                }
118            })
119            .collect();
120        Ok(Self { entries })
121    }
122
123    /// Finds the first projection for a task.
124    #[must_use]
125    pub fn find_task(&self, task_id: &str) -> Option<&ExecutionLedgerEntryV1> {
126        self.entries.iter().find(|entry| entry.task_id == task_id)
127    }
128
129    /// Verifies the source event chain and reports CLI-friendly details.
130    #[must_use]
131    pub fn verify(&self) -> ExecutionLedgerVerifyResult {
132        match ExecutionLedgerStore::from_default().and_then(|store| store.verify_chain()) {
133            Ok(valid) => ExecutionLedgerVerifyResult {
134                valid,
135                total_entries: self.entries.len(),
136                first_invalid_at: (!valid).then_some(0),
137                error: (!valid).then(|| "hash or sequence link mismatch".to_owned()),
138            },
139            Err(error) => ExecutionLedgerVerifyResult {
140                valid: false,
141                total_entries: self.entries.len(),
142                first_invalid_at: Some(0),
143                error: Some(error.to_string()),
144            },
145        }
146    }
147}
148
149#[cfg(test)]
150mod tests {
151    use super::*;
152    use lean_ctx_protocol::AcceptanceState;
153    use tempfile::tempdir;
154
155    fn task_started(task_id: &str, trace_id: &str) -> ExecutionEvent {
156        ExecutionEvent::TaskStarted {
157            task_id: task_id.to_owned(),
158            trace_id: trace_id.to_owned(),
159            envelope_ref: "envelope:task".to_owned(),
160            timestamp: "2026-08-09T12:00:00Z".to_owned(),
161            sequence_number: 0,
162            prev_hash: String::new(),
163        }
164    }
165
166    fn model_invoked(task_id: &str, trace_id: &str) -> ExecutionEvent {
167        model_invoked_with(task_id, trace_id, "invocation-1", "model-1", 20, 10)
168    }
169
170    fn model_invoked_with(
171        task_id: &str,
172        trace_id: &str,
173        invocation_id: &str,
174        model: &str,
175        tokens_in: u64,
176        tokens_out: u64,
177    ) -> ExecutionEvent {
178        ExecutionEvent::ModelInvoked {
179            task_id: task_id.to_owned(),
180            trace_id: trace_id.to_owned(),
181            plan_id: "plan-1".to_owned(),
182            invocation_id: invocation_id.to_owned(),
183            invocation_ref: format!("sha256:{invocation_id}"),
184            model: model.to_owned(),
185            provider: "provider-1".to_owned(),
186            tokens_in,
187            tokens_out,
188            latency_ms: 25,
189            timestamp: "2026-08-09T12:00:01Z".to_owned(),
190            sequence_number: 0,
191            prev_hash: String::new(),
192        }
193    }
194
195    fn context_delivered(task_id: &str, trace_id: &str) -> ExecutionEvent {
196        ExecutionEvent::ContextDelivered {
197            task_id: task_id.to_owned(),
198            trace_id: trace_id.to_owned(),
199            context_balance: ContextBalanceV1 {
200                original_tokens: 100,
201                materialized_tokens: 80,
202                delivered_tokens: 60,
203                provider_billed_tokens: 60,
204            },
205            timestamp: "2026-08-09T12:00:00Z".to_owned(),
206            sequence_number: 0,
207            prev_hash: String::new(),
208        }
209    }
210
211    fn engine_invoked(task_id: &str, trace_id: &str) -> ExecutionEvent {
212        ExecutionEvent::EngineInvoked {
213            task_id: task_id.to_owned(),
214            trace_id: trace_id.to_owned(),
215            plan_id: "plan-1".to_owned(),
216            invocation_id: "invocation-engine-1".to_owned(),
217            invocation_ref: "sha256:engine-invocation-1".to_owned(),
218            capability_id: "capability://leanctx/context".to_owned(),
219            capability_version: "1.0.0".to_owned(),
220            timestamp: "2026-08-09T12:00:01Z".to_owned(),
221            sequence_number: 0,
222            prev_hash: String::new(),
223        }
224    }
225
226    fn canonical_record(
227        task_id: &str,
228        trace_id: &str,
229        invocation_id: &str,
230        digit: char,
231    ) -> ExecutionEvent {
232        ExecutionEvent::CanonicalReceiptRecorded {
233            task_id: task_id.to_owned(),
234            trace_id: trace_id.to_owned(),
235            invocation_id: invocation_id.to_owned(),
236            receipt_id: format!("sha256:{}", digit.to_string().repeat(64)),
237            receipt_ref: format!("id:sha256:{}", digit.to_string().repeat(64)),
238            receipt_digest: format!("sha256:{}", digit.to_string().repeat(64)),
239            receipt_chain_id: "shared-chain".to_owned(),
240            receipt_sequence_number: 1,
241            previous_receipt_id: None,
242            previous_signature_digest: None,
243            timestamp: "2026-08-23T12:00:00Z".to_owned(),
244            sequence_number: 0,
245            prev_hash: String::new(),
246            entry_hash: String::new(),
247        }
248    }
249
250    #[test]
251    fn append_assigns_sequence_and_verifies_chain() {
252        let directory = tempdir().expect("temporary directory");
253        let store = ExecutionLedgerStore::new(directory.path().join("ledger.jsonl"));
254        store.append(task_started("task-1", "trace-1")).unwrap();
255        store
256            .append(ExecutionEvent::PlanCreated {
257                task_id: "task-1".to_owned(),
258                trace_id: "trace-1".to_owned(),
259                plan_id: "plan-1".to_owned(),
260                plan_ref: "plan:1".to_owned(),
261                timestamp: "2026-08-09T12:00:00Z".to_owned(),
262                sequence_number: 0,
263                prev_hash: String::new(),
264            })
265            .unwrap();
266        store
267            .append(context_delivered("task-1", "trace-1"))
268            .unwrap();
269        store.append(model_invoked("task-1", "trace-1")).unwrap();
270
271        assert_eq!(store.last_sequence(), 4);
272        assert!(store.verify_chain().unwrap());
273    }
274
275    #[test]
276    fn engine_invocation_is_auditable_without_fabricated_model_usage() {
277        let directory = tempdir().expect("temporary directory");
278        let store = ExecutionLedgerStore::new(directory.path().join("ledger.jsonl"));
279        store.append(task_started("task-1", "trace-1")).unwrap();
280        store
281            .append(ExecutionEvent::PlanCreated {
282                task_id: "task-1".to_owned(),
283                trace_id: "trace-1".to_owned(),
284                plan_id: "plan-1".to_owned(),
285                plan_ref: "sha256:plan-1".to_owned(),
286                timestamp: "2026-08-09T12:00:00Z".to_owned(),
287                sequence_number: 0,
288                prev_hash: String::new(),
289            })
290            .unwrap();
291        store
292            .append(context_delivered("task-1", "trace-1"))
293            .unwrap();
294        store.append(engine_invoked("task-1", "trace-1")).unwrap();
295
296        assert!(store.verify_chain().unwrap());
297        let summary = store.task_cost_summary("task-1");
298        assert_eq!(summary.model_calls, 0);
299        assert_eq!(
300            summary.cost_classification(),
301            lean_ctx_protocol::ReceiptValueClassificationV1::Unavailable
302        );
303        assert_eq!(
304            summary.baseline_cost_classification(),
305            lean_ctx_protocol::ReceiptValueClassificationV1::Unavailable
306        );
307        assert!(store.legacy_receipt_for_task("task-1").is_none());
308    }
309
310    #[test]
311    fn canonical_receipt_self_hash_detects_last_event_tampering() {
312        let directory = tempdir().expect("temporary directory");
313        let path = directory.path().join("ledger.jsonl");
314        let store = ExecutionLedgerStore::new(&path);
315        store.append(task_started("task-1", "trace-1")).unwrap();
316        store
317            .append(ExecutionEvent::PlanCreated {
318                task_id: "task-1".to_owned(),
319                trace_id: "trace-1".to_owned(),
320                plan_id: "plan-1".to_owned(),
321                plan_ref: "plan:1".to_owned(),
322                timestamp: "2026-08-23T11:59:58Z".to_owned(),
323                sequence_number: 0,
324                prev_hash: String::new(),
325            })
326            .unwrap();
327        store
328            .append(ExecutionEvent::ContextDelivered {
329                task_id: "task-1".to_owned(),
330                trace_id: "trace-1".to_owned(),
331                context_balance: ContextBalanceV1 {
332                    original_tokens: 1,
333                    materialized_tokens: 1,
334                    delivered_tokens: 1,
335                    provider_billed_tokens: 1,
336                },
337                timestamp: "2026-08-23T11:59:58Z".to_owned(),
338                sequence_number: 0,
339                prev_hash: String::new(),
340            })
341            .unwrap();
342        store.append(model_invoked("task-1", "trace-1")).unwrap();
343        store
344            .append(ExecutionEvent::CanonicalReceiptRecorded {
345                task_id: "task-1".to_owned(),
346                trace_id: "trace-1".to_owned(),
347                invocation_id: "invocation-1".to_owned(),
348                receipt_id: format!("sha256:{}", "1".repeat(64)),
349                receipt_ref: format!("id:sha256:{}", "2".repeat(64)),
350                receipt_digest: format!("sha256:{}", "2".repeat(64)),
351                receipt_chain_id: "chain-1".to_owned(),
352                receipt_sequence_number: 1,
353                previous_receipt_id: None,
354                previous_signature_digest: None,
355                timestamp: "2026-08-23T12:00:00Z".to_owned(),
356                sequence_number: 0,
357                prev_hash: String::new(),
358                entry_hash: String::new(),
359            })
360            .unwrap();
361        assert!(store.verify_chain().unwrap());
362        let projected = store
363            .canonical_receipt_for_task_verified("task-1")
364            .unwrap()
365            .unwrap();
366        assert_eq!(projected.invocation_id, "invocation-1");
367        assert_eq!(projected.chain_id, "chain-1");
368        assert_eq!(projected.sequence_number, 1);
369
370        let content = std::fs::read_to_string(&path).unwrap();
371        std::fs::write(&path, content.replace(&"1".repeat(64), &"3".repeat(64))).unwrap();
372
373        assert!(!store.verify_chain().unwrap());
374    }
375
376    #[test]
377    fn verify_chain_rejects_modified_event() {
378        let directory = tempdir().expect("temporary directory");
379        let path = directory.path().join("ledger.jsonl");
380        let store = ExecutionLedgerStore::new(&path);
381        store.append(task_started("task-1", "trace-1")).unwrap();
382        store
383            .append(ExecutionEvent::PlanCreated {
384                task_id: "task-1".to_owned(),
385                trace_id: "trace-1".to_owned(),
386                plan_id: "plan-1".to_owned(),
387                plan_ref: "plan:1".to_owned(),
388                timestamp: "2026-08-09T12:00:00Z".to_owned(),
389                sequence_number: 0,
390                prev_hash: String::new(),
391            })
392            .unwrap();
393        store
394            .append(context_delivered("task-1", "trace-1"))
395            .unwrap();
396        store.append(model_invoked("task-1", "trace-1")).unwrap();
397        let mut line = std::fs::read_to_string(&path).unwrap();
398        line = line.replace("envelope:task", "envelope:changed");
399        std::fs::write(&path, line).unwrap();
400
401        assert!(!store.verify_chain().unwrap());
402    }
403
404    #[test]
405    fn record_digest_detects_single_tail_event_tampering() {
406        let directory = tempdir().expect("temporary directory");
407        let path = directory.path().join("ledger.jsonl");
408        let store = ExecutionLedgerStore::new(&path);
409        store.append(task_started("task-1", "trace-1")).unwrap();
410        let content = std::fs::read_to_string(&path).unwrap();
411        std::fs::write(&path, content.replace("trace-1", "trace-x")).unwrap();
412
413        assert!(!store.verify_chain().unwrap());
414    }
415
416    #[test]
417    fn append_rejects_cross_trace_and_out_of_order_events() {
418        let directory = tempdir().expect("temporary directory");
419        let store = ExecutionLedgerStore::new(directory.path().join("ledger.jsonl"));
420        store.append(task_started("task-1", "trace-1")).unwrap();
421
422        assert!(matches!(
423            store.append(model_invoked("task-1", "trace-1")),
424            Err(ExecutionLedgerError::InvalidChain(_))
425        ));
426        assert!(matches!(
427            store.append(ExecutionEvent::PlanCreated {
428                task_id: "task-1".to_owned(),
429                trace_id: "trace-x".to_owned(),
430                plan_id: "plan-1".to_owned(),
431                plan_ref: "plan:1".to_owned(),
432                timestamp: "2026-08-09T12:00:00Z".to_owned(),
433                sequence_number: 0,
434                prev_hash: String::new(),
435            }),
436            Err(ExecutionLedgerError::InvalidChain(_))
437        ));
438        assert_eq!(store.last_sequence(), 1);
439    }
440
441    #[test]
442    fn canonical_receipt_chain_rejects_cross_task_forks_and_unlinked_outcomes() {
443        let directory = tempdir().expect("temporary directory");
444        let store = ExecutionLedgerStore::new(directory.path().join("ledger.jsonl"));
445        for (task_id, trace_id) in [("task-1", "trace-1"), ("task-2", "trace-2")] {
446            store.append(task_started(task_id, trace_id)).unwrap();
447            store
448                .append(ExecutionEvent::PlanCreated {
449                    task_id: task_id.to_owned(),
450                    trace_id: trace_id.to_owned(),
451                    plan_id: "plan-1".to_owned(),
452                    plan_ref: "plan:1".to_owned(),
453                    timestamp: "2026-08-23T11:59:00Z".to_owned(),
454                    sequence_number: 0,
455                    prev_hash: String::new(),
456                })
457                .unwrap();
458            store.append(context_delivered(task_id, trace_id)).unwrap();
459            store.append(engine_invoked(task_id, trace_id)).unwrap();
460        }
461        assert!(matches!(
462            store.append(ExecutionEvent::OutcomeRecorded {
463                task_id: "task-1".to_owned(),
464                trace_id: "trace-1".to_owned(),
465                outcome_id: "outcome-1".to_owned(),
466                receipt_id: format!("sha256:{}", "1".repeat(64)),
467                accepted: AcceptanceState::Accepted,
468                timestamp: "2026-08-23T12:00:00Z".to_owned(),
469                sequence_number: 0,
470                prev_hash: String::new(),
471            }),
472            Err(ExecutionLedgerError::InvalidChain(_))
473        ));
474        store
475            .append(canonical_record(
476                "task-1",
477                "trace-1",
478                "invocation-engine-1",
479                '1',
480            ))
481            .unwrap();
482        assert!(matches!(
483            store.append(canonical_record(
484                "task-2",
485                "trace-2",
486                "invocation-engine-1",
487                '2',
488            )),
489            Err(ExecutionLedgerError::InvalidChain(_))
490        ));
491
492        let mut duplicate_id = canonical_record("task-2", "trace-2", "invocation-engine-1", '1');
493        if let ExecutionEvent::CanonicalReceiptRecorded {
494            receipt_ref,
495            receipt_digest,
496            receipt_chain_id,
497            ..
498        } = &mut duplicate_id
499        {
500            *receipt_digest = format!("sha256:{}", "3".repeat(64));
501            *receipt_ref = format!("id:{receipt_digest}");
502            *receipt_chain_id = "independent-chain".to_owned();
503        }
504        assert!(matches!(
505            store.append(duplicate_id),
506            Err(ExecutionLedgerError::InvalidChain(_))
507        ));
508    }
509
510    #[test]
511    fn legacy_receipts_never_authorize_or_downgrade_canonical_outcomes() {
512        let directory = tempdir().expect("temporary directory");
513        let store = ExecutionLedgerStore::new(directory.path().join("ledger.jsonl"));
514        store.append(task_started("task-1", "trace-1")).unwrap();
515        store
516            .append(ExecutionEvent::PlanCreated {
517                task_id: "task-1".to_owned(),
518                trace_id: "trace-1".to_owned(),
519                plan_id: "plan-1".to_owned(),
520                plan_ref: "plan:1".to_owned(),
521                timestamp: "2026-08-23T11:59:00Z".to_owned(),
522                sequence_number: 0,
523                prev_hash: String::new(),
524            })
525            .unwrap();
526        store
527            .append(context_delivered("task-1", "trace-1"))
528            .unwrap();
529        store.append(model_invoked("task-1", "trace-1")).unwrap();
530        let canonical_id = format!("sha256:{}", "1".repeat(64));
531        store
532            .append(ExecutionEvent::ReceiptSigned {
533                task_id: "task-1".to_owned(),
534                trace_id: "trace-1".to_owned(),
535                receipt_id: canonical_id.clone(),
536                receipt_hash: "arbitrary-nonempty-hash".to_owned(),
537                signature: "arbitrary-nonempty-signature".to_owned(),
538                timestamp: "2026-08-23T12:00:00Z".to_owned(),
539                sequence_number: 0,
540                prev_hash: String::new(),
541            })
542            .unwrap();
543        let legacy_outcome = |receipt_id: &str, outcome_id: &str| ExecutionEvent::OutcomeRecorded {
544            task_id: "task-1".to_owned(),
545            trace_id: "trace-1".to_owned(),
546            outcome_id: outcome_id.to_owned(),
547            receipt_id: receipt_id.to_owned(),
548            accepted: AcceptanceState::Accepted,
549            timestamp: "2026-08-23T12:00:01Z".to_owned(),
550            sequence_number: 0,
551            prev_hash: String::new(),
552        };
553
554        assert!(
555            store
556                .append(legacy_outcome(&canonical_id, "legacy-only"))
557                .is_err()
558        );
559        store
560            .append(canonical_record("task-1", "trace-1", "invocation-1", '1'))
561            .unwrap();
562        store
563            .append(ExecutionEvent::ReceiptSigned {
564                task_id: "task-1".to_owned(),
565                trace_id: "trace-1".to_owned(),
566                receipt_id: "legacy-downgrade".to_owned(),
567                receipt_hash: "still-arbitrary".to_owned(),
568                signature: "still-arbitrary".to_owned(),
569                timestamp: "2026-08-23T12:00:02Z".to_owned(),
570                sequence_number: 0,
571                prev_hash: String::new(),
572            })
573            .unwrap();
574        assert!(
575            store
576                .append(legacy_outcome("legacy-downgrade", "downgrade"))
577                .is_err()
578        );
579        store
580            .append(legacy_outcome(&canonical_id, "canonical-outcome"))
581            .unwrap();
582        assert!(
583            store
584                .append(ExecutionEvent::OutcomeRecorded {
585                    task_id: "task-1".to_owned(),
586                    trace_id: "trace-1".to_owned(),
587                    outcome_id: "contradictory-canonical-outcome".to_owned(),
588                    receipt_id: canonical_id.clone(),
589                    accepted: AcceptanceState::Rejected,
590                    timestamp: "2026-08-23T12:00:02Z".to_owned(),
591                    sequence_number: 0,
592                    prev_hash: String::new(),
593                })
594                .is_err()
595        );
596        assert!(
597            store
598                .append(ExecutionEvent::ReceiptSigned {
599                    task_id: "task-1".to_owned(),
600                    trace_id: "trace-1".to_owned(),
601                    receipt_id: "legacy-downgrade".to_owned(),
602                    receipt_hash: "different".to_owned(),
603                    signature: "different".to_owned(),
604                    timestamp: "2026-08-23T12:00:03Z".to_owned(),
605                    sequence_number: 0,
606                    prev_hash: String::new(),
607                })
608                .is_err()
609        );
610
611        let canonical = store
612            .canonical_receipt_for_task_verified("task-1")
613            .unwrap()
614            .unwrap();
615        assert_eq!(canonical.receipt_id, canonical_id);
616        assert!(store.legacy_receipt_for_task("task-1").is_none());
617    }
618
619    #[test]
620    fn load_rejects_duplicate_json_keys() {
621        let directory = tempdir().expect("temporary directory");
622        let path = directory.path().join("ledger.jsonl");
623        let store = ExecutionLedgerStore::new(&path);
624        store.append(task_started("task-1", "trace-1")).unwrap();
625        let content = std::fs::read_to_string(&path).unwrap();
626        let duplicated = content.replacen(
627            "\"task_id\":\"task-1\",",
628            "\"task_id\":\"task-1\",\"task_id\":\"task-1\",",
629            1,
630        );
631        std::fs::write(&path, duplicated).unwrap();
632
633        assert!(matches!(
634            store.load(),
635            Err(ExecutionLedgerError::Serialization(_))
636        ));
637    }
638
639    #[test]
640    fn load_rejects_noncanonical_json_whitespace() {
641        let directory = tempdir().expect("temporary directory");
642        let path = directory.path().join("ledger.jsonl");
643        let store = ExecutionLedgerStore::new(&path);
644        store.append(task_started("task-1", "trace-1")).unwrap();
645        let content = std::fs::read_to_string(&path).unwrap();
646        std::fs::write(&path, format!(" {content}")).unwrap();
647
648        assert!(matches!(
649            store.load(),
650            Err(ExecutionLedgerError::InvalidRecord(message))
651                if message == "execution ledger record is not canonical JSON"
652        ));
653    }
654
655    #[test]
656    fn by_task_filters_events_without_losing_order() {
657        let directory = tempdir().expect("temporary directory");
658        let store = ExecutionLedgerStore::new(directory.path().join("ledger.jsonl"));
659        store.append(task_started("task-1", "trace-1")).unwrap();
660        store
661            .append(ExecutionEvent::PlanCreated {
662                task_id: "task-1".to_owned(),
663                trace_id: "trace-1".to_owned(),
664                plan_id: "plan-1".to_owned(),
665                plan_ref: "plan:1".to_owned(),
666                timestamp: "2026-08-09T12:00:00Z".to_owned(),
667                sequence_number: 0,
668                prev_hash: String::new(),
669            })
670            .unwrap();
671        store.append(task_started("task-2", "trace-2")).unwrap();
672        store
673            .append(context_delivered("task-1", "trace-1"))
674            .unwrap();
675        store.append(model_invoked("task-1", "trace-1")).unwrap();
676
677        let events = store.by_task("task-1");
678        assert_eq!(events.len(), 4);
679        assert_eq!(events[0].sequence_number(), 1);
680        assert_eq!(events[1].sequence_number(), 2);
681        assert_eq!(events[2].sequence_number(), 4);
682        assert_eq!(events[3].sequence_number(), 5);
683    }
684
685    #[test]
686    fn migration_reads_source_without_mutating_it() {
687        let directory = tempdir().expect("temporary directory");
688        let savings_path = directory.path().join("savings.jsonl");
689        let execution_path = directory.path().join("execution.jsonl");
690        let savings_event = crate::core::savings_ledger::event::SavingsEvent {
691            ts: "2026-08-09T12:00:00Z".to_owned(),
692            tool: "ctx_read".to_owned(),
693            mechanism: "compression".to_owned(),
694            model_id: "model-1".to_owned(),
695            tokenizer: "o200k_base".to_owned(),
696            baseline_tokens: 100,
697            actual_tokens: 60,
698            saved_tokens: 40,
699            bounce_adjustment: 0,
700            unit_price_per_m_usd: 1.0,
701            saved_usd: 0.00004,
702            repo_hash: "repo".to_owned(),
703            agent_id: "agent".to_owned(),
704            prev_hash: String::new(),
705            entry_hash: String::new(),
706            version: "test".to_owned(),
707            intent_tag: None,
708            outcome: None,
709            model_original: None,
710            model_routed: None,
711            routing_savings: None,
712            response_original_tokens: None,
713            response_delivered_tokens: None,
714            agent_chain_id: None,
715            chain_depth: None,
716            measurement_method: None,
717            evidence_class: None,
718            confidence: None,
719            request_id: None,
720            session_id: None,
721            trace_id: None,
722            solution_decision: None,
723            loc_added: None,
724            loc_removed: None,
725            path: None,
726            lines_added: None,
727            lines_removed: None,
728            net: None,
729            quality_signal: None,
730            attribution_group: None,
731            attribution_id: None,
732            baseline_ref: None,
733            price_version: None,
734            customer_approval: None,
735            settlement_status: None,
736            is_first_inject: None,
737            cache_read_per_m_usd: None,
738            cache_write_per_m_usd: None,
739            governor: None,
740        };
741        let mut second_savings_event = savings_event.clone();
742        second_savings_event.ts = "2026-08-09T12:00:01Z".to_owned();
743        second_savings_event.model_id = "model-2".to_owned();
744        crate::core::savings_ledger::store::append(&savings_path, savings_event).unwrap();
745        crate::core::savings_ledger::store::append(&savings_path, second_savings_event).unwrap();
746        let before = std::fs::read(&savings_path).unwrap();
747        let source_hashes: Vec<String> = crate::core::savings_ledger::store::load(&savings_path)
748            .into_iter()
749            .map(|event| event.entry_hash)
750            .collect();
751
752        let destination = ExecutionLedgerStore::new(execution_path);
753        assert_eq!(
754            migrate_from_savings_ledger_at(
755                &savings_path,
756                &destination,
757                "task-legacy",
758                "trace-legacy"
759            )
760            .unwrap(),
761            2
762        );
763        assert_eq!(before, std::fs::read(&savings_path).unwrap());
764        assert_eq!(
765            migrate_from_savings_ledger_at(
766                &savings_path,
767                &destination,
768                "task-legacy",
769                "trace-legacy"
770            )
771            .unwrap(),
772            0
773        );
774        let migrated_events = destination.by_task("task-legacy");
775        assert_eq!(migrated_events.len(), 4);
776        assert_eq!(
777            migrated_events
778                .iter()
779                .filter(|event| matches!(event, ExecutionEvent::PlanCreated { .. }))
780                .count(),
781            1
782        );
783        for source_hash in source_hashes {
784            assert!(migrated_events.iter().any(|event| {
785                matches!(event, ExecutionEvent::ModelInvoked { invocation_ref, .. } if invocation_ref.contains(&source_hash))
786            }));
787        }
788    }
789
790    #[test]
791    fn projections_summarize_cost_and_build_receipt() {
792        let directory = tempdir().expect("temporary directory");
793        let store = ExecutionLedgerStore::new(directory.path().join("ledger.jsonl"));
794        store.append(task_started("task-1", "trace-1")).unwrap();
795        store
796            .append(ExecutionEvent::PlanCreated {
797                task_id: "task-1".to_owned(),
798                trace_id: "trace-1".to_owned(),
799                plan_id: "plan-1".to_owned(),
800                plan_ref: "plan:1".to_owned(),
801                timestamp: "2026-08-09T12:00:00Z".to_owned(),
802                sequence_number: 0,
803                prev_hash: String::new(),
804            })
805            .unwrap();
806        store
807            .append(ExecutionEvent::ContextDelivered {
808                task_id: "task-1".to_owned(),
809                trace_id: "trace-1".to_owned(),
810                context_balance: ContextBalanceV1 {
811                    original_tokens: 100,
812                    materialized_tokens: 80,
813                    delivered_tokens: 60,
814                    provider_billed_tokens: 60,
815                },
816                timestamp: "2026-08-09T12:00:00Z".to_owned(),
817                sequence_number: 0,
818                prev_hash: String::new(),
819            })
820            .unwrap();
821        store
822            .append(context_delivered("task-1", "trace-1"))
823            .unwrap();
824        store
825            .append(model_invoked_with(
826                "task-1",
827                "trace-1",
828                "invocation-1",
829                "gpt-4o",
830                20,
831                10,
832            ))
833            .unwrap();
834        store
835            .append(model_invoked_with(
836                "task-1",
837                "trace-1",
838                "invocation-2",
839                "claude-sonnet-4.5",
840                30,
841                15,
842            ))
843            .unwrap();
844        store
845            .append(ExecutionEvent::ReceiptSigned {
846                task_id: "task-1".to_owned(),
847                trace_id: "trace-1".to_owned(),
848                receipt_id: "receipt-1".to_owned(),
849                receipt_hash: "sha256:receipt".to_owned(),
850                signature: "signature".to_owned(),
851                timestamp: "2026-08-09T12:00:02Z".to_owned(),
852                sequence_number: 0,
853                prev_hash: String::new(),
854            })
855            .unwrap();
856        assert!(matches!(
857            store.append(ExecutionEvent::OutcomeRecorded {
858                task_id: "task-1".to_owned(),
859                trace_id: "trace-1".to_owned(),
860                outcome_id: "outcome-1".to_owned(),
861                receipt_id: "receipt-1".to_owned(),
862                accepted: AcceptanceState::Accepted,
863                timestamp: "2026-08-09T12:00:03Z".to_owned(),
864                sequence_number: 0,
865                prev_hash: String::new(),
866            }),
867            Err(ExecutionLedgerError::InvalidChain(_))
868        ));
869
870        let summary = task_cost_summary_from_store(&store, "task-1");
871        assert_eq!(summary.total_tokens, 75);
872        assert_eq!(summary.model_calls, 2);
873        assert_eq!(
874            summary.cost_classification(),
875            lean_ctx_protocol::ReceiptValueClassificationV1::Estimated
876        );
877        assert_eq!(
878            summary.baseline_cost_classification(),
879            lean_ctx_protocol::ReceiptValueClassificationV1::Unavailable
880        );
881        let last_model_quote =
882            crate::core::gain::model_pricing::ModelPricing::load().quote(Some("claude-sonnet-4.5"));
883        let repriced_at_last_model = (last_model_quote.cost.estimate_usd(
884            summary.total_tokens_in,
885            summary.total_tokens_out,
886            0,
887            0,
888        ) * 1_000_000.0)
889            .round() as u64;
890        assert_ne!(summary.cost_micros, repriced_at_last_model);
891        let legacy_receipt = legacy_receipt_for_task_from_store(&store, "task-1").unwrap();
892        assert_eq!(
893            legacy_receipt.cost_classification(),
894            lean_ctx_protocol::ReceiptValueClassificationV1::Estimated
895        );
896        assert_eq!(
897            legacy_receipt.baseline_cost_classification(),
898            lean_ctx_protocol::ReceiptValueClassificationV1::Unavailable
899        );
900        assert_eq!(
901            legacy_receipt.legacy_receipt().actual_cost_micros,
902            summary.cost_micros
903        );
904        assert_eq!(
905            legacy_receipt.legacy_receipt().baseline_cost_micros,
906            legacy_receipt.legacy_receipt().actual_cost_micros
907        );
908        let entry = ExecutionLedgerEntryV1 {
909            task_id: "task-1".to_owned(),
910            legacy_estimated_actual_cost_micros: Some(
911                legacy_receipt.legacy_receipt().actual_cost_micros,
912            ),
913            legacy_estimated_baseline_cost_micros: Some(
914                legacy_receipt.legacy_receipt().baseline_cost_micros,
915            ),
916            legacy_receipt: Some(legacy_receipt.clone()),
917            predicted_cost_micros: None,
918            actual_etpao: None,
919            baseline_etpao: None,
920            predicted_etpao: None,
921        };
922        let serialized = serde_json::to_value(entry).unwrap();
923        assert!(serialized.get("legacy_receipt").is_some());
924        assert!(
925            serialized
926                .get("legacy_estimated_actual_cost_micros")
927                .is_some()
928        );
929        assert!(
930            serialized
931                .get("legacy_estimated_baseline_cost_micros")
932                .is_some()
933        );
934        assert!(serialized.get("receipt").is_none());
935        assert!(serialized.get("actual_cost_micros").is_none());
936        assert!(serialized.get("baseline_cost_micros").is_none());
937        let mut invalid = serialized.clone();
938        invalid["legacy_receipt"]["schema_version"] = serde_json::json!(2);
939        assert!(serde_json::from_value::<ExecutionLedgerEntryV1>(invalid).is_err());
940        let mut outcome_authority = serialized.clone();
941        outcome_authority["legacy_receipt"]["outcome_ref"] = serde_json::json!("outcome-1");
942        assert!(serde_json::from_value::<ExecutionLedgerEntryV1>(outcome_authority).is_err());
943        let mut signature_authority = serialized;
944        signature_authority["legacy_receipt"]["evidence_refs"][0]["signature_status"] =
945            serde_json::json!("Verified");
946        assert!(serde_json::from_value::<ExecutionLedgerEntryV1>(signature_authority).is_err());
947
948        let receipt = legacy_receipt.into_legacy_receipt();
949        assert_eq!(receipt.model_calls, 2);
950        assert_eq!(receipt.output_tokens, 25);
951        assert!(receipt.outcome_ref.is_none());
952    }
953}
lean_ctx/core/execution_ledger/mod.rs

lean_ctx/core/execution_ledger/
mod.rs
