lean_ctx/core/context_store/
lineage.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Decision lineage of one task: what was planned, returned or prepared,
3//! what it cost and which outcome followed — joined by `TaskId` from the
4//! execution ledger and the Context Gateway's Decision Receipts.
5//!
6//! Read-only and content-free: the ledger holds references and counts, the
7//! receipts hold decisions, counts and digests. A link that is absent is
8//! reported as a gap; nothing is inferred to fill it.
9
10use std::collections::BTreeMap;
11
12use lean_ctx_protocol::AcceptanceState;
13use lean_ctx_protocol::context_gateway::ContextDecisionReceiptV1;
14use serde::Serialize;
15
16use crate::core::context_admission::receipt_store::{LoadError, StoredReceipt, TaskReceipts};
17use crate::core::execution_ledger::ExecutionEvent;
18
19pub(crate) const LINEAGE_SCHEMA_VERSION: u32 = 1;
20
21/// One ledger observation, without its chain fields.
22#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
23pub struct LineageStepV1 {
24    pub sequence: u64,
25    pub kind: &'static str,
26    pub timestamp: String,
27    /// Identifiers and counts of the step, sorted by name.
28    pub fields: BTreeMap<&'static str, String>,
29}
30
31/// One governed delivery, summarized from its verified receipt.
32#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
33pub struct LineageDeliveryV1 {
34    pub digest: String,
35    pub verified: bool,
36    #[serde(skip_serializing_if = "Option::is_none")]
37    pub error: Option<&'static str>,
38    #[serde(skip_serializing_if = "Option::is_none")]
39    pub summary: Option<DeliverySummaryV1>,
40    #[serde(skip_serializing_if = "Option::is_none")]
41    pub egress: Option<EgressLineageV1>,
42}
43
44/// Independent evidence about dispatch, never inferred from admission or billing.
45#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
46pub struct EgressLineageV1 {
47    pub admission: Option<lean_ctx_protocol::context_gateway::EgressAdmissionReceiptV1>,
48    pub observations: Vec<lean_ctx_protocol::context_gateway::EgressTransportObservationV1>,
49    #[serde(skip_serializing_if = "Option::is_none")]
50    pub error: Option<&'static str>,
51}
52
53#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
54pub struct DeliverySummaryV1 {
55    pub outcome: String,
56    /// Existing count fields describe this boundary, never inferred model usage.
57    pub delivery_semantics: &'static str,
58    pub destination: String,
59    pub policy_digest: Option<String>,
60    pub inspected: u32,
61    pub delivered: u32,
62    pub withheld: u32,
63    pub redactions: u32,
64    pub tokens_original: u64,
65    pub tokens_delivered: u64,
66    pub final_context: Option<String>,
67}
68
69/// The task's lineage. `gaps` names every missing link in the chain
70/// plan → delivery → outcome, so an incomplete record never reads as complete.
71#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
72pub struct TaskLineageV1 {
73    pub schema_version: u32,
74    pub task_id: String,
75    /// The tenant/project scope the lineage was looked up in.
76    #[serde(skip_serializing_if = "Option::is_none")]
77    pub scope: Option<String>,
78    pub steps: Vec<LineageStepV1>,
79    pub deliveries: Vec<LineageDeliveryV1>,
80    /// `None` when the ledger could not be read or verified.
81    #[serde(skip_serializing_if = "Option::is_none")]
82    pub ledger_error: Option<String>,
83    pub outcome: String,
84    pub gaps: Vec<&'static str>,
85}
86
87/// Whether the ledger's entries for this task id belong to the requested
88/// tenant/project scope, proven by the task envelope the ledger references.
89#[derive(Debug, Clone, Copy, PartialEq, Eq)]
90pub enum LedgerScope {
91    /// The envelope names this scope, or the ledger has no entry for the task.
92    Verified,
93    /// The envelope names another scope: the entries are not shown.
94    Foreign,
95    /// No readable envelope proves the scope: the entries are not shown.
96    Unverified,
97}
98
99pub fn build(
100    task_id: &str,
101    events: Result<Vec<ExecutionEvent>, String>,
102    ledger_scope: LedgerScope,
103    receipts: &TaskReceipts,
104) -> TaskLineageV1 {
105    let (events, mut ledger_error) = match events {
106        Ok(events) => (events, None),
107        Err(error) => (Vec::new(), Some(error)),
108    };
109    let events = match ledger_scope {
110        LedgerScope::Foreign => {
111            ledger_error = Some("the ledger's task belongs to another project scope".to_owned());
112            Vec::new()
113        }
114        LedgerScope::Unverified if !events.is_empty() => {
115            ledger_error = Some("no task envelope proves the ledger's project scope".to_owned());
116            Vec::new()
117        }
118        LedgerScope::Verified | LedgerScope::Unverified => events,
119    };
120    let steps: Vec<LineageStepV1> = events.iter().map(step).collect();
121    let deliveries: Vec<LineageDeliveryV1> = receipts.entries.iter().map(delivery).collect();
122
123    let has = |kind: &str| steps.iter().any(|step| step.kind == kind);
124    // The last recorded outcome wins; a task without one stays unknown.
125    let outcome = events
126        .iter()
127        .rev()
128        .find_map(|event| match event {
129            ExecutionEvent::OutcomeRecorded { accepted, .. } => Some(*accepted),
130            _ => None,
131        })
132        .map_or_else(|| "unknown".to_owned(), acceptance_name);
133
134    let mut gaps = Vec::new();
135    if ledger_error.is_some() {
136        gaps.push("ledger_unverified");
137    }
138    if !has("plan_created") && !has("decision_recorded") {
139        gaps.push("no_plan_recorded");
140    }
141    if deliveries.is_empty() && !has("context_delivered") {
142        gaps.push("no_delivery_recorded");
143    }
144    if deliveries.iter().any(|delivery| !delivery.verified) {
145        gaps.push("delivery_unverified");
146    }
147    if !receipts.index_complete {
148        gaps.push("deliveries_incomplete");
149    }
150    if !has("outcome_recorded") {
151        gaps.push("no_outcome_recorded");
152    }
153
154    TaskLineageV1 {
155        schema_version: LINEAGE_SCHEMA_VERSION,
156        task_id: task_id.to_owned(),
157        scope: None,
158        steps,
159        deliveries,
160        ledger_error,
161        outcome,
162        gaps,
163    }
164}
165
166fn acceptance_name(value: AcceptanceState) -> String {
167    serde_json::to_value(value)
168        .ok()
169        .and_then(|v| v.as_str().map(str::to_owned))
170        .unwrap_or_else(|| "unknown".to_owned())
171}
172
173fn step(event: &ExecutionEvent) -> LineageStepV1 {
174    let mut fields = BTreeMap::new();
175    let kind = match event {
176        ExecutionEvent::TaskStarted { envelope_ref, .. } => {
177            fields.insert("envelope_ref", envelope_ref.clone());
178            "task_started"
179        }
180        ExecutionEvent::PlanCreated {
181            plan_id, plan_ref, ..
182        } => {
183            fields.insert("plan_id", plan_id.clone());
184            fields.insert("plan_ref", plan_ref.clone());
185            "plan_created"
186        }
187        ExecutionEvent::ContextDelivered {
188            context_balance, ..
189        } => {
190            fields.insert(
191                "original_tokens",
192                context_balance.original_tokens.to_string(),
193            );
194            fields.insert(
195                "materialized_tokens",
196                context_balance.materialized_tokens.to_string(),
197            );
198            fields.insert(
199                "delivered_tokens",
200                context_balance.delivered_tokens.to_string(),
201            );
202            fields.insert(
203                "provider_billed_tokens",
204                context_balance.provider_billed_tokens.to_string(),
205            );
206            "context_delivered"
207        }
208        ExecutionEvent::ModelInvoked {
209            plan_id,
210            invocation_id,
211            model,
212            provider,
213            tokens_in,
214            tokens_out,
215            latency_ms,
216            ..
217        } => {
218            fields.insert("plan_id", plan_id.clone());
219            fields.insert("invocation_id", invocation_id.clone());
220            fields.insert("model", model.clone());
221            fields.insert("provider", provider.clone());
222            fields.insert("tokens_in", tokens_in.to_string());
223            fields.insert("tokens_out", tokens_out.to_string());
224            fields.insert("latency_ms", latency_ms.to_string());
225            "model_invoked"
226        }
227        ExecutionEvent::EngineInvoked {
228            plan_id,
229            invocation_id,
230            capability_id,
231            capability_version,
232            ..
233        } => {
234            fields.insert("plan_id", plan_id.clone());
235            fields.insert("invocation_id", invocation_id.clone());
236            fields.insert("capability_id", capability_id.clone());
237            fields.insert("capability_version", capability_version.clone());
238            "engine_invoked"
239        }
240        ExecutionEvent::ReceiptSigned {
241            receipt_id,
242            receipt_hash,
243            ..
244        } => {
245            fields.insert("receipt_id", receipt_id.clone());
246            fields.insert("receipt_hash", receipt_hash.clone());
247            "receipt_signed"
248        }
249        ExecutionEvent::CanonicalReceiptRecorded {
250            invocation_id,
251            receipt_id,
252            receipt_ref,
253            receipt_digest,
254            ..
255        } => {
256            fields.insert("invocation_id", invocation_id.clone());
257            fields.insert("receipt_id", receipt_id.clone());
258            fields.insert("receipt_ref", receipt_ref.clone());
259            fields.insert("receipt_digest", receipt_digest.clone());
260            "canonical_receipt_recorded"
261        }
262        ExecutionEvent::OutcomeRecorded {
263            outcome_id,
264            receipt_id,
265            accepted,
266            ..
267        } => {
268            fields.insert("outcome_id", outcome_id.clone());
269            fields.insert("receipt_id", receipt_id.clone());
270            fields.insert("accepted", acceptance_name(*accepted));
271            "outcome_recorded"
272        }
273        ExecutionEvent::DecisionRecorded {
274            decision_id,
275            kind,
276            selected,
277            ..
278        } => {
279            fields.insert("decision_id", decision_id.clone());
280            fields.insert("decision_kind", kind.clone());
281            fields.insert("selected", selected.clone());
282            "decision_recorded"
283        }
284    };
285    LineageStepV1 {
286        sequence: event.sequence_number(),
287        kind,
288        timestamp: event.timestamp().to_owned(),
289        fields,
290    }
291}
292
293fn delivery((hex, loaded): &(String, Result<StoredReceipt, LoadError>)) -> LineageDeliveryV1 {
294    match loaded {
295        Ok(stored) => LineageDeliveryV1 {
296            digest: stored.digest.as_str().to_owned(),
297            verified: true,
298            error: None,
299            summary: Some(summarize(&stored.receipt)),
300            egress: None,
301        },
302        Err(error) => LineageDeliveryV1 {
303            digest: format!("sha256:{hex}"),
304            verified: false,
305            error: Some(match error {
306                LoadError::Missing => "missing",
307                LoadError::Unreadable => "unreadable",
308                LoadError::Tampered => "tampered",
309            }),
310            summary: None,
311            egress: None,
312        },
313    }
314}
315
316fn summarize(receipt: &ContextDecisionReceiptV1) -> DeliverySummaryV1 {
317    let name = |value: serde_json::Value| value.as_str().map(str::to_owned).unwrap_or_default();
318    let destination = &receipt.destination;
319    let is_egress = destination.provider.as_str() != "mcp-host";
320    DeliverySummaryV1 {
321        outcome: if is_egress
322            && receipt.outcome == lean_ctx_protocol::context_gateway::DeliveryOutcomeV1::Delivered
323        {
324            "prepared".to_owned()
325        } else {
326            name(serde_json::to_value(receipt.outcome).unwrap_or_default())
327        },
328        delivery_semantics: if is_egress {
329            "prepared_for_dispatch"
330        } else {
331            "returned_to_host"
332        },
333        destination: destination.model.as_ref().map_or_else(
334            || destination.provider.as_str().to_owned(),
335            |model| format!("{} / {}", destination.provider.as_str(), model.as_str()),
336        ),
337        policy_digest: receipt
338            .policy
339            .as_ref()
340            .map(|policy| policy.digest.as_str().to_owned()),
341        inspected: receipt.sources.inspected,
342        delivered: receipt.sources.selected,
343        withheld: receipt.sources.blocked,
344        redactions: receipt.security.redactions,
345        tokens_original: receipt.tokens.original,
346        tokens_delivered: receipt.tokens.delivered,
347        final_context: receipt
348            .final_context
349            .as_ref()
350            .map(|digest| digest.as_str().to_owned()),
351    }
352}
353
354/// The scope the ledger's task envelopes name, checked against `scope`.
355///
356/// The ledger is shared by every project on this machine and keyed by task id
357/// alone; events after a `TaskStarted` carry no envelope of their own. So every
358/// envelope is checked: one that names another scope means events of two
359/// tasks may be mixed, and none of them is shown.
360pub(crate) fn ledger_scope(events: &[ExecutionEvent], scope: &str) -> LedgerScope {
361    if events.is_empty() {
362        return LedgerScope::Verified;
363    }
364    let mut verdict = LedgerScope::Unverified;
365    for envelope_ref in events.iter().filter_map(|event| match event {
366        ExecutionEvent::TaskStarted { envelope_ref, .. } => Some(envelope_ref.as_str()),
367        _ => None,
368    }) {
369        let envelope = envelope_ref
370            .strip_prefix("sha256:")
371            .and_then(|hex| {
372                crate::core::engine_artifact::read_content("execution/evidence", hex, "json").ok()
373            })
374            .and_then(|bytes| {
375                serde_json::from_slice::<lean_ctx_protocol::TaskEnvelopeV1>(&bytes).ok()
376            });
377        match envelope {
378            Some(envelope)
379                if super::task_scope(envelope.tenant_id.as_ref(), &envelope.project_id)
380                    == scope =>
381            {
382                if verdict == LedgerScope::Unverified {
383                    verdict = LedgerScope::Verified;
384                }
385            }
386            Some(_) => return LedgerScope::Foreign,
387            // One unproven envelope leaves the whole history unproven.
388            None => return LedgerScope::Unverified,
389        }
390    }
391    verdict
392}
393
394/// Load a task's lineage within one tenant/project scope from this machine's
395/// stores.
396pub fn load(scope: &str, task_id: &str) -> TaskLineageV1 {
397    let events = crate::core::execution_ledger::ExecutionLedgerStore::from_default()
398        .and_then(|store| store.by_task_verified(task_id))
399        .map_err(|error| error.to_string());
400    let checked = events
401        .as_ref()
402        .map_or(LedgerScope::Verified, |events| ledger_scope(events, scope));
403    let receipts = crate::core::context_admission::receipt_store::for_task(scope, task_id);
404    let mut lineage = build(task_id, events, checked, &receipts);
405    lineage.scope = Some(scope.to_owned());
406    for delivery in &mut lineage.deliveries {
407        if !delivery.verified {
408            continue;
409        }
410        let evidence = lean_ctx_protocol::Sha256Digest::new(delivery.digest.clone())
411            .map_err(|_| "invalid decision receipt digest")
412            .and_then(|digest| {
413                crate::core::context_admission::egress::load_egress_for_decision(&digest)
414            });
415        delivery.egress = match evidence {
416            Ok(None) => None,
417            Ok(Some(view)) => Some(EgressLineageV1 {
418                admission: view.admission.map(|stored| stored.admission),
419                observations: view
420                    .observations
421                    .into_iter()
422                    .map(|stored| stored.observation)
423                    .collect(),
424                error: None,
425            }),
426            Err(error) => Some(EgressLineageV1 {
427                admission: None,
428                observations: Vec::new(),
429                error: Some(error),
430            }),
431        };
432    }
433    if lineage
434        .deliveries
435        .iter()
436        .filter_map(|delivery| delivery.egress.as_ref())
437        .any(|egress| egress.error.is_some() || egress.admission.is_none())
438    {
439        lineage.gaps.push("egress_admission_unverified");
440    }
441    if lineage
442        .deliveries
443        .iter()
444        .filter_map(|delivery| delivery.egress.as_ref())
445        .any(|egress| {
446            egress.admission.as_ref().is_some_and(|admission| {
447                admission.outcome
448                    == lean_ctx_protocol::context_gateway::EgressAdmissionOutcomeV1::Prepared
449            }) && egress.observations.is_empty()
450        })
451    {
452        lineage.gaps.push("no_transport_observation");
453    }
454    lineage
455}
456
457#[cfg(test)]
458#[path = "lineage_tests.rs"]
459mod tests;
lean_ctx/core/context_store/lineage.rs

lean_ctx/core/context_store/
lineage.rs
