lean_ctx/core/execution_ledger/host/
checkpoint_resume.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Host-admitted reader adapter; the complete signed source remains recoverable.
3
4use anyhow::{Result, ensure};
5use base64::{Engine as _, engine::general_purpose::STANDARD};
6use ed25519_dalek::Signature;
7use lean_ctx_protocol::{
8    CONTEXT_CHECKPOINT_V2_SIGNATURE_DOMAIN, CONTEXT_CHECKPOINT_V3_SIGNATURE_DOMAIN,
9    ContextCheckpointDecisionStatusV1, ContextCheckpointV2, ContextCheckpointV3, ProjectId,
10    Sha256Digest, TenantId, WorkspaceId,
11};
12use serde::{Deserialize, Serialize};
13use std::path::PathBuf;
14
15use super::{HostReceiptAuthority, canonical_serialize, digest};
16use crate::core::{
17    context_checkpoint::{verify_checkpoint_v2, verify_checkpoint_v3},
18    engine_artifact,
19    session::{
20        Decision, EvidenceKind, EvidenceRecord, Finding, ProgressEntry, SessionState, TaskInfo,
21    },
22};
23
24#[derive(Deserialize)]
25#[serde(deny_unknown_fields)]
26pub struct HostCheckpointResumeRequest {
27    schema_version: u32,
28    artifact_digest: Sha256Digest,
29}
30
31fn default_canonical_admission() -> bool {
32    true
33}
34
35/// The operator, never the checkpoint or request, chooses the receiving scope.
36#[derive(Deserialize)]
37#[serde(deny_unknown_fields)]
38pub(super) struct CheckpointResumeTarget {
39    project_root: PathBuf,
40    project_id: ProjectId,
41    tenant_id: TenantId,
42    workspace_id: WorkspaceId,
43    #[serde(default = "default_canonical_admission")]
44    canonical: bool,
45}
46
47#[derive(Deserialize, Serialize)]
48#[serde(deny_unknown_fields)]
49pub(super) struct Envelope {
50    pub(super) schema_version: String,
51    pub(super) checkpoint: ContextCheckpointV2,
52    pub(super) signer: Signer,
53    pub(super) signature: String,
54}
55
56#[derive(Deserialize, Serialize)]
57#[serde(deny_unknown_fields)]
58pub(super) struct UnicodeEnvelope {
59    pub(super) schema_version: String,
60    pub(super) checkpoint: ContextCheckpointV3,
61    pub(super) signer: Signer,
62    pub(super) signature: String,
63}
64
65#[derive(Deserialize, Serialize)]
66#[serde(deny_unknown_fields)]
67pub(super) struct Signer {
68    pub(super) key_id: String,
69    pub(super) public_key_digest: Sha256Digest,
70}
71
72impl HostReceiptAuthority {
73    pub fn resume_checkpoint(
74        &self,
75        request: &HostCheckpointResumeRequest,
76    ) -> Result<serde_json::Value> {
77        self.validate_current().map_err(anyhow::Error::msg)?;
78        ensure!(request.schema_version == 1, "unsupported resume request");
79        let target = self
80            .checkpoint_resume
81            .as_ref()
82            .ok_or_else(|| anyhow::anyhow!("checkpoint resume not authorized"))?;
83        ensure!(
84            target.project_root.is_absolute(),
85            "absolute receiving root required"
86        );
87        let root = std::fs::canonicalize(&target.project_root)?;
88        ensure!(
89            root.is_dir() && !crate::core::pathutil::is_broad_or_unsafe_root(&root),
90            "unsafe receiving root"
91        );
92        let root = root
93            .to_str()
94            .ok_or_else(|| anyhow::anyhow!("receiving root is not UTF-8"))?;
95        let bytes = engine_artifact::read_content(
96            "execution/checkpoints",
97            request.artifact_digest.hex(),
98            "json",
99        )
100        .map_err(anyhow::Error::msg)?;
101        ensure!(
102            digest(&bytes).map_err(anyhow::Error::msg)? == request.artifact_digest,
103            "checkpoint artifact digest mismatch"
104        );
105        let envelope_schema = serde_json::from_slice::<serde_json::Value>(&bytes)?
106            .get("schema_version")
107            .and_then(serde_json::Value::as_str)
108            .unwrap_or_default()
109            .to_owned();
110        if envelope_schema == "leanctx.host-checkpoint/v2" {
111            return self.resume_unicode_checkpoint(request, target, root, &bytes);
112        }
113        let envelope: Envelope = serde_json::from_slice(&bytes)?;
114        ensure!(
115            envelope.schema_version == "leanctx.host-checkpoint/v1"
116                && canonical_serialize(&envelope) == bytes,
117            "unsupported or noncanonical checkpoint envelope"
118        );
119        ensure!(
120            envelope.signer.key_id == self.signer_admission.key_id
121                && envelope.signer.public_key_digest == self.signer_admission.public_key_digest,
122            "checkpoint signer not admitted"
123        );
124        let checkpoint = &envelope.checkpoint;
125        ensure!(
126            checkpoint.lineage.project_id == target.project_id
127                && checkpoint.lineage.tenant_id == target.tenant_id
128                && checkpoint.lineage.workspace_id == target.workspace_id,
129            "checkpoint receiving scope mismatch"
130        );
131        let mut payload = CONTEXT_CHECKPOINT_V2_SIGNATURE_DOMAIN.to_vec();
132        payload.extend(checkpoint.canonical_bytes()?);
133        self.signing_key.verifying_key().verify_strict(
134            &payload,
135            &Signature::from_slice(&STANDARD.decode(&envelope.signature)?)?,
136        )?;
137        self.with_checkpoint_lineage(&checkpoint.lineage.artifact_lineage.receipt_refs, |task, plan, receipts, outcomes| {
138            let verified = verify_checkpoint_v2(checkpoint.clone(), task, Some(plan), receipts, outcomes)?;
139            self.validate_current().map_err(anyhow::Error::msg)?;
140            let mut session = project_session(verified.checkpoint(), root, &bytes, &request.artifact_digest)?;
141            if target.canonical {
142                session.canonical_checkpoint = Some(Box::new(crate::core::context_checkpoint::CanonicalSession::admitted(
143                    &verified, request.artifact_digest.clone(), root.into(),
144                )?));
145            }
146            session.save_new().map_err(|error| anyhow::anyhow!("{error}; recovery candidate: {}", session.id))?;
147            Ok(serde_json::json!({
148                "schema_version":"leanctx.host-checkpoint-resume-result/v1",
149                "session_id":session.id, "artifact_digest":request.artifact_digest,
150                "checkpoint_digest":verified.digest()?, "legacy_projection_created":true,
151                "canonical_session_adopted":target.canonical, "source_preserved":true,
152                "projection_only_fields":["task_progress_percentage","accepted_decisions","findings","next_steps"],
153                "source_only_fields":["identity","lineage","task_status","decision_status_and_evidence",
154                    "exact_progress","files","profile_id","policy_pins","package_pins","session_state","learning_state","carrier","encryption_metadata"]
155            }))
156        })
157    }
158
159    fn resume_unicode_checkpoint(
160        &self,
161        request: &HostCheckpointResumeRequest,
162        target: &CheckpointResumeTarget,
163        root: &str,
164        bytes: &[u8],
165    ) -> Result<serde_json::Value> {
166        let envelope: UnicodeEnvelope = serde_json::from_slice(bytes)?;
167        ensure!(
168            envelope.schema_version == "leanctx.host-checkpoint/v2"
169                && canonical_serialize(&envelope) == bytes,
170            "unsupported or noncanonical Unicode checkpoint envelope"
171        );
172        ensure!(
173            envelope.signer.key_id == self.signer_admission.key_id
174                && envelope.signer.public_key_digest == self.signer_admission.public_key_digest,
175            "checkpoint signer not admitted"
176        );
177        let checkpoint = &envelope.checkpoint;
178        ensure!(
179            checkpoint.lineage.project_id == target.project_id
180                && checkpoint.lineage.tenant_id == target.tenant_id
181                && checkpoint.lineage.workspace_id == target.workspace_id,
182            "checkpoint receiving scope mismatch"
183        );
184        let mut payload = CONTEXT_CHECKPOINT_V3_SIGNATURE_DOMAIN.to_vec();
185        payload.extend(checkpoint.canonical_bytes()?);
186        self.signing_key.verifying_key().verify_strict(
187            &payload,
188            &Signature::from_slice(&STANDARD.decode(&envelope.signature)?)?,
189        )?;
190        self.with_checkpoint_lineage(
191            &checkpoint.lineage.artifact_lineage.receipt_refs,
192            |task, plan, receipts, outcomes| {
193                let verified =
194                    verify_checkpoint_v3(checkpoint.clone(), task, Some(plan), receipts, outcomes)?;
195                self.validate_current().map_err(anyhow::Error::msg)?;
196                let mut session = project_session_v3(
197                    verified.checkpoint(),
198                    root,
199                    bytes,
200                    &request.artifact_digest,
201                )?;
202                if target.canonical {
203                    session.canonical_checkpoint = Some(Box::new(
204                        crate::core::context_checkpoint::CanonicalSession::admitted_v3(
205                            &verified,
206                            request.artifact_digest.clone(),
207                            root.into(),
208                        )?,
209                    ));
210                }
211                session.save_new().map_err(|error| {
212                    anyhow::anyhow!("{error}; recovery candidate: {}", session.id)
213                })?;
214                Ok(serde_json::json!({
215                    "schema_version":"leanctx.host-checkpoint-resume-result/v2",
216                    "session_id":session.id, "artifact_digest":request.artifact_digest,
217                    "checkpoint_digest":verified.checkpoint().digest()?,
218                    "legacy_projection_created":true,
219                    "canonical_session_adopted":target.canonical, "source_preserved":true,
220                    "projection_only_fields":["task_progress_percentage","accepted_decisions","findings","next_steps"],
221                    "source_only_fields":["identity","lineage","task_status","decision_status_and_evidence",
222                        "exact_progress","files","profile_id","policy_pins","package_pins","session_state","learning_state","carrier","encryption_metadata"]
223                }))
224            },
225        )
226    }
227}
228
229/// A compatibility view is not a demoted checkpoint or an authority for pins.
230pub(super) fn project_session(
231    checkpoint: &ContextCheckpointV2,
232    root: &str,
233    source: &[u8],
234    source_digest: &Sha256Digest,
235) -> Result<SessionState> {
236    let mut session = SessionState::new();
237    session.project_root = Some(root.into());
238    let live = &checkpoint.live_state;
239    session.task = Some(TaskInfo {
240        description: live.task.title.as_str().into(),
241        intent: None,
242        progress_pct: if live.progress.total_steps == 0 {
243            None
244        } else {
245            Some(u8::try_from(
246                (u64::from(live.progress.completed_steps) * 100)
247                    / u64::from(live.progress.total_steps),
248            )?)
249        },
250    });
251    let timestamp = session.started_at;
252    session.findings = live
253        .findings
254        .iter()
255        .map(|finding| Finding {
256            file: None,
257            line: None,
258            summary: finding.as_str().into(),
259            timestamp,
260        })
261        .collect();
262    session.decisions = live
263        .decisions
264        .iter()
265        .filter(|decision| decision.status == ContextCheckpointDecisionStatusV1::Accepted)
266        .map(|decision| Decision {
267            summary: decision.statement.as_str().into(),
268            rationale: Some(decision.rationale.as_str().into()),
269            timestamp,
270        })
271        .collect();
272    session.next_steps = live
273        .next_steps
274        .iter()
275        .map(|step| step.as_str().into())
276        .collect();
277    session.progress.push(ProgressEntry {
278        action: "checkpoint_resume".into(),
279        detail: Some(live.handoff_summary.as_str().into()),
280        timestamp,
281    });
282    // Retain the complete, exact signed envelope, including fields with no legacy
283    // display equivalent. No observation time, path, grant or policy is inferred.
284    session.evidence.push(EvidenceRecord {
285        kind: EvidenceKind::Manual,
286        key: "canonical_checkpoint_source".into(),
287        value: Some(
288            serde_json::json!({"schema_version":1,"artifact_digest":source_digest,
289            "envelope_json":std::str::from_utf8(source)?,"projection_only":true})
290            .to_string(),
291        ),
292        tool: None,
293        input_md5: None,
294        output_md5: None,
295        agent_id: None,
296        client_name: None,
297        task_id: Some(checkpoint.lineage.task_id.as_str().into()),
298        timestamp,
299    });
300    Ok(session)
301}
302
303/// Project Unicode V3 into the same compatibility view while retaining the
304/// complete V3 value in the existing canonical session envelope.
305pub(super) fn project_session_v3(
306    checkpoint: &ContextCheckpointV3,
307    root: &str,
308    source: &[u8],
309    source_digest: &Sha256Digest,
310) -> Result<SessionState> {
311    let mut session = SessionState::new();
312    session.project_root = Some(root.into());
313    let live = &checkpoint.live_state;
314    session.task = Some(TaskInfo {
315        description: live.task.title.as_str().into(),
316        intent: None,
317        progress_pct: Some(u8::try_from(
318            (u64::from(live.progress.completed_steps) * 100) / u64::from(live.progress.total_steps),
319        )?),
320    });
321    let timestamp = session.started_at;
322    session.findings = live
323        .findings
324        .iter()
325        .map(|finding| Finding {
326            file: None,
327            line: None,
328            summary: finding.as_str().into(),
329            timestamp,
330        })
331        .collect();
332    session.decisions = live
333        .decisions
334        .iter()
335        .filter(|decision| decision.status == ContextCheckpointDecisionStatusV1::Accepted)
336        .map(|decision| Decision {
337            summary: decision.statement.as_str().into(),
338            rationale: Some(decision.rationale.as_str().into()),
339            timestamp,
340        })
341        .collect();
342    session.next_steps = live
343        .next_steps
344        .iter()
345        .map(|step| step.as_str().into())
346        .collect();
347    session.progress.push(ProgressEntry {
348        action: "checkpoint_resume".into(),
349        detail: Some(live.handoff_summary.as_str().into()),
350        timestamp,
351    });
352    session.evidence.push(EvidenceRecord {
353        kind: EvidenceKind::Manual,
354        key: "canonical_checkpoint_source".into(),
355        value: Some(
356            serde_json::json!({"schema_version":2,"artifact_digest":source_digest,
357            "envelope_json":std::str::from_utf8(source)?,"projection_only":true})
358            .to_string(),
359        ),
360        tool: None,
361        input_md5: None,
362        output_md5: None,
363        agent_id: None,
364        client_name: None,
365        task_id: Some(checkpoint.lineage.task_id.as_str().into()),
366        timestamp,
367    });
368    Ok(session)
369}
lean_ctx/core/execution_ledger/host/checkpoint_resume.rs

lean_ctx/core/execution_ledger/host/
checkpoint_resume.rs
