#[allow(clippy::wildcard_imports)]
use super::*;
use crate::sealed_origin::engine::{self, Presentation, Sources};
use lean_ctx::core::archive::{authority::ArchiveAuthority, delivered, presented};
const UNKNOWN: &str = "egress.sealed_source_authority_unknown";
const NOT_PRESENTED: &str = "egress.sealed_source_not_presented";
const LAPSED: &str = "egress.sealed_authority_lapsed";
const PLAN_SOURCE: &str = "object://caller/plan-1";
const FILE_TEXT: &str = "release notes: nothing to hide\n";
const PLAN_TEXT: &str = "prepared context: the open tickets of this sprint";
fn tool_call_reply() -> (&'static str, Vec<u8>) {
let message = serde_json::json!({
"id": "msg_0", "type": "message", "role": "assistant",
"content": [{"type": "tool_use", "id": "toolu_1", "name": "read",
"input": {"path": "notes.txt"}}]
});
("application/json", serde_json::to_vec(&message).unwrap())
}
fn tool_messages(blocks: &[&str]) -> Vec<serde_json::Value> {
let blocks: Vec<serde_json::Value> = blocks
.iter()
.map(|text| serde_json::json!({"type": "text", "text": text}))
.collect();
vec![
serde_json::json!({"role": "user", "content": "hello"}),
serde_json::json!({"role": "assistant", "content": [
{"type": "tool_use", "id": "toolu_1", "name": "read",
"input": {"path": "notes.txt"}}
]}),
serde_json::json!({"role": "user", "content": [
{"type": "tool_result", "tool_use_id": "toolu_1", "content": blocks}
]}),
]
}
fn tool_turn(blocks: &[&str]) -> serde_json::Value {
serde_json::json!({
"model": "claude-test", "max_tokens": 16, "messages": tool_messages(blocks)
})
}
fn tool_returning_turn(blocks: &[&str]) -> serde_json::Value {
let mut messages = tool_messages(blocks);
messages.push(serde_json::json!({"role": "assistant", "content": [
{"type": "thinking", "thinking": "plan", "signature": SEALED_A}
]}));
messages.push(serde_json::json!({"role": "user", "content": "continue"}));
serde_json::json!({"model": "claude-test", "max_tokens": 16, "messages": messages})
}
fn three_answers() -> Vec<(&'static str, Vec<u8>)> {
vec![
tool_call_reply(),
thinking_message(&[SEALED_A]),
empty_reply(),
]
}
struct Scenario {
_bound: crate::store_binding::BoundHostForTest,
_mode: lean_ctx::test_env::EnvVarGuard,
_data: lean_ctx::core::data_dir::IsolatedDataDir,
_live: LivePolicy,
_directory: tempfile::TempDir,
root: std::path::PathBuf,
upstream: String,
server: Upstream,
}
fn configure_project_root(root: &std::path::Path) {
write_gateway_config(&format!(
"{GOVERNED}\n[proxy]\nproject_root = '{}'\n",
root.display()
));
}
fn project_with_pack() -> (tempfile::TempDir, std::path::PathBuf) {
let directory = tempfile::tempdir().expect("project");
let root = directory.path().canonicalize().expect("canonical root");
write_pack(&root);
(directory, root)
}
impl Scenario {
async fn start() -> Self {
let data = lean_ctx::core::data_dir::isolated_data_dir();
let mode = proxy_mode("cache");
let live = LivePolicy {
_guard: lean_ctx::core::policy::runtime::lock_test_override(),
};
engine::reset();
let (directory, root) = project_with_pack();
configure_project_root(&root);
let bound = crate::store_binding::bound_host(&root);
let (upstream, server) = responding_upstream(three_answers()).await;
Self {
_bound: bound,
_mode: mode,
_data: data,
_live: live,
_directory: directory,
root,
upstream,
server,
}
}
fn source_file(&self, content: &str) -> std::path::PathBuf {
let path = self.root.join("notes.txt");
std::fs::write(&path, content).expect("source");
path
}
async fn turn(&self, body: &serde_json::Value) -> Response {
anthropic_turn(&self.upstream, body, None).await
}
async fn produce(&self, blocks: &[&str]) {
let opening = self.turn(&opening_turn()).await;
assert_eq!(opening.status(), StatusCode::OK, "opening turn");
let tool = self.turn(&tool_turn(blocks)).await;
assert_eq!(tool.status(), StatusCode::OK, "tool turn");
}
async fn assert_carried_on(self, blocks: &[&str]) {
let returned = tool_returning_turn(blocks);
let carried = self.turn(&returned).await;
assert_eq!(carried.status(), StatusCode::OK);
let bodies = self.server.await.expect("upstream");
assert_eq!(bodies.len(), 3);
assert_eq!(bodies[2], serde_json::to_vec(&returned).unwrap());
}
async fn assert_not_carried_on(self, blocks: &[&str], code: &str) {
let refused = self.turn(&tool_returning_turn(blocks)).await;
assert_sealed_refusal(&refused, code, code);
assert_eq!(
self.server.await.expect("upstream").len(),
2,
"the return never reached the destination"
);
}
}
fn as_engine<T>(root: &std::path::Path, operation: impl FnOnce() -> T) -> T {
lean_ctx::core::policy::runtime::REQUEST_PROJECT
.sync_scope(std::cell::RefCell::new(Some(root.to_path_buf())), operation)
}
#[tokio::test]
#[serial_test::serial(cache_telemetry)]
async fn real_file_lineage_carries_a_sealed_item_and_lapses_when_the_file_is_deleted() {
let scenario = Scenario::start().await;
let path = scenario.source_file(FILE_TEXT);
let delivered_text = as_engine(&scenario.root, || {
let path = path.to_str().expect("utf-8 path");
let admitted = lean_ctx::core::rooted_read::read_file_for_tool_rooted(
path,
scenario.root.to_str().expect("utf-8 root"),
"ctx_read",
)
.expect("admitted read");
let authority = ArchiveAuthority::file(&scenario.root, path, "ctx_read", &admitted)
.expect("protection is active: the read is bound");
assert!(
delivered::record_lineage(
&delivered::digest(&admitted),
&delivered::Lineage::of_files(authority)
),
"the Engine records the lineage of what it delivers"
);
admitted
});
let blocks = [delivered_text.as_str()];
scenario.produce(&blocks).await;
let returned = tool_returning_turn(&blocks);
let carried = scenario.turn(&returned).await;
assert_eq!(carried.status(), StatusCode::OK, "every source holds");
std::fs::remove_file(&path).expect("delete the source");
let refused = scenario.turn(&returned).await;
assert_sealed_refusal(&refused, LAPSED, "source deleted");
let bodies = scenario.server.await.expect("upstream");
assert_eq!(bodies.len(), 3, "only the first return arrived");
assert_eq!(bodies[2], serde_json::to_vec(&returned).unwrap());
}
fn shell_call_reply() -> (&'static str, Vec<u8>) {
let message = serde_json::json!({
"id": "msg_0", "type": "message", "role": "assistant",
"content": [{"type": "tool_use", "id": "toolu_1", "name": "bash",
"input": {"command": "make"}}]
});
("application/json", serde_json::to_vec(&message).unwrap())
}
fn shell_messages(block: &str) -> Vec<serde_json::Value> {
vec![
serde_json::json!({"role": "user", "content": "hello"}),
serde_json::json!({"role": "assistant", "content": [
{"type": "tool_use", "id": "toolu_1", "name": "bash",
"input": {"command": "make"}}
]}),
serde_json::json!({"role": "user", "content": [
{"type": "tool_result", "tool_use_id": "toolu_1",
"content": [{"type": "text", "text": block}]}
]}),
]
}
const HOST_EXPAND_TOOL: &str = "mcp__lean-ctx__ctx_expand";
fn shell_tools() -> serde_json::Value {
serde_json::json!([{
"name": HOST_EXPAND_TOOL,
"description": "Retrieve archived tool output.",
"input_schema": {"type": "object"}
}])
}
fn shell_turn(block: &str) -> serde_json::Value {
serde_json::json!({
"model": "claude-test", "max_tokens": 16, "tools": shell_tools(),
"messages": shell_messages(block)
})
}
const ENGINE_CALL: &str = "toolu_engine_1";
const ENGINE_SHELL: &str = "mcp__lean-ctx__ctx_shell";
fn configure_pruning_project_root(root: &std::path::Path) {
write_gateway_config(&format!(
"{GOVERNED}\n[proxy]\nproject_root = '{}'\nhistory_mode = \"rolling\"\n",
root.display()
));
}
fn engine_call_reply(tool: &str) -> (&'static str, Vec<u8>) {
let message = serde_json::json!({
"id": "msg_0", "type": "message", "role": "assistant",
"content": [{"type": "tool_use", "id": ENGINE_CALL, "name": tool,
"input": {"command": "make"}}]
});
("application/json", serde_json::to_vec(&message).unwrap())
}
fn engine_messages(tool: &str, block: &str) -> Vec<serde_json::Value> {
let mut messages = vec![
serde_json::json!({"role": "user", "content": "hello"}),
serde_json::json!({"role": "assistant", "content": [
{"type": "tool_use", "id": ENGINE_CALL, "name": tool,
"input": {"command": "make"}}
]}),
serde_json::json!({"role": "user", "content": [
{"type": "tool_result", "tool_use_id": ENGINE_CALL,
"content": [{"type": "text", "text": block}]}
]}),
];
for step in 1..=6 {
messages.push(serde_json::json!({"role": "user", "content": format!("go on {step}")}));
}
messages
}
fn engine_turn(tool: &str, block: &str, tools: &serde_json::Value) -> serde_json::Value {
serde_json::json!({
"model": "claude-test", "max_tokens": 16, "tools": tools,
"messages": engine_messages(tool, block)
})
}
fn engine_returning_turn(tool: &str, block: &str) -> serde_json::Value {
let mut messages = engine_messages(tool, block);
messages.push(serde_json::json!({"role": "assistant", "content": [
{"type": "thinking", "thinking": "plan", "signature": SEALED_A}
]}));
messages.push(serde_json::json!({"role": "user", "content": "continue"}));
serde_json::json!({
"model": "claude-test", "max_tokens": 16, "tools": shell_tools(), "messages": messages
})
}
fn engine_answers_the_call(scenario: &Scenario, content: &str) -> (std::path::PathBuf, String) {
crate::store_binding::engine_delivers(
&scenario.root,
"notes.txt",
content,
content,
Some(ENGINE_CALL),
);
(scenario.root.join("notes.txt"), content.to_owned())
}
fn pruned_shell_result(original: &str) -> String {
let lines: Vec<&str> = original.lines().collect();
format!(
"{}\n[...{} lines pruned by lean-ctx...]\n{}",
lines[..3].join("\n"),
lines.len() - 5,
lines[lines.len() - 2..].join("\n")
)
}
fn assert_pruned_with_stub(sent: &str, original: &str, tool: &str) -> String {
assert!(
sent.len() < original.len(),
"the forward path did not shorten the block ({} -> {})",
original.len(),
sent.len()
);
let (_, rest) = sent
.rsplit_once("ctx_expand(id=\"")
.expect("the block ends with a recovery stub");
let id = rest.split('"').next().expect("an id").to_owned();
let recovery_id = lean_ctx::core::archive::RecoveryId::parse(&id).expect("a recovery id");
assert_eq!(
sent,
format!(
"{}\n{}",
pruned_shell_result(original),
crate::compress::recovery_stub_for(tool, &recovery_id)
),
"the pruned form, and as its last line the stub that names {tool}"
);
id
}
fn sent_block(body: &[u8]) -> String {
let body: serde_json::Value = serde_json::from_slice(body).expect("JSON request");
body["messages"][2]["content"][0]["content"][0]["text"]
.as_str()
.expect("the tool-result text block")
.to_owned()
}
fn compressible_log() -> String {
(1..=400)
.map(|i| format!("[info] processed item {i:04} ok"))
.collect::<Vec<_>>()
.join("\n")
}
fn engine_delivers_file(scenario: &Scenario, content: &str) -> (std::path::PathBuf, String) {
let path = scenario.source_file(content);
let delivered_text = as_engine(&scenario.root, || {
let path = path.to_str().expect("utf-8 path");
let admitted = lean_ctx::core::rooted_read::read_file_for_tool_rooted(
path,
scenario.root.to_str().expect("utf-8 root"),
"ctx_read",
)
.expect("admitted read");
let authority = ArchiveAuthority::file(&scenario.root, path, "ctx_read", &admitted)
.expect("protection is active: the read is bound");
assert!(delivered::record_lineage(
&delivered::digest(&admitted),
&delivered::Lineage::of_files(authority)
));
admitted
});
(path, delivered_text)
}
fn reason_codes(response: &Response) -> Option<String> {
response
.headers()
.get(crate::admitted::REASON_CODES_HEADER)
.and_then(|value| value.to_str().ok())
.map(str::to_owned)
}
#[tokio::test]
#[serial_test::serial(cache_telemetry)]
async fn sealed_item_of_a_request_with_a_compressible_block_follows_the_blocks_lineage() {
let mut scenario = Scenario::start().await;
configure_pruning_project_root(&scenario.root);
let (upstream, server) = responding_upstream(vec![
engine_call_reply(ENGINE_SHELL),
thinking_message(&[SEALED_A]),
empty_reply(),
])
.await;
scenario.upstream = upstream;
let (path, original) = engine_answers_the_call(&scenario, &compressible_log());
let opening = scenario.turn(&opening_turn()).await;
assert_eq!(opening.status(), StatusCode::OK, "opening turn");
let tool = scenario
.turn(&engine_turn(ENGINE_SHELL, &original, &shell_tools()))
.await;
assert_eq!(tool.status(), StatusCode::OK, "tool turn");
assert_eq!(reason_codes(&tool), None, "nothing stayed whole");
let returned = engine_returning_turn(ENGINE_SHELL, &original);
let carried = scenario.turn(&returned).await;
assert_eq!(
carried.status(),
StatusCode::OK,
"the lineage holds: {:?}",
reason_codes(&carried)
);
let bodies = server.await.expect("upstream");
assert_eq!(bodies.len(), 3);
let sent = sent_block(&bodies[1]);
let id = assert_pruned_with_stub(&sent, &original, HOST_EXPAND_TOOL);
assert_eq!(
crate::compress::resolve_recovery(&id),
Some(Ok(original.clone())),
"the id resolves to the exact original bytes"
);
assert_eq!(
sent_block(&bodies[2]),
sent,
"the next turn sends the same block: the stub and its id survive"
);
std::fs::remove_file(&path).expect("delete the source");
let refused = scenario.turn(&returned).await;
assert_sealed_refusal(&refused, LAPSED, "source deleted");
assert_eq!(
crate::compress::resolve_recovery(&id),
Some(Err("recovery.source_no_longer_admitted"))
);
}
#[tokio::test]
#[serial_test::serial(cache_telemetry)]
async fn a_block_kept_whole_under_protection_is_named_with_the_forwarded_response() {
let mut scenario = Scenario::start().await;
let (upstream, server) =
responding_upstream(vec![shell_call_reply(), thinking_message(&[SEALED_A])]).await;
scenario.upstream = upstream;
let (_path, delivered_text) = engine_delivers_file(&scenario, &compressible_log());
let unknown = format!("{delivered_text}\n[info] a line the Engine never delivered");
let opening = scenario.turn(&opening_turn()).await;
assert_eq!(opening.status(), StatusCode::OK, "opening turn");
assert_eq!(reason_codes(&opening), None);
let tool = scenario.turn(&shell_turn(&unknown)).await;
assert_eq!(tool.status(), StatusCode::OK, "forwarded, not refused");
assert!(!tool.headers().contains_key(crate::admitted::REFUSAL_HEADER));
assert_eq!(
reason_codes(&tool).as_deref(),
Some("recovery.handle_not_followable")
);
let bodies = server.await.expect("upstream");
assert_eq!(bodies.len(), 2);
assert_eq!(sent_block(&bodies[1]), unknown, "the block left whole");
}
#[tokio::test]
#[serial_test::serial(cache_telemetry)]
async fn real_presented_lineage_lapsed_by_time_refuses_the_sealed_item_until_presented_again() {
let scenario = Scenario::start().await;
let binding = as_engine(&scenario.root, || {
let binding = presented::PresentedBinding::admitted_now(
&scenario.root,
PLAN_SOURCE,
"source-one",
"sha256:7d1a54127b222502f5b79b5fb0803061152a44f92b37e23c6527baf665d4da9a",
)
.expect("the policy identity of the project resolves");
assert!(presented::record_presentation(&binding));
assert!(delivered::record_lineage(
&delivered::digest(PLAN_TEXT),
&delivered::Lineage::of_presented(vec![binding.clone()])
));
binding
});
scenario.produce(&[PLAN_TEXT]).await;
let returned = tool_returning_turn(&[PLAN_TEXT]);
let clock = presented::TestClock::past_validity();
let lapsed = scenario.turn(&returned).await;
assert_sealed_refusal(&lapsed, NOT_PRESENTED, "validity elapsed");
as_engine(&scenario.root, || {
assert!(presented::record_presentation(&binding));
});
let carried = scenario.turn(&returned).await;
assert_eq!(carried.status(), StatusCode::OK, "presented again");
drop(clock);
let bodies = scenario.server.await.expect("upstream");
assert_eq!(
bodies.len(),
3,
"only the return after the new presentation"
);
assert_eq!(bodies[2], serde_json::to_vec(&returned).unwrap());
}
#[tokio::test]
#[serial_test::serial(cache_telemetry)]
async fn real_presented_lineage_carries_a_sealed_item_while_it_is_presented() {
let scenario = Scenario::start().await;
as_engine(&scenario.root, || {
let binding = presented::PresentedBinding::admitted_now(
&scenario.root,
PLAN_SOURCE,
"source-one",
"sha256:7d1a54127b222502f5b79b5fb0803061152a44f92b37e23c6527baf665d4da9a",
)
.expect("the policy identity of the project resolves");
assert!(presented::record_presentation(&binding));
assert!(
delivered::record_lineage(
&delivered::digest(PLAN_TEXT),
&delivered::Lineage::of_presented(vec![binding])
),
"the Engine records the lineage of the prepared context"
);
});
scenario.produce(&[PLAN_TEXT]).await;
scenario.assert_carried_on(&[PLAN_TEXT]).await;
}
#[tokio::test]
#[serial_test::serial(cache_telemetry)]
async fn sealed_item_is_carried_on_while_the_file_source_of_its_request_holds() {
let scenario = Scenario::start().await;
let path = scenario.source_file(FILE_TEXT);
engine::deliver(FILE_TEXT, Sources::file(&scenario.root, &path));
scenario.produce(&[FILE_TEXT]).await;
std::fs::write(&path, format!("{FILE_TEXT}\n// edited")).expect("edit the source");
scenario.assert_carried_on(&[FILE_TEXT]).await;
}
#[tokio::test]
#[serial_test::serial(cache_telemetry)]
async fn sealed_item_is_carried_on_while_the_presented_source_of_its_request_holds() {
let scenario = Scenario::start().await;
engine::deliver(PLAN_TEXT, Sources::presented(PLAN_SOURCE));
engine::present(PLAN_SOURCE, Some(Presentation::Admitted));
scenario.produce(&[PLAN_TEXT]).await;
scenario.assert_carried_on(&[PLAN_TEXT]).await;
}
#[tokio::test]
#[serial_test::serial(cache_telemetry)]
async fn sealed_item_is_carried_on_while_file_and_presented_sources_both_hold() {
let scenario = Scenario::start().await;
let path = scenario.source_file(FILE_TEXT);
engine::deliver(FILE_TEXT, Sources::file(&scenario.root, &path));
engine::deliver(PLAN_TEXT, Sources::presented(PLAN_SOURCE));
engine::present(PLAN_SOURCE, Some(Presentation::Admitted));
scenario.produce(&[FILE_TEXT, PLAN_TEXT]).await;
scenario.assert_carried_on(&[FILE_TEXT, PLAN_TEXT]).await;
}
#[tokio::test]
#[serial_test::serial(cache_telemetry)]
async fn one_lapsed_source_of_a_joined_lineage_refuses_the_sealed_item() {
let scenario = Scenario::start().await;
let path = scenario.source_file(FILE_TEXT);
engine::deliver(FILE_TEXT, Sources::file(&scenario.root, &path));
engine::deliver(PLAN_TEXT, Sources::presented(PLAN_SOURCE));
engine::present(PLAN_SOURCE, Some(Presentation::Admitted));
scenario.produce(&[FILE_TEXT, PLAN_TEXT]).await;
engine::present(PLAN_SOURCE, None);
scenario
.assert_not_carried_on(&[FILE_TEXT, PLAN_TEXT], NOT_PRESENTED)
.await;
}
#[tokio::test]
#[serial_test::serial(cache_telemetry)]
async fn sealed_item_is_refused_when_a_presented_source_is_not_presented_again() {
let scenario = Scenario::start().await;
engine::deliver(PLAN_TEXT, Sources::presented(PLAN_SOURCE));
engine::present(PLAN_SOURCE, Some(Presentation::Admitted));
scenario.produce(&[PLAN_TEXT]).await;
engine::present(PLAN_SOURCE, None);
scenario
.assert_not_carried_on(&[PLAN_TEXT], NOT_PRESENTED)
.await;
}
#[tokio::test]
#[serial_test::serial(cache_telemetry)]
async fn sealed_item_lapses_when_a_presented_source_is_no_longer_admitted() {
let scenario = Scenario::start().await;
engine::deliver(PLAN_TEXT, Sources::presented(PLAN_SOURCE));
engine::present(PLAN_SOURCE, Some(Presentation::Admitted));
scenario.produce(&[PLAN_TEXT]).await;
engine::present(PLAN_SOURCE, Some(Presentation::Refused));
scenario.assert_not_carried_on(&[PLAN_TEXT], LAPSED).await;
}
#[tokio::test]
#[serial_test::serial(cache_telemetry)]
async fn sealed_item_lapses_when_a_source_of_its_request_is_deleted() {
let scenario = Scenario::start().await;
let path = scenario.source_file(FILE_TEXT);
engine::deliver(FILE_TEXT, Sources::file(&scenario.root, &path));
scenario.produce(&[FILE_TEXT]).await;
std::fs::remove_file(&path).expect("delete the source");
scenario.assert_not_carried_on(&[FILE_TEXT], LAPSED).await;
}
use lean_ctx::test_env::CurrentDirGuard as WorkingDirectory;
#[tokio::test]
#[serial_test::serial(cache_telemetry)]
async fn working_directory_in_another_project_does_not_change_the_lineage_check() {
let scenario = Scenario::start().await;
let path = scenario.source_file(FILE_TEXT);
engine::deliver(FILE_TEXT, Sources::file(&scenario.root, &path));
let (_other_directory, other) = project_with_pack();
let _elsewhere = WorkingDirectory::enter(&other);
scenario.produce(&[FILE_TEXT]).await;
scenario.assert_carried_on(&[FILE_TEXT]).await;
}
#[tokio::test]
#[serial_test::serial(cache_telemetry)]
async fn no_resolved_project_under_protection_refuses_a_sealed_item_with_sources() {
let _data = lean_ctx::core::data_dir::isolated_data_dir();
let _community = lean_ctx::core::policy::runtime::TestPolicyOverride::set(None);
write_gateway_config(GOVERNED);
engine::reset();
let directory = tempfile::tempdir().expect("project");
let root = directory.path().canonicalize().expect("canonical root");
let path = root.join("notes.txt");
std::fs::write(&path, FILE_TEXT).expect("source");
engine::deliver(FILE_TEXT, Sources::file(&root, &path));
let _inside = WorkingDirectory::enter(&root);
let (upstream, server) = responding_upstream(three_answers()).await;
let opening = anthropic_turn(&upstream, &opening_turn(), None).await;
assert_eq!(opening.status(), StatusCode::OK);
let tool = anthropic_turn(&upstream, &tool_turn(&[FILE_TEXT]), None).await;
assert_eq!(tool.status(), StatusCode::OK);
let refused = anthropic_turn(&upstream, &tool_returning_turn(&[FILE_TEXT]), None).await;
assert_sealed_refusal(&refused, UNKNOWN, "no resolved project");
assert_eq!(server.await.expect("upstream").len(), 2);
}
#[tokio::test]
#[serial_test::serial(cache_telemetry)]
async fn configured_project_root_without_a_pack_carries_a_sealed_item_with_file_lineage() {
let _data = lean_ctx::core::data_dir::isolated_data_dir();
let _mode = proxy_mode("cache");
let _community = lean_ctx::core::policy::runtime::TestPolicyOverride::set(None);
engine::reset();
let directory = tempfile::tempdir().expect("project");
let root = directory.path().canonicalize().expect("canonical root");
configure_project_root(&root);
let path = root.join("notes.txt");
std::fs::write(&path, FILE_TEXT).expect("source");
engine::deliver(FILE_TEXT, Sources::file(&root, &path));
let unrelated = tempfile::tempdir().expect("unrelated directory");
let _elsewhere = WorkingDirectory::enter(unrelated.path());
let (upstream, server) = responding_upstream(three_answers()).await;
for turn in [opening_turn(), tool_turn(&[FILE_TEXT])] {
let answered = anthropic_turn(&upstream, &turn, None).await;
assert_eq!(answered.status(), StatusCode::OK);
}
let returned = tool_returning_turn(&[FILE_TEXT]);
let carried = anthropic_turn(&upstream, &returned, None).await;
assert_eq!(carried.status(), StatusCode::OK);
let bodies = server.await.expect("upstream");
assert_eq!(bodies.len(), 3);
assert_eq!(bodies[2], serde_json::to_vec(&returned).unwrap());
}
#[tokio::test]
#[serial_test::serial(cache_telemetry)]
async fn unusable_configured_project_root_refuses_every_request() {
let _data = lean_ctx::core::data_dir::isolated_data_dir();
let _community = lean_ctx::core::policy::runtime::TestPolicyOverride::set(None);
let directory = tempfile::tempdir().expect("project");
let missing = directory.path().join("missing");
for unusable in [missing.as_path(), std::path::Path::new("/")] {
configure_project_root(unusable);
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let upstream = format!("http://{}", listener.local_addr().unwrap());
let refused = anthropic_turn(&upstream, &opening_turn(), None).await;
assert_eq!(refused.status(), StatusCode::SERVICE_UNAVAILABLE);
let codes = refused
.headers()
.get(crate::admitted::REASON_CODES_HEADER)
.and_then(|value| value.to_str().ok())
.unwrap_or_default();
assert!(codes.contains("gateway.project_root_unusable"), "{codes}");
assert!(
tokio::time::timeout(Duration::from_millis(50), listener.accept())
.await
.is_err(),
"nothing is sent for a project the proxy cannot verify"
);
}
}
#[tokio::test]
#[serial_test::serial(cache_telemetry)]
async fn sealed_item_is_refused_when_a_tool_result_block_of_its_request_has_no_lineage() {
let rewrapped = format!("<result>{FILE_TEXT}</result>");
let cases: [(&str, Vec<&str>); 3] = [
("not the Engine's output", vec!["listing from a host tool"]),
("re-wrapped by the host", vec![rewrapped.as_str()]),
("one block of two", vec![FILE_TEXT, "appended by the host"]),
];
for (case, blocks) in cases {
let scenario = Scenario::start().await;
let path = scenario.source_file(FILE_TEXT);
engine::deliver(FILE_TEXT, Sources::file(&scenario.root, &path));
scenario.produce(&blocks).await;
let refused = scenario.turn(&tool_returning_turn(&blocks)).await;
assert_sealed_refusal(&refused, UNKNOWN, case);
assert_eq!(scenario.server.await.expect("upstream").len(), 2, "{case}");
}
}
#[tokio::test]
#[serial_test::serial(cache_telemetry)]
async fn forged_lineage_claims_of_a_client_are_ignored() {
async fn claiming_turn(scenario: &Scenario, body: &serde_json::Value, claim: &str) -> Response {
let mut body = body.clone();
body["metadata"] = serde_json::json!({"lean_ctx_lineage": claim, "user_id": "u"});
let request = Request::builder()
.method("POST")
.uri("/v1/messages")
.header(CONTENT_TYPE, "application/json")
.header("x-leanctx-lineage", claim)
.header("x-leanctx-source-presented", PLAN_SOURCE)
.header("x-leanctx-project", "/srv/another-project")
.body(Body::from(serde_json::to_vec(&body).unwrap()))
.unwrap();
let response = lean_ctx::core::policy::runtime::REQUEST_PROJECT
.scope(
std::cell::RefCell::new(Some(scenario.root.clone())),
tokio::time::timeout(
ROUTE_BOUND,
crate::anthropic::handler(State(proxy_test_state(&scenario.upstream)), request),
),
)
.await
.expect("bounded route")
.expect("an answer or a typed refusal");
let (parts, body) = response.into_parts();
let body = to_bytes(body, usize::MAX).await.expect("response body");
Response::from_parts(parts, Body::from(body))
}
let scenario = Scenario::start().await;
let path = scenario.source_file(FILE_TEXT);
engine::deliver(FILE_TEXT, Sources::file(&scenario.root, &path));
let claim = delivered::digest(FILE_TEXT);
let forged = format!("[lean-ctx lineage {claim}]\nsomething else entirely");
let blocks = [forged.as_str()];
let opening = claiming_turn(&scenario, &opening_turn(), &claim).await;
assert_eq!(opening.status(), StatusCode::OK);
let tool = claiming_turn(&scenario, &tool_turn(&blocks), &claim).await;
assert_eq!(tool.status(), StatusCode::OK);
let refused = claiming_turn(&scenario, &tool_returning_turn(&blocks), &claim).await;
assert_sealed_refusal(&refused, UNKNOWN, "forged claim");
assert_eq!(scenario.server.await.expect("upstream").len(), 2);
}
#[tokio::test]
#[serial_test::serial(cache_telemetry)]
async fn host_authored_text_only_is_a_complete_lineage_without_sources() {
let _data = lean_ctx::core::data_dir::isolated_data_dir();
let _mode = proxy_mode("cache");
let _community = lean_ctx::core::policy::runtime::TestPolicyOverride::set(None);
write_gateway_config(GOVERNED);
engine::reset();
let pasted = "here is my file:\nfn main() { println!(\"pasted by hand\"); }";
let opening = serde_json::json!({
"model": "claude-test", "max_tokens": 16,
"system": "be brief",
"messages": [{"role": "user", "content": pasted}]
});
let returned = serde_json::json!({
"model": "claude-test", "max_tokens": 16,
"system": "be brief",
"messages": [
{"role": "user", "content": pasted},
{"role": "assistant", "content": [
{"type": "thinking", "thinking": "plan", "signature": SEALED_A}
]},
{"role": "user", "content": "continue"}
]
});
let (upstream, server) =
responding_upstream(vec![thinking_message(&[SEALED_A]), empty_reply()]).await;
assert_eq!(
anthropic_turn(&upstream, &opening, None).await.status(),
StatusCode::OK
);
assert_eq!(
anthropic_turn(&upstream, &returned, None).await.status(),
StatusCode::OK
);
assert_eq!(
server.await.expect("upstream")[1],
serde_json::to_vec(&returned).unwrap()
);
}
#[tokio::test]
#[serial_test::serial(cache_telemetry)]
async fn developer_mode_returns_sealed_items_without_any_lineage() {
let _data = lean_ctx::core::data_dir::isolated_data_dir();
let _mode = proxy_mode("cache");
let _community = lean_ctx::core::policy::runtime::TestPolicyOverride::set(None);
engine::reset();
let blocks = ["listing from a host tool"];
let (upstream, server) = responding_upstream(three_answers()).await;
for turn in [opening_turn(), tool_turn(&blocks)] {
let answered = anthropic_turn(&upstream, &turn, None).await;
assert_eq!(answered.status(), StatusCode::OK);
}
let returned = tool_returning_turn(&blocks);
let forwarded = anthropic_turn(&upstream, &returned, None).await;
assert_eq!(forwarded.status(), StatusCode::OK);
let bodies = server.await.expect("upstream");
assert_eq!(bodies.len(), 3);
assert_eq!(bodies[2], serde_json::to_vec(&returned).unwrap());
}
fn is_recovery_note(decision: &lean_ctx_protocol::context_gateway::ContextDecisionV1) -> bool {
decision
.reason_codes
.iter()
.any(|reason| reason.as_str().starts_with("recovery."))
}
#[tokio::test]
#[serial_test::serial(cache_telemetry)]
async fn the_persisted_receipt_notes_a_block_kept_whole_and_counts_it_as_no_source() {
let mut scenario = Scenario::start().await;
let (upstream, server) = responding_upstream(vec![shell_call_reply(), empty_reply()]).await;
scenario.upstream = upstream;
let (_path, delivered_text) = engine_delivers_file(&scenario, &compressible_log());
let unknown = format!("{delivered_text}\n[info] a line the Engine never delivered");
let opening = scenario.turn(&opening_turn()).await;
assert_eq!(opening.status(), StatusCode::OK, "opening turn");
assert!(
proxy_receipts()
.iter()
.all(|receipt| !receipt.decisions.iter().any(is_recovery_note)),
"nothing stayed whole in the opening turn"
);
let tool = scenario.turn(&shell_turn(&unknown)).await;
assert_eq!(tool.status(), StatusCode::OK, "forwarded, not refused");
assert_eq!(server.await.expect("upstream").len(), 2);
let receipts = proxy_receipts();
let noted: Vec<&ContextDecisionReceiptV1> = receipts
.iter()
.filter(|receipt| receipt.decisions.iter().any(is_recovery_note))
.collect();
assert_eq!(noted.len(), 1, "the receipt of the tool turn, and only it");
let receipt = noted[0];
let (notes, sources): (Vec<_>, Vec<_>) = receipt
.decisions
.iter()
.partition(|decision| is_recovery_note(decision));
assert_eq!(notes.len(), 1, "one note for the one block");
assert_eq!(
notes[0].disposition,
lean_ctx_protocol::context_gateway::ContextDispositionV1::Allow
);
assert_eq!(
notes[0]
.reason_codes
.iter()
.map(lean_ctx_protocol::context_gateway::ReasonCodeV1::as_str)
.collect::<Vec<_>>(),
["recovery.handle_not_followable"]
);
assert!(!sources.is_empty(), "the request has sources");
assert_eq!(receipt.sources.inspected as usize, sources.len());
assert_eq!(
receipt.sources.permitted as usize,
sources
.iter()
.filter(|decision| decision.disposition.delivers_content())
.count()
);
assert_eq!(receipt.sources.selected, receipt.sources.permitted);
assert_eq!(receipt.sources.blocked, 0);
}
#[tokio::test]
#[serial_test::serial(cache_telemetry)]
async fn a_stub_the_final_admission_would_alter_is_taken_back() {
const TAKEN_BACK: &str = "recovery.stub_not_admissible";
let mut scenario = Scenario::start().await;
let (upstream, server) = responding_upstream(vec![
engine_call_reply(ENGINE_SHELL),
empty_reply(),
empty_reply(),
])
.await;
scenario.upstream = upstream;
write_gateway_config(&format!(
"{GOVERNED}\n[proxy]\nproject_root = '{}'\nhistory_mode = \"rolling\"\n\n\
[secret_detection]\ncustom_patterns = ['[0-9a-f]{{64}}']\n",
scenario.root.display()
));
let (_path, original) = engine_answers_the_call(&scenario, &compressible_log());
let turn = engine_turn(ENGINE_SHELL, &original, &shell_tools());
let opening = scenario.turn(&opening_turn()).await;
assert_eq!(opening.status(), StatusCode::OK, "opening turn");
let guarded = scenario.turn(&turn).await;
assert_eq!(guarded.status(), StatusCode::OK, "forwarded, not refused");
assert!(
!guarded
.headers()
.contains_key(crate::admitted::REFUSAL_HEADER)
);
assert_eq!(reason_codes(&guarded).as_deref(), Some(TAKEN_BACK));
let noted: Vec<ContextDecisionReceiptV1> = proxy_receipts()
.into_iter()
.filter(|receipt| has_reason(receipt, TAKEN_BACK))
.collect();
assert_eq!(noted.len(), 1, "the receipt of the guarded turn names it");
assert!(
!has_reason(¬ed[0], "secret.redacted"),
"the decision on the stubbed body is not in the receipt"
);
configure_pruning_project_root(&scenario.root);
let control = scenario.turn(&turn).await;
assert_eq!(control.status(), StatusCode::OK, "control turn");
assert_eq!(reason_codes(&control), None, "nothing stayed whole");
let bodies = server.await.expect("upstream");
assert_eq!(bodies.len(), 3);
assert_eq!(
sent_block(&bodies[1]),
original,
"the block left as its original, whole"
);
let id = assert_pruned_with_stub(&sent_block(&bodies[2]), &original, HOST_EXPAND_TOOL);
assert_eq!(
crate::compress::resolve_recovery(&id),
Some(Ok(original.clone())),
"the id of the control's stub resolves to the exact original"
);
assert_eq!(
proxy_receipts()
.iter()
.filter(|receipt| has_reason(receipt, TAKEN_BACK))
.count(),
1,
"the control turn notes nothing"
);
}
#[path = "expand_tool_tests.rs"]
mod expand_tool;