lean_ctx_server/server/
native_receipts.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Host-owned native receipt handoff; never populated from MCP arguments.
4
5use std::{
6    path::Path,
7    sync::{
8        Arc, OnceLock,
9        atomic::{AtomicBool, Ordering},
10    },
11};
12
13use lean_ctx_protocol::{EngineInvocationV1, EngineObservationV1, ExecutionPlanV1, TaskEnvelopeV1};
14use rmcp::model::CallToolResult;
15use serde_json::{Map, Value};
16
17use lean_ctx::core::execution_ledger::host::{HostReceiptAttempt, HostReceiptAuthority};
18
19tokio::task_local! {
20    pub(crate) static NATIVE_RECEIPT_CAPTURE: Option<Arc<NativeReceiptCapture>>;
21}
22
23pub(crate) fn current_capture() -> Option<Arc<NativeReceiptCapture>> {
24    NATIVE_RECEIPT_CAPTURE.try_with(Clone::clone).ok().flatten()
25}
26
27/// One immutable lifecycle task, one attempt, one Engine result, one publisher.
28pub(crate) struct NativeReceiptCapture {
29    authority: Arc<HostReceiptAuthority>,
30    task: TaskEnvelopeV1,
31    plan: OnceLock<ExecutionPlanV1>,
32    attempt: OnceLock<HostReceiptAttempt>,
33    execution: OnceLock<(EngineInvocationV1, EngineObservationV1)>,
34    publication_claimed: AtomicBool,
35}
36
37impl NativeReceiptCapture {
38    pub(crate) fn new(authority: Arc<HostReceiptAuthority>, task: TaskEnvelopeV1) -> Self {
39        Self {
40            authority,
41            task,
42            plan: OnceLock::new(),
43            attempt: OnceLock::new(),
44            execution: OnceLock::new(),
45            publication_claimed: AtomicBool::new(false),
46        }
47    }
48
49    pub(crate) fn begin(
50        &self,
51        task: &TaskEnvelopeV1,
52        plan: &ExecutionPlanV1,
53    ) -> Result<(), String> {
54        if task != &self.task || self.plan.set(plan.clone()).is_err() {
55            return Err("mcp_receipt_task_mismatch_or_duplicate".into());
56        }
57        let handoff = lean_ctx::core::context_kernel::bridge::runtime::current_handoff();
58        let context = match &handoff {
59            lean_ctx::core::context_kernel::bridge::runtime::KernelPlanningHandoff::Prepared(
60                prepared,
61            ) => Some(prepared.decision()),
62            _ => None,
63        };
64        let attempt = self.authority.begin_with_context(task, plan, context)?;
65        self.attempt
66            .set(attempt)
67            .map_err(|_| "mcp_receipt_duplicate_attempt".into())
68    }
69
70    pub(crate) fn complete(
71        &self,
72        invocation: EngineInvocationV1,
73        observation: EngineObservationV1,
74    ) -> Result<(), String> {
75        if self.attempt.get().is_none() {
76            return Err("mcp_receipt_attempt_missing".into());
77        }
78        self.execution
79            .set((invocation, observation))
80            .map_err(|_| "mcp_receipt_duplicate_execution".into())
81    }
82
83    pub(crate) fn publish(&self, result: &CallToolResult) -> Result<Value, String> {
84        if self.publication_claimed.swap(true, Ordering::AcqRel) {
85            return Err("mcp_receipt_publication_already_claimed".into());
86        }
87        if result.is_error == Some(true) || result.content.len() != 1 {
88            return Err("mcp_receipt_delivery_unavailable".into());
89        }
90        let text = result.content[0]
91            .as_text()
92            .ok_or("mcp_receipt_delivery_unavailable")?;
93        let attempt = self.attempt.get().ok_or("mcp_receipt_attempt_missing")?;
94        let (invocation, observation) = self
95            .execution
96            .get()
97            .ok_or("mcp_receipt_execution_missing")?;
98        // The shared publisher checks byte equality with the verified Engine view.
99        let published = self
100            .authority
101            .publish(attempt, invocation, observation, &text.text)?;
102        let mut metadata = serde_json::json!({
103            "schema_version":1, "receipt_id":published.receipt_id,
104            "receipt_ref":published.receipt_ref, "receipt_digest":published.receipt_digest,
105            "outcome":"unknown", "delivery":"native_engine_view"
106        });
107        if let Some(decision) = attempt.context_decision() {
108            let bytes = lean_ctx::core::canonical::canonical_serialize(decision);
109            let digest = lean_ctx::core::execution_ledger::host::digest(&bytes)?;
110            metadata["context_decision_ref"] =
111                format!("artifact://execution/evidence/{}", digest.hex()).into();
112        }
113        if let Some(protocol) = self.authority.published_context_protocol(
114            attempt,
115            invocation,
116            observation,
117            &published,
118        )? {
119            // Persist the real evaluator's Unknown observation, not acceptance proof.
120            let bytes = lean_ctx::core::canonical::canonical_serialize(&protocol.accepted_outcome);
121            let digest = lean_ctx::core::execution_ledger::host::digest(&bytes)?;
122            drop(
123                lean_ctx::core::engine_interface::persist_engine_artifact_content(
124                    "execution/evidence",
125                    digest.hex(),
126                    "json",
127                    &bytes,
128                )?,
129            );
130            metadata["outcome_observation_ref"] =
131                format!("artifact://execution/evidence/{}", digest.hex()).into();
132        }
133        Ok(metadata)
134    }
135}
136
137pub(crate) fn requested(name: &str, args: Option<&Map<String, Value>>) -> bool {
138    let Ok((name, args)) = crate::tools::registered::ctx_call::resolve(name, args) else {
139        return false;
140    };
141    name == "ctx_read"
142        && args
143            .as_ref()
144            .and_then(|args| args.get("engine_interface"))
145            .and_then(Value::as_str)
146            == Some("v1")
147}
148
149/// Snapshot explicit host authority once during transport startup, never per request.
150pub fn load_configured_host_authority() -> Result<Option<Arc<HostReceiptAuthority>>, String> {
151    let path = std::env::var_os("LEAN_CTX_RECEIPT_HOST_CONFIG");
152    load_host_authority(path.as_deref().map(Path::new))
153}
154
155/// Read only the explicitly configured host file, before the MCP transport starts.
156/// Never consult tool/session arguments or discover a signing key implicitly.
157pub fn load_host_authority(
158    path: Option<&Path>,
159) -> Result<Option<Arc<HostReceiptAuthority>>, String> {
160    let Some(path) = path else {
161        return Ok(None);
162    };
163    if !path.is_absolute() {
164        return Err("mcp_receipt_host_path_invalid".into());
165    }
166    let mut options = std::fs::OpenOptions::new();
167    options.read(true);
168    #[cfg(unix)]
169    {
170        use std::os::unix::fs::OpenOptionsExt;
171        options.custom_flags(libc::O_NOFOLLOW | libc::O_CLOEXEC | libc::O_NONBLOCK);
172    }
173    #[cfg(windows)]
174    {
175        use std::os::windows::fs::OpenOptionsExt;
176        options.custom_flags(windows_sys::Win32::Storage::FileSystem::FILE_FLAG_OPEN_REPARSE_POINT);
177    }
178    #[cfg(not(any(unix, windows)))]
179    return Err("mcp_receipt_host_platform_unsupported".into());
180
181    let mut file = options
182        .open(path)
183        .map_err(|_| "mcp_receipt_host_unavailable")?;
184    let metadata = file
185        .metadata()
186        .map_err(|_| "mcp_receipt_host_unavailable")?;
187    if !metadata.is_file() || metadata.len() > 16 * 1024 {
188        return Err("mcp_receipt_host_file_invalid".into());
189    }
190    #[cfg(unix)]
191    {
192        use std::os::unix::fs::MetadataExt;
193        // SAFETY: geteuid takes no pointers and only reads the process credential.
194        let effective_uid = unsafe { libc::geteuid() };
195        if metadata.mode() & 0o077 != 0 || metadata.uid() != effective_uid {
196            return Err("mcp_receipt_host_permissions_invalid".into());
197        }
198    }
199    #[cfg(windows)]
200    {
201        use std::os::windows::fs::MetadataExt;
202        if metadata.file_attributes()
203            & windows_sys::Win32::Storage::FileSystem::FILE_ATTRIBUTE_REPARSE_POINT
204            != 0
205        {
206            return Err("mcp_receipt_host_file_invalid".into());
207        }
208    }
209    HostReceiptAuthority::from_reader(&mut file)
210        .map(|host| Some(Arc::new(host)))
211        .map_err(str::to_owned)
212}
213
214#[cfg(test)]
215mod tests {
216    use super::*;
217    use base64::{Engine as _, engine::general_purpose::STANDARD};
218    use ed25519_dalek::{Signature, SigningKey};
219    use lean_ctx_protocol::{AcceptanceState, ReceiptDocumentV1};
220    use sha2::{Digest, Sha256};
221
222    fn host_config(ledger: &Path) -> Vec<u8> {
223        let key = SigningKey::from_bytes(&[53; 32]);
224        let now = chrono::Utc::now();
225        serde_json::to_vec(&serde_json::json!({
226            "schema_version":1,
227            "signing_key_hex":lean_ctx::core::agent_identity::hex_encode(&key.to_bytes()),
228            "signer":{
229                "key_id":"mcp-native-test-key",
230                "public_key_digest":format!("sha256:{}", lean_ctx::core::agent_identity::hex_encode(&Sha256::digest(key.verifying_key().as_bytes()))),
231                "admitted_at":(now-chrono::Duration::days(1)).to_rfc3339_opts(chrono::SecondsFormat::Secs,true),
232                "expires_at":(now+chrono::Duration::days(1)).to_rfc3339_opts(chrono::SecondsFormat::Secs,true),
233                "revoked_at":null
234            },
235            "ledger_path":ledger
236        })).unwrap()
237    }
238
239    fn authority(ledger: &Path) -> Arc<HostReceiptAuthority> {
240        Arc::new(HostReceiptAuthority::from_reader(&mut host_config(ledger).as_slice()).unwrap())
241    }
242
243    #[test]
244    fn receipt_selection_uses_semantic_v1_identity_only() {
245        let args = serde_json::json!({"engine_interface":"v1"})
246            .as_object()
247            .unwrap()
248            .clone();
249        assert!(requested("ctx_read", Some(&args)));
250        assert!(!requested("ctx_read", None));
251        assert!(!requested("ctx_shell", Some(&args)));
252        let wrapped = serde_json::json!({"name":"ctx_read","arguments":args})
253            .as_object()
254            .unwrap()
255            .clone();
256        assert!(requested("ctx_call", Some(&wrapped)));
257    }
258
259    #[test]
260    fn explicit_loader_rejects_unsafe_files_without_discovering_authority() {
261        let _data = lean_ctx::core::data_dir::isolated_data_dir();
262        let directory = tempfile::tempdir().unwrap();
263        let path = directory.path().join("receipt-host.json");
264        let mut options = std::fs::OpenOptions::new();
265        options.create_new(true).write(true);
266        #[cfg(unix)]
267        {
268            use std::os::unix::fs::OpenOptionsExt;
269            options.mode(0o600);
270        }
271        use std::io::Write;
272        options
273            .open(&path)
274            .unwrap()
275            .write_all(&host_config(&directory.path().join("ledger.jsonl")))
276            .unwrap();
277        assert!(load_host_authority(None).unwrap().is_none());
278        assert!(load_host_authority(Some(Path::new("relative.json"))).is_err());
279        assert!(load_host_authority(Some(directory.path())).is_err());
280        assert!(load_host_authority(Some(&path)).unwrap().is_some());
281        let oversized = directory.path().join("oversized.json");
282        std::fs::write(&oversized, vec![b' '; 16 * 1024 + 1]).unwrap();
283        assert_eq!(
284            load_host_authority(Some(&oversized)).err().unwrap(),
285            "mcp_receipt_host_file_invalid"
286        );
287        #[cfg(unix)]
288        {
289            use std::os::unix::{ffi::OsStrExt, fs::PermissionsExt};
290            let fifo = directory.path().join("fifo");
291            let fifo_name = std::ffi::CString::new(fifo.as_os_str().as_bytes()).unwrap();
292            // SAFETY: the live CString is NUL-terminated and names only this fixture's FIFO.
293            let created = unsafe { libc::mkfifo(fifo_name.as_ptr(), 0o600) };
294            assert_eq!(created, 0);
295            assert_eq!(
296                load_host_authority(Some(&fifo)).err().unwrap(),
297                "mcp_receipt_host_file_invalid"
298            );
299            let link = directory.path().join("linked.json");
300            std::os::unix::fs::symlink(&path, &link).unwrap();
301            assert!(load_host_authority(Some(&link)).is_err());
302            std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o644)).unwrap();
303            assert!(load_host_authority(Some(&path)).is_err());
304        }
305    }
306
307    fn request(path: &Path, id: &str) -> rmcp::model::CallToolRequestParams {
308        serde_json::from_value(serde_json::json!({
309            "name":"ctx_read", "arguments":{"path":path,"mode":"aggressive","engine_interface":"v1"},
310            "_meta":{"requestId":id}
311        })).unwrap()
312    }
313
314    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
315    async fn real_mcp_path_publishes_unknown_for_exact_body_and_keeps_task_replays_scoped() {
316        // Preserve real guards while isolating other tests' process-global role
317        // and budget state; this runs the test harness, not MCP startup cleanup.
318        if std::env::var_os("LEAN_CTX_NATIVE_RECEIPT_SUBPROCESS").is_none() {
319            // Snapshot the environment/cwd only while other tests cannot lend
320            // this child their temporary policy, configuration or path settings.
321            let _environment = lean_ctx::core::data_dir::test_env_lock();
322            let output = std::process::Command::new(std::env::current_exe().unwrap())
323                .args(["--exact", "server::native_receipts::tests::real_mcp_path_publishes_unknown_for_exact_body_and_keeps_task_replays_scoped", "--nocapture"])
324                .env("LEAN_CTX_NATIVE_RECEIPT_SUBPROCESS", "1")
325                .env("LEAN_CTX_ROLE", "coder")
326                .output().unwrap();
327            assert!(
328                output.status.success(),
329                "{}\n{}",
330                String::from_utf8_lossy(&output.stdout),
331                String::from_utf8_lossy(&output.stderr)
332            );
333            return;
334        }
335        let _data = lean_ctx::core::data_dir::isolated_data_dir();
336        let directory = tempfile::tempdir().unwrap();
337        let path = directory.path().join("native.rs");
338        std::fs::write(
339            &path,
340            "pub fn native_receipt_marker() { let value = 42; }\n".repeat(40),
341        )
342        .unwrap();
343        let ledger_path = directory.path().join("ledger.jsonl");
344        let mut server =
345            crate::tools::LeanCtxServer::new_with_project_root(directory.path().to_str());
346        server.native_receipt_authority = Some(authority(&ledger_path));
347        let response = server
348            .call_tool_guarded(request(&path, "receipt-call-1"))
349            .await
350            .unwrap();
351        assert_ne!(response.is_error, Some(true), "{response:?}");
352        let metadata = &response.meta.as_ref().unwrap().0["canonical_receipt"];
353        assert_eq!(metadata["outcome"], "unknown");
354        let digest = metadata["receipt_digest"]
355            .as_str()
356            .unwrap()
357            .strip_prefix("sha256:")
358            .unwrap();
359        let data = lean_ctx::core::data_dir::lean_ctx_data_dir().unwrap();
360        let bytes = std::fs::read(
361            data.join("execution/receipts")
362                .join(format!("{digest}.json")),
363        )
364        .unwrap();
365        let receipt = ReceiptDocumentV1::from_canonical_bytes(&bytes).unwrap();
366        let signature =
367            Signature::from_slice(&STANDARD.decode(&receipt.signature).unwrap()).unwrap();
368        SigningKey::from_bytes(&[53; 32])
369            .verifying_key()
370            .verify_strict(&receipt.signing_bytes().unwrap(), &signature)
371            .unwrap();
372        assert_eq!(receipt.outcome.state, AcceptanceState::Unknown);
373        let body = &response.content[0].as_text().unwrap().text;
374        assert!(body.contains("native_receipt_marker"));
375        let output_tokens = lean_ctx::core::tokens::count_tokens(body) as u64;
376        assert!(
377            receipt
378                .values
379                .iter()
380                .any(|value| value.name == "output_tokens" && value.value == Some(output_tokens))
381        );
382        let replay = server
383            .call_tool_guarded(request(&path, "receipt-call-1"))
384            .await
385            .unwrap();
386        assert_eq!(
387            replay.meta.as_ref().unwrap().0["canonical_receipt"],
388            *metadata
389        );
390        let next = server
391            .call_tool_guarded(request(&path, "receipt-call-2"))
392            .await
393            .unwrap();
394        assert_ne!(
395            next.meta.as_ref().unwrap().0["canonical_receipt"]["receipt_digest"],
396            metadata["receipt_digest"]
397        );
398        let mut wrapped = request(&path, "receipt-call-3");
399        let arguments = wrapped.arguments.take().unwrap();
400        wrapped.name = "ctx_call".into();
401        wrapped.arguments = Some(
402            serde_json::from_value(serde_json::json!({
403                "name":"ctx_read", "arguments":arguments
404            }))
405            .unwrap(),
406        );
407        let wrapped_response = server.call_tool_guarded(wrapped).await.unwrap();
408        assert_eq!(
409            wrapped_response.meta.as_ref().unwrap().0["canonical_receipt"]["outcome"],
410            "unknown"
411        );
412        assert_eq!(&wrapped_response.content[0].as_text().unwrap().text, body);
413        let serialized = serde_json::to_string(&response).unwrap();
414        assert!(!serialized.contains(&lean_ctx::core::agent_identity::hex_encode(&[53; 32])));
415        use lean_ctx::core::execution_ledger::{ExecutionEvent, ExecutionLedgerStore};
416        let events = ExecutionLedgerStore::new(ledger_path)
417            .load_verified()
418            .unwrap();
419        let recorded: std::collections::BTreeSet<_> = events
420            .iter()
421            .filter_map(|event| {
422                if let ExecutionEvent::CanonicalReceiptRecorded { receipt_digest, .. } = event {
423                    Some(receipt_digest.as_str())
424                } else {
425                    None
426                }
427            })
428            .collect();
429        let expected: std::collections::BTreeSet<_> = [
430            metadata,
431            &next.meta.as_ref().unwrap().0["canonical_receipt"],
432            &wrapped_response.meta.as_ref().unwrap().0["canonical_receipt"],
433        ]
434        .into_iter()
435        .map(|value| value["receipt_digest"].as_str().unwrap())
436        .collect();
437        assert_eq!(expected.len(), 3);
438        assert_eq!(recorded, expected);
439        assert_eq!(
440            events
441                .iter()
442                .filter(|event| matches!(event, ExecutionEvent::CanonicalReceiptRecorded { .. }))
443                .count(),
444            3
445        );
446        assert!(
447            !events
448                .iter()
449                .any(|event| matches!(event, ExecutionEvent::OutcomeRecorded { .. }))
450        );
451    }
452
453    #[test]
454    fn terminal_capture_refuses_changed_delivery_and_duplicate_publication() {
455        use lean_ctx::core::engine_interface::{NativeContextEngine, planning};
456        let _data = lean_ctx::core::data_dir::isolated_data_dir();
457        let directory = tempfile::tempdir().unwrap();
458        let root = std::fs::canonicalize(directory.path()).unwrap();
459        let ledger_path = root.join("ledger.jsonl");
460        let task: TaskEnvelopeV1 = serde_json::from_value(serde_json::json!({
461            "schema_version":1,"task_id":"capture-task","trace_id":"capture-trace",
462            "project_id":"capture-project","session_id":"capture-session","agent_id":"capture-agent",
463            "complexity":"unknown","created_at":"2026-01-01T00:00:00Z"
464        })).unwrap();
465        let admission = lean_ctx_protocol::EnginePolicyAdmissionV1 {
466            policy_ref: lean_ctx_protocol::ProtocolReference::new("policy:capture-test").unwrap(),
467            decision: lean_ctx_protocol::EnginePolicyDecisionV1::Admitted,
468        };
469        let plan = planning::native_plan(&task, &admission).unwrap();
470        let capture = NativeReceiptCapture::new(authority(&ledger_path), task.clone());
471        capture.begin(&task, &plan).unwrap();
472        let (invocation, observation) = NativeContextEngine::with_root(&root)
473            .unwrap()
474            .execute_ctx_read_rooted_snapshot_with_plan(
475                root.join("captured.rs").to_str().unwrap(),
476                "fn original() {}",
477                admission,
478                &task,
479                &plan,
480            )
481            .unwrap();
482        capture.complete(invocation, observation).unwrap();
483        let changed = CallToolResult::success(vec![rmcp::model::ContentBlock::text(
484            "changed post-processing",
485        )]);
486        assert!(capture.publish(&changed).is_err());
487        assert_eq!(
488            capture.publish(&changed).unwrap_err(),
489            "mcp_receipt_publication_already_claimed"
490        );
491        use lean_ctx::core::execution_ledger::{ExecutionEvent, ExecutionLedgerStore};
492        let events = ExecutionLedgerStore::new(ledger_path)
493            .load_verified()
494            .unwrap();
495        assert_eq!(events.len(), 1);
496        assert!(matches!(events[0], ExecutionEvent::TaskStarted { .. }));
497    }
498}
lean_ctx_server/server/native_receipts.rs

lean_ctx_server/server/
native_receipts.rs
