lean_ctx/core/execution_ledger/host/
checkpoint.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Canonical continuation from authenticated artifacts, not a parallel state store.
3
4use anyhow::{Result, ensure};
5use base64::{Engine as _, engine::general_purpose::STANDARD};
6use ed25519_dalek::{Signer as _, VerifyingKey};
7use lean_ctx_protocol::{
8    AcceptanceState, AcceptedOutcomeV1, ContextCheckpointV1, ContextCheckpointV2,
9    ContextCheckpointV3, EvidenceKind, EvidenceRefV1, ExecutionPlanV1, ReceiptId, Sha256Digest,
10    SignatureStatus, TaskEnvelopeV1,
11};
12use serde::Deserialize;
13
14use super::{
15    HostReceiptAuthority, ReceiptSignerAdmissionV1, canonical_serialize, digest, outcome, task_lock,
16};
17use crate::core::{
18    context_checkpoint::{
19        VerifiedReceiptDocumentV1, build_checkpoint_v2, verify_checkpoint_v2, verify_checkpoint_v3,
20    },
21    engine_interface::persist_engine_artifact_content,
22    execution_ledger::ExecutionEvent,
23};
24
25/// Bound shared by local lineage and operator-transferred evidence sets.
26pub(super) const MAX_CHECKPOINT_RECEIPTS: usize = 64;
27
28#[derive(Deserialize)]
29#[serde(deny_unknown_fields)]
30pub struct HostCheckpointRequest {
31    pub schema_version: u32,
32    pub checkpoint_json: String,
33    pub receipt_digests: Vec<Sha256Digest>,
34}
35
36impl HostReceiptAuthority {
37    /// Read the just-published receipt through the existing verified lineage owner.
38    /// This is not a historical lookup or a caller-supplied artifact-path API.
39    pub fn published_receipt_json(
40        &self,
41        publication: &crate::core::execution_ledger::PublishedCanonicalReceipt,
42    ) -> Result<String> {
43        self.validate_current().map_err(anyhow::Error::msg)?;
44        let digest = Sha256Digest::new(publication.receipt_digest.clone())?;
45        // Validate the supplied publication's canonical location before the
46        // common lineage reader admits its signature and ledger membership.
47        let (_, original) = publication.read_canonical()?;
48        self.with_checkpoint_lineage(std::slice::from_ref(&digest), |_, _, receipts, _| {
49            let receipt = receipts
50                .first()
51                .ok_or_else(|| anyhow::anyhow!("receipt missing"))?;
52            ensure!(
53                receipt.canonical_bytes() == original,
54                "receipt bytes changed"
55            );
56            self.validate_current().map_err(anyhow::Error::msg)?;
57            String::from_utf8(original).map_err(Into::into)
58        })
59    }
60
61    pub fn export_checkpoint(&self, request: &HostCheckpointRequest) -> Result<serde_json::Value> {
62        self.validate_current().map_err(anyhow::Error::msg)?;
63        ensure!(
64            self.allow_checkpoint_signing,
65            "checkpoint signing not authorized"
66        );
67        ensure!(
68            matches!(request.schema_version, 1..=3)
69                && !request.receipt_digests.is_empty()
70                && request.receipt_digests.len() <= MAX_CHECKPOINT_RECEIPTS,
71            "invalid checkpoint request"
72        );
73        ensure!(
74            crate::core::secret_detection::detect_secrets(&request.checkpoint_json).is_empty(),
75            "checkpoint contains credential-shaped material"
76        );
77        // G08: the caller's checkpoint is not bound yet — digest and signature
78        // are computed below, by this host. So its sources are admitted by
79        // the Context Gateway FIRST, in every mode, and what is verified,
80        // digested, signed, stored and returned is the ADMITTED checkpoint:
81        // a value the gateway masks is masked in it, one it withholds
82        // refuses the request. Unchanged sources leave the caller's canonical
83        // bytes exactly as they were. A masked checkpoint is re-encoded by
84        // its own type; a mask its text type cannot carry is a refusal. The
85        // answer says so (`gateway_redacted`), next to the checkpoint it
86        // returns and the digest of that checkpoint.
87        let mut admitted: serde_json::Value = serde_json::from_str(&request.checkpoint_json)?;
88        let redacted = super::checkpoint_transfer::admit_fresh_checkpoint(&mut admitted)?;
89        let checkpoint_json = if redacted {
90            let canonical = match request.schema_version {
91                3 => serde_json::from_value::<ContextCheckpointV3>(admitted)?.canonical_bytes()?,
92                1 => serde_json::from_value::<ContextCheckpointV1>(admitted)?.canonical_bytes()?,
93                _ => serde_json::from_value::<ContextCheckpointV2>(admitted)?.canonical_bytes()?,
94            };
95            String::from_utf8(canonical)?
96        } else {
97            request.checkpoint_json.clone()
98        };
99        let checkpoint_json = checkpoint_json.as_bytes();
100        self.with_checkpoint_lineage(
101            &request.receipt_digests,
102            |task, plan, receipts, outcomes| {
103                let (checkpoint, checkpoint_digest, payload, envelope_version, result_version) =
104                    if request.schema_version == 3 {
105                        let checkpoint =
106                            ContextCheckpointV3::from_canonical_bytes(checkpoint_json)?;
107                        let verified =
108                            verify_checkpoint_v3(checkpoint, task, Some(plan), receipts, outcomes)?;
109                        (
110                            serde_json::to_value(verified.checkpoint())?,
111                            verified.checkpoint().digest()?,
112                            verified.signing_bytes()?,
113                            "leanctx.host-checkpoint/v2",
114                            "leanctx.host-checkpoint-result/v2",
115                        )
116                    } else {
117                        let verified = if request.schema_version == 1 {
118                            let checkpoint =
119                                ContextCheckpointV1::from_canonical_bytes(checkpoint_json)?;
120                            build_checkpoint_v2(&checkpoint, task, Some(plan), receipts, outcomes)?
121                        } else {
122                            // Reauthorize exact continued V2 against this host's artifacts;
123                            // never demote it or treat a package signature as admission.
124                            let checkpoint =
125                                ContextCheckpointV2::from_canonical_bytes(checkpoint_json)?;
126                            verify_checkpoint_v2(checkpoint, task, Some(plan), receipts, outcomes)?
127                        };
128                        (
129                            serde_json::to_value(verified.checkpoint())?,
130                            verified.digest()?,
131                            verified.signing_payload()?.bytes().to_vec(),
132                            "leanctx.host-checkpoint/v1",
133                            "leanctx.host-checkpoint-result/v1",
134                        )
135                    };
136                self.validate_current().map_err(anyhow::Error::msg)?;
137                let envelope = serde_json::json!({
138                    "schema_version":envelope_version,
139                    "checkpoint":checkpoint,
140                    "signer":{"key_id":self.signer_admission.key_id,
141                        "public_key_digest":self.signer_admission.public_key_digest},
142                    "signature":STANDARD.encode(self.signing_key.sign(&payload).to_bytes()),
143                });
144                let bytes = canonical_serialize(&envelope);
145                ensure!(
146                    bytes.len() <= 1024 * 1024,
147                    "checkpoint artifact exceeds read bound"
148                );
149                let hash = digest(&bytes).map_err(anyhow::Error::msg)?;
150                drop(
151                    persist_engine_artifact_content(
152                        "execution/checkpoints",
153                        hash.hex(),
154                        "json",
155                        &bytes,
156                    )
157                    .map_err(anyhow::Error::msg)?,
158                );
159                let mut result = serde_json::json!({"schema_version":result_version,
160                "artifact_digest":hash,"checkpoint_digest":checkpoint_digest,
161                "artifact":envelope,"session_adopted":false});
162                if redacted {
163                    // Only then: an unchanged checkpoint keeps the answer's
164                    // shape byte for byte.
165                    result["gateway_redacted"] = serde_json::Value::Bool(true);
166                }
167                Ok(result)
168            },
169        )
170    }
171
172    /// Keep one receipt/head/lineage authority and retain its lock through the consumer.
173    pub(super) fn with_checkpoint_lineage<T>(
174        &self,
175        hashes: &[Sha256Digest],
176        consume: impl FnOnce(
177            &TaskEnvelopeV1,
178            &ExecutionPlanV1,
179            &[VerifiedReceiptDocumentV1],
180            &[AcceptedOutcomeV1],
181        ) -> Result<T>,
182    ) -> Result<T> {
183        let receipts = verify_receipts(
184            hashes,
185            &self.signer_admission,
186            &self.signing_key.verifying_key(),
187            &|hash| Ok(outcome::publication(hash)?.read_canonical()?.1),
188        )?;
189        let first = receipts
190            .first()
191            .ok_or_else(|| anyhow::anyhow!("missing receipt"))?;
192        let (task, plan) = read_task_and_plan(first, &LocalCheckpointEvidence)?;
193        let _lock = task_lock(&task).map_err(anyhow::Error::msg)?;
194        let events = self.ledger.by_task_verified(task.task_id.as_str())?;
195        for receipt in &receipts {
196            ensure!(
197                events.iter().any(|event| matches!(event,
198                ExecutionEvent::CanonicalReceiptRecorded { receipt_id, receipt_digest, .. }
199                if receipt_id == receipt.receipt_id().as_str()
200                    && receipt_digest == receipt.canonical_digest().as_str())),
201                "receipt not recorded by this host"
202            );
203        }
204        let head = self
205            .ledger
206            .canonical_receipt_for_task_verified(task.task_id.as_str())?
207            .ok_or_else(|| anyhow::anyhow!("host receipt missing"))?;
208        ensure!(
209            receipts
210                .iter()
211                .any(|receipt| receipt.canonical_digest().as_str() == head.receipt_digest),
212            "checkpoint omits the current receipt"
213        );
214        let outcomes = reconstruct_outcomes(&plan, &receipts, &LocalCheckpointEvidence)?;
215        consume(&task, &plan, &receipts, &outcomes)
216    }
217}
218
219/// Digest-addressed evidence the checkpoint verifier requires.
220///
221/// One rule set serves this host's artifact store and an operator-transferred
222/// package, so a transferred package can never satisfy a weaker evidence
223/// requirement than a local resume.
224pub(super) trait CheckpointEvidence {
225    fn canonical_bytes(&self, expected: &Sha256Digest) -> Result<Vec<u8>>;
226}
227
228/// This host's already admitted evidence artifacts.
229pub(super) struct LocalCheckpointEvidence;
230
231impl CheckpointEvidence for LocalCheckpointEvidence {
232    fn canonical_bytes(&self, expected: &Sha256Digest) -> Result<Vec<u8>> {
233        outcome::read_bytes(expected)
234    }
235}
236
237fn read_evidence_value<T: serde::de::DeserializeOwned>(
238    evidence: &dyn CheckpointEvidence,
239    expected: &Sha256Digest,
240) -> Result<T> {
241    Ok(serde_json::from_slice(
242        &evidence.canonical_bytes(expected)?,
243    )?)
244}
245
246/// Verify every receipt in the requested set against one admitted signer.
247/// The reader supplies bytes; the signature, identity and admission checks are
248/// the protocol verifier's, never the reader's.
249pub(super) fn verify_receipts(
250    hashes: &[Sha256Digest],
251    admission: &ReceiptSignerAdmissionV1,
252    verifying_key: &VerifyingKey,
253    read: &dyn Fn(&Sha256Digest) -> Result<Vec<u8>>,
254) -> Result<Vec<VerifiedReceiptDocumentV1>> {
255    ensure!(
256        !hashes.is_empty() && hashes.len() <= MAX_CHECKPOINT_RECEIPTS,
257        "invalid receipt set"
258    );
259    let mut receipts = Vec::with_capacity(hashes.len());
260    for hash in hashes {
261        receipts.push(VerifiedReceiptDocumentV1::from_canonical_bytes(
262            &read(hash)?,
263            admission,
264            verifying_key,
265        )?);
266    }
267    Ok(receipts)
268}
269
270/// Read the exact task and plan the first verified receipt binds.
271pub(super) fn read_task_and_plan(
272    first: &VerifiedReceiptDocumentV1,
273    evidence: &dyn CheckpointEvidence,
274) -> Result<(TaskEnvelopeV1, ExecutionPlanV1)> {
275    let task: TaskEnvelopeV1 = read_evidence_value(evidence, &first.document().lineage.task_ref)?;
276    let plan: ExecutionPlanV1 = read_evidence_value(evidence, &first.document().lineage.plan_ref)?;
277    Ok((task, plan))
278}
279
280/// Reconstruct the immutable outcome snapshots the signed receipts bind.
281pub(super) fn reconstruct_outcomes(
282    plan: &ExecutionPlanV1,
283    receipts: &[VerifiedReceiptDocumentV1],
284    evidence: &dyn CheckpointEvidence,
285) -> Result<Vec<AcceptedOutcomeV1>> {
286    let mut outcomes = Vec::new();
287    for receipt in receipts {
288        let document = receipt.document();
289        if document.outcome.state == AcceptanceState::Unknown {
290            continue;
291        }
292        let reference = document
293            .outcome
294            .outcome_ref
295            .as_ref()
296            .ok_or_else(|| anyhow::anyhow!("outcome reference missing"))?;
297        // The signed receipt binds this immutable evaluation snapshot. Only
298        // cyclic links added by its owner after signing are reconstructed.
299        let mut value: AcceptedOutcomeV1 = read_evidence_value(evidence, reference)?;
300        ensure!(
301            value.outcome_id.as_str()
302                == document
303                    .outcome
304                    .outcome_id
305                    .as_ref()
306                    .ok_or_else(|| anyhow::anyhow!("outcome identity missing"))?
307                    .as_str()
308                && value.accepted == document.outcome.state
309                && value.receipt_id.is_none()
310                && value.plan_id.is_none(),
311            "unsupported outcome snapshot"
312        );
313        value.plan_id = Some(plan.plan_id.clone());
314        value.receipt_id = Some(ReceiptId::new(document.receipt_id.as_str())?);
315        value.evidence_refs.push(EvidenceRefV1 {
316            schema_version: Some(1),
317            kind: EvidenceKind::QualityMeasurement,
318            uri: format!("artifact://execution/evidence/{}", reference.hex()),
319            digest: reference.as_str().into(),
320            signature_status: SignatureStatus::NotSigned,
321            media_type: Some("application/json".into()),
322            extensions: Default::default(),
323        });
324        outcomes.push(value);
325    }
326    Ok(outcomes)
327}
lean_ctx/core/execution_ledger/host/checkpoint.rs

lean_ctx/core/execution_ledger/host/
checkpoint.rs
