lean_ctx/core/engine_interface/
source_execution.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Execute an admitted source snapshot through the existing native and receipt authorities.
4
5use std::path::Path;
6
7use lean_ctx_protocol::{
8    EngineContextSourceMaterializationRequestV1, EngineContextSourcePlanResponseV1,
9    EngineInvocationIdV1, EngineInvocationV1, EngineObservationV1, EnginePolicyAdmissionV1,
10    EnginePolicyDecisionV1, ExecutionPlanV1, ProtocolReference, TaskEnvelopeV1,
11};
12
13use super::{
14    ENGINE_TRANSPORT_POLICY_REF, EngineTransportView, NativeContextEngine,
15    NativeContextEngineRequest, bind_transport_root, context_plan, planning, sha256_digest,
16    verified_output_view,
17};
18use crate::core::{
19    canonical::canonical_serialize,
20    execution_ledger::{PublishedCanonicalReceipt, host::HostReceiptAuthority},
21};
22
23pub struct SourceExecutionResult {
24    pub source_plan: EngineContextSourcePlanResponseV1,
25    pub execution_plan: ExecutionPlanV1,
26    pub invocation: EngineInvocationV1,
27    pub observation: EngineObservationV1,
28    pub view: EngineTransportView,
29    pub receipt: PublishedCanonicalReceipt,
30}
31
32/// A caller declaration is not a scheduler recommendation or source authorization.
33/// The immutable kernel decision supplies the context binding in this process.
34pub fn execute(
35    root: &Path,
36    task: &TaskEnvelopeV1,
37    declared_plan: &ExecutionPlanV1,
38    request: &EngineContextSourceMaterializationRequestV1,
39    authority: &HostReceiptAuthority,
40) -> Result<SourceExecutionResult, &'static str> {
41    authority.require_context_decision_signing()?;
42    let root = bind_transport_root(root).map_err(|_| "unsafe_root")?;
43    let admission = EnginePolicyAdmissionV1 {
44        policy_ref: reference(ENGINE_TRANSPORT_POLICY_REF)?,
45        decision: EnginePolicyDecisionV1::Admitted,
46    };
47    planning::validate_native_plan(task, declared_plan, &admission)
48        .map_err(|_| "invalid_host_plan")?;
49    if declared_plan.provider != "local-native"
50        || declared_plan.context_plan_id.is_some()
51        || task.region_policy_ref.is_some()
52        || task.model_policy_ref.is_some()
53        || request.source_plan.planning.task_id != task.task_id
54        || declared_plan.context_token_limit().is_none()
55    {
56        return Err("invalid_host_source_plan");
57    }
58    let (snapshot, decision) = context_plan::materialize_sources_with_decision(&root, request)?;
59    if decision.context_projection() != &snapshot.plan.result.plan
60        || declared_plan.context_token_limit() != Some(snapshot.plan.result.plan.budget_tokens)
61    {
62        return Err("host_source_decision_mismatch");
63    }
64    let mut plan = declared_plan.clone();
65    plan.context_plan_id = Some(snapshot.plan.result.plan.context_plan_id.clone());
66    let plan = decision
67        .bind_execution_plan(plan)
68        .map_err(|_| "host_source_decision_mismatch")?;
69    planning::validate_native_plan(task, &plan, &admission)
70        .map_err(|_| "invalid_host_source_plan")?;
71    let engine = NativeContextEngine::with_root(&root).map_err(|_| "unsafe_root")?;
72    let source_plan_digest = sha256_digest(&canonical_serialize(&snapshot.plan))
73        .map_err(|_| "host_source_evidence_unavailable")?;
74    let input_ref = reference(&format!(
75        "input:source-materialization-sha256:{}",
76        snapshot.materialized_digest.hex()
77    ))?;
78    let mut source_refs = vec![
79        input_ref.clone(),
80        reference(&format!(
81            "artifact://execution/evidence/{}",
82            source_plan_digest.hex()
83        ))?,
84    ];
85    // Keep invocation lineage bounded: the persisted descriptor-only source
86    // plan above binds every selected source without one ref per source.
87    source_refs
88        .extend(planning::binding_refs(task, &plan).map_err(|_| "invalid_host_source_plan")?);
89    let identity = sha256_digest(&canonical_serialize(&(
90        "source-execution-v1",
91        &plan,
92        &source_refs,
93        &snapshot.materialized_digest,
94    )))
95    .map_err(|_| "host_source_evidence_unavailable")?;
96    let native = NativeContextEngineRequest {
97        invocation_id: EngineInvocationIdV1::new(format!(
98            "engine-invocation-{}",
99            &identity.hex()[..32]
100        ))
101        .map_err(|_| "invalid_host_source_plan")?,
102        input_ref,
103        input_digest: snapshot.materialized_digest.clone(),
104        source_refs,
105        policy_admission: admission,
106        // No invented filesystem identity: the adapter consumes the bound bytes directly.
107        paths: vec![],
108        mode: "aggressive".into(),
109        budget_tokens: plan.context_token_limit(),
110        timeout_ms: task.latency_budget_ms.unwrap_or(30_000).min(30_000),
111    };
112    // Validate all record bounds before durable intent or capability execution.
113    let expected_invocation = engine
114        .invocation_for(&native)
115        .map_err(|_| "invalid_host_source_plan")?;
116    if let Some(recorded) =
117        authority.replay_source_context(task, &plan, &expected_invocation, &decision)?
118    {
119        let view = verified_output_view(&recorded.invocation, &recorded.observation)
120            .map_err(|_| "host_engine_evidence_unavailable")?;
121        // A stored signature is not a current source/policy authorization lease.
122        if context_plan::materialize(&root, request)? != snapshot {
123            return Err("source_plan_changed");
124        }
125        // The disclosed view derives from the materialized context only.
126        context_plan::materialize_impl::record_derived_delivery(
127            &root,
128            &snapshot.content,
129            &view.text,
130        );
131        return Ok(SourceExecutionResult {
132            source_plan: snapshot.plan,
133            execution_plan: plan,
134            invocation: recorded.invocation,
135            observation: recorded.observation,
136            view,
137            receipt: recorded.published_receipt,
138        });
139    }
140    let attempt = authority.begin_with_context(task, &plan, Some(&decision))?;
141    planning::persist_evidence(task).map_err(|_| "host_source_evidence_unavailable")?;
142    planning::persist_evidence(&plan).map_err(|_| "host_source_evidence_unavailable")?;
143    planning::persist_evidence(&snapshot.plan).map_err(|_| "host_source_evidence_unavailable")?;
144    let (invocation, observation) = engine
145        .execute_materialized(native, &snapshot.content)
146        .map_err(|_| "host_engine_evidence_unavailable")?;
147    let view = verified_output_view(&invocation, &observation)
148        .map_err(|_| "host_engine_evidence_unavailable")?;
149    // Replay is not an authorization lease. Reuse the same policy/materialization
150    // authority immediately before disclosure; a changed snapshot cannot publish.
151    if context_plan::materialize(&root, request)? != snapshot {
152        return Err("source_plan_changed");
153    }
154    let receipt = authority.publish(&attempt, &invocation, &observation, &view.text)?;
155    // The disclosed view derives from the materialized context only.
156    context_plan::materialize_impl::record_derived_delivery(&root, &snapshot.content, &view.text);
157    Ok(SourceExecutionResult {
158        source_plan: snapshot.plan,
159        execution_plan: plan,
160        invocation,
161        observation,
162        view,
163        receipt,
164    })
165}
166
167fn reference(value: &str) -> Result<ProtocolReference, &'static str> {
168    ProtocolReference::new(value).map_err(|_| "invalid_host_source_plan")
169}
lean_ctx/core/engine_interface/source_execution.rs

lean_ctx/core/engine_interface/
source_execution.rs
