--- a/Cargo.lock
+++ b/Cargo.lock
@@ -333,7 +333,9 @@
 name = "airouter-tools"
 version = "0.1.2"
 dependencies = [
+ "airouter-agent-protocol",
  "airouter-catalog",
+ "airouter-mcp",
  "async-trait",
  "axum",
  "base64 0.22.1",
@@ -348,6 +350,7 @@
  "tempfile",
  "thiserror",
  "tokio",
+ "tokio-util",
  "toml",
  "tower",
  "tracing",
--- a/crates/airouter-tools/Cargo.toml
+++ b/crates/airouter-tools/Cargo.toml
@@ -5,6 +5,9 @@
 license.workspace = true
 
 [dependencies]
+airouter-mcp = { path = "../airouter-mcp", features = ["stdio"] }
+airouter-agent-protocol = { path = "../airouter-agent-protocol" }
+tokio-util = { version = "0.7", features = ["rt"] }
 airouter-catalog = { path = "../airouter-catalog", default-features = false }
 async-trait.workspace = true
 axum.workspace = true
--- /dev/null
+++ b/crates/airouter-tools/examples/mcp-fixture.rs
@@ -0,0 +1,16 @@
+#[path = "../tests/support/mcp_fixture.rs"]
+mod fixture;
+#[tokio::main]
+async fn main() {
+    let args: Vec<_> = std::env::args_os().skip(1).collect();
+    assert_eq!(args.len(), 2, "fixture ROOT ADAPTER");
+    let f = fixture::Fixture::start(
+        std::path::Path::new(&args[0]),
+        std::path::Path::new(&args[1]),
+    )
+    .await;
+    // Touch readiness only after both private listeners and the ledger exist.
+    let _ = &f.runtime;
+    std::fs::write(f.root.join("ready"), b"ready").unwrap();
+    std::future::pending::<()>().await;
+}
--- a/crates/airouter-tools/src/api.rs
+++ b/crates/airouter-tools/src/api.rs
@@ -89,12 +89,13 @@
     }
 }
 
-struct Inner {
-    store: Store,
-    active: HashMap<Uuid, (String, watch::Sender<bool>)>,
+pub(crate) struct Inner {
+    pub(crate) store: Store,
+    pub(crate) active: HashMap<Uuid, (String, watch::Sender<bool>)>,
 }
 pub struct Runtime {
-    inner: Mutex<Inner>,
+    pub(crate) inner: Mutex<Inner>,
+    pub(crate) mcp: Option<Arc<crate::mcp::Broker>>,
     applications: Vec<Application>,
     pub models: Models,
     executor: Arc<dyn Executor>,
@@ -116,6 +117,7 @@
                 store,
                 active: HashMap::new(),
             }),
+            mcp: None,
             applications,
             models,
             executor,
@@ -125,6 +127,10 @@
             accepting: AtomicBool::new(true),
             desktops: None,
         }
+    }
+    pub fn with_mcp(mut self, broker: Arc<crate::mcp::Broker>) -> Self {
+        self.mcp = Some(broker);
+        self
     }
     pub fn with_desktops(mut self, desktops: Arc<crate::desktop::Manager>) -> Self {
         self.desktops = Some(desktops);
@@ -163,7 +169,7 @@
     }
 }
 
-struct Error(String);
+pub(crate) struct Error(pub(crate) String);
 impl From<String> for Error {
     fn from(s: String) -> Self {
         Self(s)
@@ -204,7 +210,10 @@
             | "dns_resolution_failed"
             | "web_content_type_denied" => StatusCode::BAD_REQUEST,
             "web_timeout" => StatusCode::GATEWAY_TIMEOUT,
-            "unavailable" | "service_unavailable" => StatusCode::SERVICE_UNAVAILABLE,
+            "mcp_configuration_unavailable"
+            | "mcp_authorization_unavailable"
+            | "unavailable"
+            | "service_unavailable" => StatusCode::SERVICE_UNAVAILABLE,
             _ => StatusCode::INTERNAL_SERVER_ERROR,
         };
         (status, Json(ApiError { code: self.0 })).into_response()
@@ -212,6 +221,7 @@
 }
 pub fn router(runtime: Arc<Runtime>) -> Router {
     Router::new()
+        .merge(crate::mcp::api::routes())
         .route("/healthz", get(health))
         .route("/v1/catalog", get(catalog))
         .route(
@@ -281,7 +291,7 @@
             .into_response(),
     }
 }
-fn scope(app: &Application, id: Uuid) -> Result<(), Error> {
+pub(crate) fn scope(app: &Application, id: Uuid) -> Result<(), Error> {
     if app.scopes.contains(&id) {
         Ok(())
     } else {
@@ -300,6 +310,12 @@
 ) -> Json<Catalog> {
     let mut tools = rt.models.tools(&app.tools);
     for tool in &mut tools {
+        if tool.name == crate::mcp::TOOL {
+            tool.input_schema = crate::mcp::schema();
+            tool.unavailable_reason = rt.mcp.is_none().then(|| "mcp_not_configured".into());
+            tool.available = tool.unavailable_reason.is_none();
+            continue;
+        }
         tool.unavailable_reason = if tool.name == crate::desktop::protocol::TOOL {
             rt.desktops
                 .as_ref()
@@ -404,13 +420,14 @@
     Path((scope_id, request_id)): Path<(Uuid, Uuid)>,
 ) -> Result<Json<Job>, Error> {
     scope(&app, scope_id)?;
-    Ok(Json(
-        rt.inner
-            .lock()
-            .unwrap()
-            .store
-            .job_by_request(&app.id, scope_id, request_id)?,
-    ))
+    let job = rt
+        .inner
+        .lock()
+        .unwrap()
+        .store
+        .job_by_request(&app.id, scope_id, request_id)?;
+    public_job(&job)?;
+    Ok(Json(job))
 }
 
 fn start_job(
@@ -420,6 +437,20 @@
     request: CreateJob,
     inline: Option<&[u8]>,
 ) -> Result<(StatusCode, Json<Job>), Error> {
+    start_admitted_job(rt, app, scope_id, request, inline, false)
+}
+pub(crate) fn start_admitted_job(
+    rt: Arc<Runtime>,
+    app: Application,
+    scope_id: Uuid,
+    request: CreateJob,
+    inline: Option<&[u8]>,
+    registered: bool,
+) -> Result<(StatusCode, Json<Job>), Error> {
+    let mcp = matches!(request.operation, Operation::McpRead { .. });
+    if mcp != registered {
+        return Err("tool_denied".into());
+    }
     scope(&app, scope_id)?;
     if !(1..=300).contains(&request.timeout_seconds) || request.request_id.is_nil() {
         return Err("invalid_request".into());
@@ -432,6 +463,11 @@
     }
     let (job, rx, permit, input) = {
         let mut inner = rt.inner.lock().unwrap();
+        if inline.is_none() {
+            inner
+                .store
+                .public_artifact(&app.id, scope_id, request.input)?;
+        }
         let replay = match inline {
             Some(bytes) => inner.store.replay_run(&app.id, scope_id, &request, bytes)?,
             None => inner.store.replay(&app.id, scope_id, &request)?,
@@ -439,14 +475,24 @@
         if let Some(job) = replay {
             return Ok((StatusCode::OK, Json(job)));
         }
-        if rt
-            .executor
-            .unavailable_reason(&rt.models, request.operation.name())
-            .is_some()
+        if !mcp
+            && rt
+                .executor
+                .unavailable_reason(&rt.models, request.operation.name())
+                .is_some()
         {
             return Err("unavailable".into());
         }
-        if !matches!(request.operation, Operation::WebFetch {} | Operation::WebSearch {} | Operation::WebExtract {} | Operation::ResearchBundle {} | Operation::DocumentIngest { .. }) {
+        if !mcp
+            && !matches!(
+                request.operation,
+                Operation::WebFetch {}
+                    | Operation::WebSearch {}
+                    | Operation::WebExtract {}
+                    | Operation::ResearchBundle {}
+                    | Operation::DocumentIngest { .. }
+            )
+        {
             rt.models.plan(&request.operation)?;
         }
         if inner.active.values().any(|(id, _)| id == &app.id) {
@@ -471,10 +517,23 @@
         let executor = rt.executor.clone();
         let operation = current.request.operation.clone();
         let seconds = current.request.timeout_seconds;
-        let result =
-            tokio::spawn(async move { executor.execute(input, operation, seconds, rx).await })
-                .await
-                .unwrap_or_else(|_| Execution::failed("worker_task_failed"));
+        let result = tokio::spawn({
+            let broker = rt.mcp.clone();
+            let app = current.application.clone();
+            let scope = current.scope;
+            async move {
+                if matches!(operation, Operation::McpRead { .. }) {
+                    match broker {
+                        Some(broker) => broker.execute(app, scope, operation, seconds, rx).await,
+                        None => Execution::failed("unavailable"),
+                    }
+                } else {
+                    executor.execute(input, operation, seconds, rx).await
+                }
+            }
+        })
+        .await
+        .unwrap_or_else(|_| Execution::failed("worker_task_failed"));
         if matches!(
             result.error.as_deref(),
             Some("cleanup_failed" | "worker_task_failed")
@@ -505,9 +564,9 @@
     Path((scope_id, id)): Path<(Uuid, Uuid)>,
 ) -> Result<Json<Job>, Error> {
     scope(&app, scope_id)?;
-    Ok(Json(
-        rt.inner.lock().unwrap().store.job(&app.id, scope_id, id)?,
-    ))
+    let job = rt.inner.lock().unwrap().store.job(&app.id, scope_id, id)?;
+    public_job(&job)?;
+    Ok(Json(job))
 }
 async fn cancel(
     State(rt): State<Arc<Runtime>>,
@@ -516,6 +575,7 @@
 ) -> Result<Json<Job>, Error> {
     scope(&app, scope_id)?;
     let mut inner = rt.inner.lock().unwrap();
+    public_job(&inner.store.job(&app.id, scope_id, id)?)?;
     let job = inner.store.cancel(&app.id, scope_id, id)?;
     if let Some((_, tx)) = inner.active.get(&id) {
         let _ = tx.send(true);
@@ -659,3 +719,11 @@
     }
     Ok(output)
 }
+
+fn public_job(job: &Job) -> Result<(), Error> {
+    if matches!(job.request.operation, Operation::McpRead { .. }) {
+        Err("scope_denied".into())
+    } else {
+        Ok(())
+    }
+}
--- a/crates/airouter-tools/src/catalog.rs
+++ b/crates/airouter-tools/src/catalog.rs
@@ -31,6 +31,7 @@
 }
 
 pub const NAMES: &[&str] = &[
+    "mcp_read",
     "document_convert",
     "pdf_inspect",
     "pdf_extract",
@@ -63,8 +64,8 @@
             id:tool.name.clone(),version:env!("CARGO_PKG_VERSION").into(),kind:Kind::Tool,
             title:tool.name.replace('_', " "),description:tool.description,
             executor:"airouter-tools".into(),tags:vec!["utility".into(), tool.name.replace('_', " ")],
-            input_schema:if desktop { crate::desktop::protocol::schema() } else { tool.input_schema },
-            output_schema:json!({"type":"object","description":if desktop {"Desktop operation response; see the desktop protocol."} else {"Utility job record with phase, artifact references, errors and resource evidence."}}),
+            input_schema:if desktop { crate::desktop::protocol::schema() } else if tool.name=="mcp_read" { crate::mcp::schema() } else { tool.input_schema },
+            output_schema:json!({"type":"object","description":if desktop {"Desktop operation response; see the desktop protocol."} else if tool.name=="mcp_read" {"Registered read handle; evidence delivery requires the dedicated MCP result route and current frontend/data authority."} else {"Utility job record with phase, artifact references, errors and resource evidence."}}),
             examples:vec![],source_ref:"/srv/airouter/crates/airouter-tools/src/catalog.rs".into(),
             documentation:"/srv/airouter/crates/airouter-tools/README.md".into(),
         }
@@ -134,6 +135,7 @@
         NAMES.iter().filter(|n| grants.iter().any(|g| g == **n)).map(|name| {
             let reason = self.availability(name);
             let (description, properties, required): (&str, Value, Vec<&str>) = match *name {
+                "mcp_read" => ("Discover and execute registered read-only MCP tools through the dedicated authenticated frontend routes. Results require fresh data grants and caller context.",json!({}),vec![]),
                 "document_convert" => ("Convert an uploaded document with Pandoc. No filters, scripts, or external resources.",
                     json!({"from":{"enum":["markdown","html","docx","rst"]},"to":{"enum":["markdown","html","docx","plain","rst"]}}), vec!["from","to"]),
                 "pdf_inspect" => ("Inspect uploaded PDF page count and metadata.",json!({}),vec![]),
@@ -165,6 +167,7 @@
             mounts: vec![],
         };
         let (cmd, output, mime): (Vec<String>, String, &str) = match op {
+            Operation::McpRead { .. } => return Err("mcp_operation_not_sandboxed".into()),
             Operation::WebFetch {} => return Err("web_operation_not_sandboxed".into()),
             Operation::WebSearch {} => return Err("web_operation_not_sandboxed".into()),
             Operation::WebExtract {} => return Err("web_operation_not_sandboxed".into()),
--- a/crates/airouter-tools/src/lib.rs
+++ b/crates/airouter-tools/src/lib.rs
@@ -6,5 +6,7 @@
 pub mod desktop;
 pub mod protocol;
 pub mod store;
+pub mod web;
 pub mod worker;
-pub mod web;
+
+pub mod mcp;
--- a/crates/airouter-tools/src/main.rs
+++ b/crates/airouter-tools/src/main.rs
@@ -31,6 +31,8 @@
     desktop_worker: Option<PathBuf>,
     #[serde(default)]
     models: Models,
+    #[serde(default)]
+    mcp_registry: Option<PathBuf>,
     applications: Vec<App>,
 }
 #[derive(Deserialize)]
@@ -109,6 +111,9 @@
     )
     .await?;
     let mut runtime = Runtime::new(store, apps, config.models, Arc::new(sandbox));
+    if let Some(registry) = config.mcp_registry {
+        runtime = runtime.with_mcp(Arc::new(airouter_tools::mcp::Broker::new(registry)));
+    }
     if config.desktops {
         let mut backend = airouter_tools::desktop::backend::NativeBackend::new(
             config.cgroup_root,
--- /dev/null
+++ b/crates/airouter-tools/src/mcp/api.rs
@@ -0,0 +1,182 @@
+use super::{Authority, CONTEXT_HEADER, Caller, ReadRequest, SUBJECT_HEADER, TOOL};
+use crate::{
+    api::{Application, Error, Runtime, scope, start_admitted_job},
+    protocol::{CreateJob, Job, Operation, Phase},
+};
+use axum::{
+    Extension, Json, Router,
+    extract::{Path, State},
+    http::{HeaderMap, StatusCode},
+    routing::{get, post},
+};
+use serde_json::{Value, json};
+use std::sync::Arc;
+use uuid::Uuid;
+
+pub(crate) fn routes() -> Router<Arc<Runtime>> {
+    Router::new()
+        .route("/v1/scopes/{scope}/mcp/catalog", get(catalog))
+        .route("/v1/scopes/{scope}/mcp/runs", post(read))
+        .route("/v1/scopes/{scope}/mcp/runs/{id}", get(result))
+        .route("/v1/scopes/{scope}/mcp/runs/{id}/cancel", post(cancel))
+}
+fn caller(headers: &HeaderMap) -> Result<Caller, Error> {
+    let get = |key: &str| -> Result<&str, Error> {
+        if headers.get_all(key).iter().count() != 1 {
+            return Err("scope_denied".into());
+        }
+        headers
+            .get(key)
+            .and_then(|h| h.to_str().ok())
+            .ok_or_else(|| "scope_denied".into())
+    };
+    Ok(Caller::new(get(SUBJECT_HEADER)?, get(CONTEXT_HEADER)?)?)
+}
+fn broker(rt: &Runtime, app: &Application, id: Uuid) -> Result<Arc<super::Broker>, Error> {
+    scope(app, id)?;
+    if !app.tools.iter().any(|t| t == TOOL) {
+        return Err("tool_denied".into());
+    }
+    rt.mcp.clone().ok_or_else(|| "unavailable".into())
+}
+async fn catalog(
+    State(rt): State<Arc<Runtime>>,
+    Extension(app): Extension<Application>,
+    Path(id): Path<Uuid>,
+    headers: HeaderMap,
+) -> Result<Json<Value>, Error> {
+    let caller = caller(&headers)?;
+    let b = broker(&rt, &app, id)?;
+    let bindings = b.bindings(&app.id, id, &caller)?;
+    if bindings.is_empty() {
+        return Err("scope_denied".into());
+    }
+    let mut servers = vec![];
+    // Keep aggregate discovery bounded even when policy registers many endpoints.
+    for binding in bindings {
+        let (binding, authority) = b.authorize(&app.id, id, &caller, &binding.server).await?;
+        servers.push(b.discover(&binding, &authority).await?);
+        if serde_json::to_vec(&servers)
+            .map_err(|_| Error::from("mcp_response_invalid"))?
+            .len()
+            > super::MAX_RESULT
+        {
+            return Err("output_limit".into());
+        }
+    }
+    Ok(Json(
+        json!({"schema":"airouter.mcp.catalog.v1","servers":servers}),
+    ))
+}
+async fn read(
+    State(rt): State<Arc<Runtime>>,
+    Extension(app): Extension<Application>,
+    Path(id): Path<Uuid>,
+    headers: HeaderMap,
+    Json(request): Json<ReadRequest>,
+) -> Result<(StatusCode, Json<Value>), Error> {
+    let caller = caller(&headers)?;
+    let b = broker(&rt, &app, id)?;
+    if request.request_id.is_nil()
+        || !super::identifier(&request.server)
+        || !super::identifier(&request.name)
+        || !request.arguments.is_object()
+        || request.arguments.to_string().len() > 8192
+    {
+        return Err("invalid_request".into());
+    }
+    let (binding, authority) = b.authorize(&app.id, id, &caller, &request.server).await?;
+    if !binding.tools.contains(&request.name) {
+        return Err("tool_denied".into());
+    }
+    let op = Operation::McpRead {
+        server: request.server,
+        name: request.name,
+        arguments: request.arguments,
+        authority,
+    };
+    let req = CreateJob {
+        request_id: request.request_id,
+        input: Uuid::nil(),
+        operation: op,
+        timeout_seconds: 20,
+    };
+    let (status, Json(job)) =
+        start_admitted_job(rt.clone(), app.clone(), id, req, Some(b"{}"), true)?;
+    // Admission/duplicate responses carry only a handle. Evidence always goes
+    // through the fresh checks in result(), including after restart/lost reply.
+    Ok((status, Json(view(&job, None))))
+}
+fn view(job: &Job, result: Option<Value>) -> Value {
+    json!({"schema":"airouter.mcp.read.v1","request_id":job.request.request_id,"job_id":job.id,"phase":job.phase,"complete":job.phase.terminal(),"succeeded":job.phase==Phase::Succeeded&&result.as_ref().is_some_and(|r|r["isError"]!=true),"error":job.error,"result":result,"untrusted_evidence":true})
+}
+fn bound(job: &Job, caller: &Caller) -> Result<(String, Authority), Error> {
+    let Operation::McpRead {
+        server, authority, ..
+    } = &job.request.operation
+    else {
+        return Err("not_found".into());
+    };
+    if authority.subject != caller.subject || authority.context != caller.context {
+        return Err("scope_denied".into());
+    }
+    Ok((server.clone(), authority.clone()))
+}
+async fn result(
+    State(rt): State<Arc<Runtime>>,
+    Extension(app): Extension<Application>,
+    Path((id, request)): Path<(Uuid, Uuid)>,
+    headers: HeaderMap,
+) -> Result<Json<Value>, Error> {
+    let caller = caller(&headers)?;
+    let b = broker(&rt, &app, id)?;
+    let job = rt
+        .inner
+        .lock()
+        .unwrap()
+        .store
+        .job_by_request(&app.id, id, request)?;
+    let (server, authority) = bound(&job, &caller)?;
+    b.revalidate(&app.id, id, &server, &authority).await?;
+    let result = if job.phase == Phase::Succeeded {
+        if job.artifacts.len() != 1 || job.artifacts[0].bytes > super::MAX_RESULT as u64 {
+            return Err("mcp_response_invalid".into());
+        }
+        let bytes = rt.inner.lock().unwrap().store.read_registered_artifact(
+            &app.id,
+            id,
+            job.artifacts[0].id,
+        )?;
+        Some(
+            serde_json::from_slice::<Value>(&bytes)
+                .map_err(|_| Error::from("mcp_response_invalid"))?,
+        )
+    } else {
+        None
+    };
+    b.revalidate(&app.id, id, &server, &authority).await?;
+    Ok(Json(view(&job, result)))
+}
+async fn cancel(
+    State(rt): State<Arc<Runtime>>,
+    Extension(app): Extension<Application>,
+    Path((id, request)): Path<(Uuid, Uuid)>,
+    headers: HeaderMap,
+) -> Result<Json<Value>, Error> {
+    let caller = caller(&headers)?;
+    let b = broker(&rt, &app, id)?;
+    let job = rt
+        .inner
+        .lock()
+        .unwrap()
+        .store
+        .job_by_request(&app.id, id, request)?;
+    let (server, authority) = bound(&job, &caller)?;
+    b.revalidate(&app.id, id, &server, &authority).await?;
+    let mut inner = rt.inner.lock().unwrap();
+    let job = inner.store.cancel(&app.id, id, job.id)?;
+    if let Some((_, tx)) = inner.active.get(&job.id) {
+        let _ = tx.send(true);
+    }
+    Ok(Json(view(&job, None)))
+}
--- /dev/null
+++ b/crates/airouter-tools/src/mcp/executor.rs
@@ -0,0 +1,215 @@
+use super::{Authority, Binding, Broker, MAX_RESULT};
+use crate::{
+    protocol::Operation,
+    store::{digest, now},
+    worker::Execution,
+};
+use airouter_agent_protocol::{
+    Effect, HostExecution, MCP_VERSION, McpOperation, ServerGrant, ToolCall, ToolGrant,
+};
+use airouter_mcp::{EndpointRegistration, PreparedEndpoint, stdio_transport};
+use serde_json::{Value, json};
+use std::{sync::Arc, time::Duration};
+use tokio::sync::watch;
+use tokio_util::sync::CancellationToken;
+use uuid::Uuid;
+
+fn execution(
+    binding: &Binding,
+    authority: &Authority,
+    name: &str,
+    arguments: Value,
+    method: &str,
+    deadline: i64,
+) -> HostExecution {
+    let call = method == "tools/call";
+    HostExecution {
+        run_id: Uuid::nil(),
+        revision: binding.revision,
+        application: binding.application.clone(),
+        subject: authority.subject.clone(),
+        accounting_scope: binding.scope,
+        deadline,
+        server_grant: ServerGrant {
+            id: binding.server.clone(),
+            protocol_version: Some(MCP_VERSION.into()),
+            endpoint_binding_sha256: Some(authority.registration.clone()),
+            tools: binding
+                .tools
+                .iter()
+                .map(|n| ToolGrant {
+                    name: n.clone(),
+                    effect: Effect::ReadOnly,
+                })
+                .collect(),
+        },
+        operation: McpOperation {
+            id: Uuid::new_v4(),
+            server: binding.server.clone(),
+            request: json!({"jsonrpc":"2.0","id":1,"method":method,"params":if call{json!({"name":name,"arguments":arguments})}else{json!({})}}),
+            call: call.then(|| ToolCall {
+                id: "registered-read".into(),
+                name: name.into(),
+                arguments,
+            }),
+            digest: "registered-read".into(),
+            expires_at: deadline,
+            approval_actor: None,
+        },
+    }
+}
+impl Broker {
+    async fn endpoint(&self, b: &Binding, a: &Authority) -> Result<PreparedEndpoint, String> {
+        let bytes = super::registry::protected(&b.command, 64 * 1024 * 1024, false)?;
+        if digest(&bytes) != b.command_sha256 {
+            return Err("mcp_endpoint_changed".into());
+        }
+        drop(bytes);
+        let mut command = tokio::process::Command::new(&b.command);
+        command
+            .args([&b.socket, &b.credential_file])
+            .env_clear()
+            .current_dir("/");
+        let transport = stdio_transport(command).map_err(|_| "mcp_endpoint_unavailable")?;
+        tokio::time::timeout(
+            Duration::from_secs(5),
+            PreparedEndpoint::prepare(
+                EndpointRegistration {
+                    server_id: b.server.clone(),
+                    binding_sha256: a.registration.clone(),
+                    protocol_version: MCP_VERSION.into(),
+                },
+                transport,
+            ),
+        )
+        .await
+        .map_err(|_| "mcp_timeout")?
+        .map_err(|_| "mcp_endpoint_unavailable".into())
+    }
+    async fn invoke(
+        &self,
+        b: &Binding,
+        a: &Authority,
+        name: Option<&str>,
+        arguments: Value,
+        seconds: u64,
+        ct: CancellationToken,
+    ) -> Result<Value, String> {
+        let deadline = (now() + seconds) as i64;
+        let endpoint = tokio::select! {biased; _=ct.cancelled()=>return Err("cancelled".into()), e=self.endpoint(b,a)=>e?};
+        let result = async {
+            let list = endpoint
+                .handle_cancellable(
+                    &execution(b, a, "", json!({}), "tools/list", deadline),
+                    ct.clone(),
+                )
+                .await
+                .map_err(|_| "mcp_dispatch_failed")?;
+            let catalog = list["result"]["tools"]
+                .as_array()
+                .ok_or("mcp_catalog_invalid")?;
+            let tools: Vec<_> = catalog
+                .iter()
+                .filter(|t| b.tools.iter().any(|n| t["name"] == *n))
+                .cloned()
+                .collect();
+            if tools.len() != b.tools.len()
+                || tools.iter().any(|t| {
+                    t["annotations"]["readOnlyHint"] != true
+                        || t["annotations"]["destructiveHint"] == true
+                })
+            {
+                return Err("mcp_catalog_invalid".into());
+            }
+            let Some(name) = name else {
+                return Ok(json!({"server":b.server,"tools":tools}));
+            };
+            if !b.tools.iter().any(|n| n == name) {
+                return Err("tool_denied".into());
+            }
+            let value = endpoint
+                .handle_cancellable(
+                    &execution(b, a, name, arguments, "tools/call", deadline),
+                    ct.clone(),
+                )
+                .await
+                .map_err(|error| match error {
+                    airouter_mcp::DispatchError::CancelledUncertain
+                    | airouter_mcp::DispatchError::CancelledBeforeDispatch => "cancelled",
+                    airouter_mcp::DispatchError::DeadlineUncertain
+                    | airouter_mcp::DispatchError::DeadlineBeforeDispatch => "mcp_timeout",
+                    _ => "mcp_dispatch_failed",
+                })?;
+            let result = value.get("result").ok_or("mcp_response_invalid")?.clone();
+            if serde_json::to_vec(&result)
+                .map_err(|_| "mcp_response_invalid")?
+                .len()
+                > MAX_RESULT
+            {
+                return Err("output_limit".into());
+            }
+            Ok(result)
+        }
+        .await;
+        endpoint.close().await.map_err(|_| "cleanup_failed")?;
+        result
+    }
+    pub(crate) async fn discover(&self, b: &Binding, a: &Authority) -> Result<Value, String> {
+        let value = self
+            .invoke(b, a, None, json!({}), 10, CancellationToken::new())
+            .await?;
+        self.revalidate(&b.application, b.scope, &b.server, a)
+            .await?;
+        if serde_json::to_vec(&value)
+            .map_err(|_| "mcp_response_invalid")?
+            .len()
+            > MAX_RESULT
+        {
+            return Err("output_limit".into());
+        }
+        Ok(value)
+    }
+    pub(crate) async fn execute(
+        self: Arc<Self>,
+        app: String,
+        scope: Uuid,
+        operation: Operation,
+        seconds: u64,
+        mut cancelled: watch::Receiver<bool>,
+    ) -> Execution {
+        let Operation::McpRead {
+            server,
+            name,
+            arguments,
+            authority,
+        } = operation
+        else {
+            return Execution::failed("tool_denied");
+        };
+        let ct = CancellationToken::new();
+        let signal = ct.clone();
+        let watcher = tokio::spawn(async move {
+            if *cancelled.borrow() || cancelled.changed().await.is_ok() {
+                signal.cancel();
+            }
+        });
+        let result: Result<Vec<u8>, String> = async {
+            let b = self.revalidate(&app, scope, &server, &authority).await?;
+            let value = self
+                .invoke(&b, &authority, Some(&name), arguments, seconds, ct)
+                .await?;
+            self.revalidate(&app, scope, &server, &authority).await?;
+            serde_json::to_vec(&value).map_err(|_| "mcp_response_invalid".into())
+        }
+        .await;
+        watcher.abort();
+        match result {
+            Ok(bytes) => Execution {
+                error: None,
+                events: None,
+                outputs: vec![("mcp-result.json".into(), "application/json".into(), bytes)],
+            },
+            Err(e) => Execution::failed(e),
+        }
+    }
+}
--- /dev/null
+++ b/crates/airouter-tools/src/mcp/mod.rs
@@ -0,0 +1,65 @@
+//! Registered read-only MCP jobs. Launch bindings and data credentials stay in
+//! this host process; generic document workers and artifact routes cannot use them.
+pub(crate) mod api;
+mod executor;
+mod registry;
+pub use registry::{Binding, Broker, Registry};
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use uuid::Uuid;
+
+pub const TOOL: &str = "mcp_read";
+pub const MAX_RESULT: usize = 60 * 1024;
+pub const SUBJECT_HEADER: &str = "x-airouter-mcp-subject";
+pub const CONTEXT_HEADER: &str = "x-airouter-mcp-context";
+
+/// Assertions supplied by an authenticated frontend, never tool arguments.
+/// Frontends must revalidate the live human/session/run before every request and
+/// before delivering results. Host policy independently binds app/scope/subject.
+#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
+#[serde(deny_unknown_fields)]
+pub struct Authority {
+    pub subject: String,
+    pub context: String,
+    pub registration: String,
+    pub data_revision: String,
+}
+#[derive(Clone, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct ReadRequest {
+    pub request_id: Uuid,
+    pub server: String,
+    pub name: String,
+    pub arguments: Value,
+}
+#[derive(Clone, Debug)]
+pub struct Caller {
+    pub subject: String,
+    pub context: String,
+}
+impl Caller {
+    pub fn new(subject: &str, context: &str) -> Result<Self, String> {
+        if !identifier(subject)
+            || context.len() != 64
+            || !context
+                .bytes()
+                .all(|c| c.is_ascii_hexdigit() && !c.is_ascii_uppercase())
+        {
+            return Err("scope_denied".into());
+        }
+        Ok(Self {
+            subject: subject.into(),
+            context: context.into(),
+        })
+    }
+}
+pub(crate) fn identifier(s: &str) -> bool {
+    !s.is_empty()
+        && s.len() <= 128
+        && s.bytes()
+            .all(|b| b.is_ascii_alphanumeric() || b"-_.:/".contains(&b))
+}
+pub fn schema() -> Value {
+    serde_json::json!({"type":"object","description":"Use the dedicated /v1/scopes/{scope}/mcp routes with authenticated frontend subject/context headers. Generic utility jobs cannot dispatch this operation.",
+    "properties":{"request_id":{"type":"string","format":"uuid"},"server":{"type":"string"},"name":{"type":"string"},"arguments":{"type":"object"}},"required":["request_id","server","name","arguments"],"additionalProperties":false})
+}
--- /dev/null
+++ b/crates/airouter-tools/src/mcp/registry.rs
@@ -0,0 +1,216 @@
+use super::{Authority, Caller, identifier};
+use crate::store::{digest, now};
+use serde::{Deserialize, Serialize};
+use std::{
+    fs::OpenOptions,
+    io::Read,
+    os::unix::fs::{MetadataExt, OpenOptionsExt},
+    path::{Path, PathBuf},
+    time::Duration,
+};
+use uuid::Uuid;
+
+#[derive(Clone, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct Registry {
+    pub version: u32,
+    pub bindings: Vec<Binding>,
+}
+#[derive(Clone, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct Binding {
+    pub application: String,
+    pub scope: Uuid,
+    pub subject: String,
+    pub server: String,
+    pub enabled: bool,
+    pub revision: u64,
+    pub expires_at: u64,
+    pub command: PathBuf,
+    pub command_sha256: String,
+    pub socket: PathBuf,
+    pub credential_file: PathBuf,
+    pub authority_path: String,
+    pub revision_header: String,
+    pub tools: Vec<String>,
+}
+#[derive(Clone)]
+pub struct Broker {
+    pub(crate) registry: PathBuf,
+}
+
+pub(crate) fn protected(path: &Path, limit: usize, private: bool) -> Result<Vec<u8>, String> {
+    if !path.is_absolute()
+        || path
+            .components()
+            .any(|c| matches!(c, std::path::Component::ParentDir))
+    {
+        return Err("mcp_configuration_unavailable".into());
+    }
+    for parent in path.ancestors().skip(1) {
+        let m = std::fs::symlink_metadata(parent).map_err(|_| "mcp_configuration_unavailable")?;
+        // Root's sticky temporary directories cannot replace another owner's
+        // private child. All other ancestors must be protected directories.
+        if !m.is_dir()
+            || (m.uid() != 0 && m.uid() != nix::unistd::geteuid().as_raw())
+            || (m.mode() & 0o022 != 0 && !(m.uid() == 0 && m.mode() & 0o1000 != 0))
+        {
+            return Err("mcp_configuration_unavailable".into());
+        }
+    }
+    let file = OpenOptions::new()
+        .read(true)
+        .custom_flags(nix::libc::O_NOFOLLOW | nix::libc::O_CLOEXEC | nix::libc::O_NONBLOCK)
+        .open(path)
+        .map_err(|_| "mcp_configuration_unavailable")?;
+    let m = file
+        .metadata()
+        .map_err(|_| "mcp_configuration_unavailable")?;
+    if !m.is_file()
+        || m.nlink() != 1
+        || (m.uid() != 0 && m.uid() != nix::unistd::geteuid().as_raw())
+        || m.mode() & (if private { 0o077 } else { 0o022 }) != 0
+        || m.len() > limit as u64
+    {
+        return Err("mcp_configuration_unavailable".into());
+    }
+    let mut bytes = Vec::new();
+    file.take(limit as u64 + 1)
+        .read_to_end(&mut bytes)
+        .map_err(|_| "mcp_configuration_unavailable")?;
+    if bytes.len() > limit {
+        return Err("mcp_configuration_unavailable".into());
+    }
+    Ok(bytes)
+}
+impl Broker {
+    pub fn new(registry: PathBuf) -> Self {
+        Self { registry }
+    }
+    pub(crate) fn bindings(
+        &self,
+        app: &str,
+        scope: Uuid,
+        caller: &Caller,
+    ) -> Result<Vec<Binding>, String> {
+        let bytes = protected(&self.registry, 128 * 1024, true)?;
+        let r: Registry =
+            serde_json::from_slice(&bytes).map_err(|_| "mcp_configuration_unavailable")?;
+        if r.version != 1 || r.bindings.len() > 32 {
+            return Err("mcp_configuration_unavailable".into());
+        }
+        let mut seen = std::collections::HashSet::new();
+        for b in &r.bindings {
+            if !identifier(&b.application)
+                || !identifier(&b.subject)
+                || !identifier(&b.server)
+                || b.scope.is_nil()
+                || b.revision == 0
+                || !b.command.is_absolute()
+                || !b.socket.is_absolute()
+                || !b.credential_file.is_absolute()
+                || b.command_sha256.len() != 64
+                || !b.command_sha256.bytes().all(|c| c.is_ascii_hexdigit())
+                || !b.authority_path.starts_with('/')
+                || b.authority_path.starts_with("//")
+                || b.authority_path.contains(['?', '#', '\r', '\n'])
+                || b.revision_header.is_empty()
+                || !b
+                    .revision_header
+                    .bytes()
+                    .all(|c| c.is_ascii_lowercase() || c == b'-')
+                || b.tools.is_empty()
+                || b.tools.len() > 16
+                || b.tools.iter().any(|s| !identifier(s))
+                || !seen.insert((&b.application, b.scope, &b.subject, &b.server))
+            {
+                return Err("mcp_configuration_unavailable".into());
+            }
+        }
+        Ok(r.bindings
+            .into_iter()
+            .filter(|b| {
+                b.application == app
+                    && b.scope == scope
+                    && b.subject == caller.subject
+                    && b.enabled
+                    && b.expires_at > now()
+            })
+            .collect())
+    }
+    pub(crate) async fn authorize(
+        &self,
+        app: &str,
+        scope: Uuid,
+        caller: &Caller,
+        server: &str,
+    ) -> Result<(Binding, Authority), String> {
+        let b = self
+            .bindings(app, scope, caller)?
+            .into_iter()
+            .find(|b| b.server == server)
+            .ok_or("scope_denied")?;
+        let token = protected(&b.credential_file, 1025, true)?;
+        let token = std::str::from_utf8(&token)
+            .map_err(|_| "mcp_configuration_unavailable")?
+            .trim_end_matches('\n');
+        if !(32..=1024).contains(&token.len())
+            || !token
+                .bytes()
+                .all(|c| c.is_ascii_alphanumeric() || b"-._~+/=".contains(&c))
+        {
+            return Err("mcp_configuration_unavailable".into());
+        }
+        let client = reqwest::Client::builder()
+            .unix_socket(b.socket.as_path())
+            .no_proxy()
+            .redirect(reqwest::redirect::Policy::none())
+            .timeout(Duration::from_secs(2))
+            .build()
+            .map_err(|_| "mcp_authorization_unavailable")?;
+        let response = client
+            .get(format!("http://localhost{}", b.authority_path))
+            .bearer_auth(token)
+            .send()
+            .await
+            .map_err(|_| "mcp_authorization_unavailable")?;
+        if !response.status().is_success() {
+            return Err(if matches!(response.status().as_u16(), 401 | 403) {
+                "scope_denied"
+            } else {
+                "mcp_authorization_unavailable"
+            }
+            .into());
+        }
+        let revision = response
+            .headers()
+            .get(&b.revision_header)
+            .and_then(|v| v.to_str().ok())
+            .filter(|s| s.len() == 64 && s.bytes().all(|c| c.is_ascii_hexdigit()))
+            .ok_or("mcp_authorization_unavailable")?
+            .to_owned();
+        let authority = Authority {
+            subject: caller.subject.clone(),
+            context: caller.context.clone(),
+            registration: digest(
+                &serde_json::to_vec(&b).map_err(|_| "mcp_configuration_unavailable")?,
+            ),
+            data_revision: revision,
+        };
+        Ok((b, authority))
+    }
+    pub(crate) async fn revalidate(
+        &self,
+        app: &str,
+        scope: Uuid,
+        server: &str,
+        expected: &Authority,
+    ) -> Result<Binding, String> {
+        let caller = Caller::new(&expected.subject, &expected.context)?;
+        let (b, current) = self.authorize(app, scope, &caller, server).await?;
+        if &current != expected {
+            return Err("scope_denied".into());
+        }
+        Ok(b)
+    }
+}
--- a/crates/airouter-tools/src/protocol.rs
+++ b/crates/airouter-tools/src/protocol.rs
@@ -17,6 +17,13 @@
     deny_unknown_fields
 )]
 pub enum Operation {
+    /// Admitted only through the dedicated registered MCP boundary.
+    McpRead {
+        server: String,
+        name: String,
+        arguments: serde_json::Value,
+        authority: crate::mcp::Authority,
+    },
     DocumentConvert {
         from: DocumentFormat,
         to: DocumentFormat,
@@ -69,6 +76,7 @@
 impl Operation {
     pub fn name(&self) -> &'static str {
         match self {
+            Self::McpRead { .. } => "mcp_read",
             Self::DocumentConvert { .. } => "document_convert",
             Self::PdfInspect { .. } => "pdf_inspect",
             Self::PdfExtract { .. } => "pdf_extract",
--- a/crates/airouter-tools/src/store.rs
+++ b/crates/airouter-tools/src/store.rs
@@ -130,6 +130,19 @@
         }
         let cutoff = now().saturating_sub(RETENTION_SECONDS);
         let mut next = self.ledger.clone();
+        // The job carries the private artifact policy. Remove its inputs and
+        // outputs atomically with that policy, even when output was created
+        // seconds later and has not reached its own retention cutoff yet.
+        let retired_private: std::collections::HashSet<Uuid> = next
+            .jobs
+            .values()
+            .filter(|j| {
+                j.phase.terminal()
+                    && j.created_at < cutoff
+                    && matches!(j.request.operation, Operation::McpRead { .. })
+            })
+            .flat_map(|j| std::iter::once(j.request.input).chain(j.artifacts.iter().map(|a| a.id)))
+            .collect();
         next.jobs
             .retain(|_, j| !j.phase.terminal() || j.created_at >= cutoff);
         let running: Vec<Uuid> = next
@@ -141,7 +154,10 @@
         let expired: Vec<Uuid> = next
             .artifacts
             .values()
-            .filter(|a| a.created_at < cutoff && !running.contains(&a.id))
+            .filter(|a| {
+                retired_private.contains(&a.id)
+                    || (a.created_at < cutoff && !running.contains(&a.id))
+            })
             .map(|a| a.id)
             .collect();
         for id in &expired {
@@ -166,7 +182,26 @@
             .cloned()
             .ok_or_else(|| "not_found".into())
     }
+    pub fn public_artifact(&self, app: &str, scope: Uuid, id: Uuid) -> Result<(), String> {
+        self.artifact(app, scope, id)?;
+        if self.ledger.jobs.values().any(|job| {
+            matches!(job.request.operation, Operation::McpRead { .. })
+                && (job.request.input == id || job.artifacts.iter().any(|a| a.id == id))
+        }) {
+            return Err("scope_denied".into());
+        }
+        Ok(())
+    }
     pub fn read_artifact(&self, app: &str, scope: Uuid, id: Uuid) -> Result<Vec<u8>, String> {
+        self.public_artifact(app, scope, id)?;
+        self.read_registered_artifact(app, scope, id)
+    }
+    pub(crate) fn read_registered_artifact(
+        &self,
+        app: &str,
+        scope: Uuid,
+        id: Uuid,
+    ) -> Result<Vec<u8>, String> {
         let a = self.artifact(app, scope, id)?;
         let f = OpenOptions::new()
             .read(true)
--- a/crates/airouter-tools/src/web.rs
+++ b/crates/airouter-tools/src/web.rs
@@ -3,8 +3,11 @@
 //! This module validates every URL before connection and every redirect after
 //! resolution. It deliberately contains no model or tool-specific behavior.
 
-use std::{net::{IpAddr, SocketAddr}, time::Duration};
 use serde::Serialize;
+use std::{
+    net::{IpAddr, SocketAddr},
+    time::Duration,
+};
 use tokio::net::lookup_host;
 use url::Url;
 
@@ -30,26 +33,49 @@
     let mut results = Vec::new();
     let mut cursor = 0;
     while results.len() < max_results {
-        let Some(anchor) = html[cursor..].find("class=\"result__a\"") else { break };
+        let Some(anchor) = html[cursor..].find("class=\"result__a\"") else {
+            break;
+        };
         let start = cursor + anchor;
-        let Some(tag_start) = html[..start].rfind('<') else { break };
-        let Some(tag_end_rel) = html[start..].find('>') else { break };
+        let Some(tag_start) = html[..start].rfind('<') else {
+            break;
+        };
+        let Some(tag_end_rel) = html[start..].find('>') else {
+            break;
+        };
         let tag_end = start + tag_end_rel;
-        let Some(close_rel) = html[tag_end + 1..].find("</a>") else { break };
+        let Some(close_rel) = html[tag_end + 1..].find("</a>") else {
+            break;
+        };
         let close = tag_end + 1 + close_rel;
         let tag = &html[tag_start..=tag_end];
-        let Some(raw_href) = attribute(tag, "href") else { cursor = close + 4; continue };
+        let Some(raw_href) = attribute(tag, "href") else {
+            cursor = close + 4;
+            continue;
+        };
         let url = decode_result_url(&raw_href);
-        if !url.starts_with("http://") && !url.starts_with("https://") { cursor = close + 4; continue; }
+        if !url.starts_with("http://") && !url.starts_with("https://") {
+            cursor = close + 4;
+            continue;
+        }
         let title = clean_text(&html[tag_end + 1..close]);
         let after = &html[close + 4..];
-        let snippet = after.find("class=\"result__snippet\"").and_then(|offset| {
-            let s = close + 4 + offset;
-            let end = html[s..].find('>')? + s;
-            let finish = html[end + 1..].find("</a>")? + end + 1;
-            Some(clean_text(&html[end + 1..finish]))
-        }).unwrap_or_default();
-        if !title.is_empty() { results.push(SearchResult { title, url, snippet }); }
+        let snippet = after
+            .find("class=\"result__snippet\"")
+            .and_then(|offset| {
+                let s = close + 4 + offset;
+                let end = html[s..].find('>')? + s;
+                let finish = html[end + 1..].find("</a>")? + end + 1;
+                Some(clean_text(&html[end + 1..finish]))
+            })
+            .unwrap_or_default();
+        if !title.is_empty() {
+            results.push(SearchResult {
+                title,
+                url,
+                snippet,
+            });
+        }
         cursor = close + 4;
     }
     results
@@ -63,10 +89,19 @@
 
 fn decode_result_url(raw: &str) -> String {
     let href = html_unescape(raw);
-    let absolute = if href.starts_with("//") { format!("https:{href}") } else { href };
-    Url::parse(&absolute).ok().and_then(|url| {
-        url.query_pairs().find(|(k, _)| k == "uddg").map(|(_, v)| v.into_owned())
-    }).unwrap_or(absolute)
+    let absolute = if href.starts_with("//") {
+        format!("https:{href}")
+    } else {
+        href
+    };
+    Url::parse(&absolute)
+        .ok()
+        .and_then(|url| {
+            url.query_pairs()
+                .find(|(k, _)| k == "uddg")
+                .map(|(_, v)| v.into_owned())
+        })
+        .unwrap_or(absolute)
 }
 
 fn clean_text(raw: &str) -> String {
@@ -85,7 +120,10 @@
             _ => {}
         }
     }
-    html_unescape(&out).split_whitespace().collect::<Vec<_>>().join(" ")
+    html_unescape(&out)
+        .split_whitespace()
+        .collect::<Vec<_>>()
+        .join(" ")
 }
 
 /// Remove executable/style blocks before turning markup into plain text.
@@ -93,8 +131,16 @@
     let mut remaining = html.to_owned();
     for tag in ["script", "style", "noscript"] {
         loop {
-            let Some(start) = remaining.to_ascii_lowercase().find(&format!("<{tag}")) else { break };
-            let Some(end_rel) = remaining[start..].to_ascii_lowercase().find(&format!("</{tag}>")) else { remaining.replace_range(start.., ""); break };
+            let Some(start) = remaining.to_ascii_lowercase().find(&format!("<{tag}")) else {
+                break;
+            };
+            let Some(end_rel) = remaining[start..]
+                .to_ascii_lowercase()
+                .find(&format!("</{tag}>"))
+            else {
+                remaining.replace_range(start.., "");
+                break;
+            };
             remaining.replace_range(start..start + end_rel + tag.len() + 3, " ");
         }
     }
@@ -102,9 +148,13 @@
 }
 
 fn html_unescape(raw: &str) -> String {
-    raw.replace("&amp;", "&").replace("&quot;", "\"")
-        .replace("&#x27;", "'").replace("&#39;", "'")
-        .replace("&lt;", "<").replace("&gt;", ">").replace("&nbsp;", " ")
+    raw.replace("&amp;", "&")
+        .replace("&quot;", "\"")
+        .replace("&#x27;", "'")
+        .replace("&#39;", "'")
+        .replace("&lt;", "<")
+        .replace("&gt;", ">")
+        .replace("&nbsp;", " ")
 }
 
 impl Default for Policy {
@@ -166,7 +216,10 @@
 pub async fn resolve_public(raw: &str) -> Result<(Url, Vec<SocketAddr>), Error> {
     let url = validate_syntax(raw)?;
     let host = url.host_str().ok_or(Error::InvalidUrl)?;
-    if url.host().is_some_and(|h| matches!(h, url::Host::Ipv4(_) | url::Host::Ipv6(_))) {
+    if url
+        .host()
+        .is_some_and(|h| matches!(h, url::Host::Ipv4(_) | url::Host::Ipv6(_)))
+    {
         let port = url.port_or_known_default().ok_or(Error::InvalidUrl)?;
         let ip = host.parse::<IpAddr>().map_err(|_| Error::InvalidUrl)?;
         return Ok((url, vec![SocketAddr::new(ip, port)]));
@@ -184,7 +237,10 @@
     if addresses.is_empty() {
         return Err(Error::DnsResolutionFailed);
     }
-    if addresses.iter().any(|address| is_private_or_reserved(address.ip())) {
+    if addresses
+        .iter()
+        .any(|address| is_private_or_reserved(address.ip()))
+    {
         return Err(Error::PrivateAddress);
     }
     Ok(())
@@ -222,14 +278,29 @@
 
     #[test]
     fn rejects_non_web_urls_and_credentials() {
-        assert_eq!(validate_syntax("file:///etc/passwd"), Err(Error::UnsupportedScheme));
-        assert_eq!(validate_syntax("https://user:pass@example.com"), Err(Error::CredentialsInUrl));
+        assert_eq!(
+            validate_syntax("file:///etc/passwd"),
+            Err(Error::UnsupportedScheme)
+        );
+        assert_eq!(
+            validate_syntax("https://user:pass@example.com"),
+            Err(Error::CredentialsInUrl)
+        );
     }
 
     #[test]
     fn rejects_private_ip_literals() {
-        for host in ["127.0.0.1", "10.0.0.1", "192.168.1.1", "169.254.1.1", "[::1]"] {
-            assert_eq!(validate_syntax(&format!("https://{host}/")), Err(Error::PrivateAddress));
+        for host in [
+            "127.0.0.1",
+            "10.0.0.1",
+            "192.168.1.1",
+            "169.254.1.1",
+            "[::1]",
+        ] {
+            assert_eq!(
+                validate_syntax(&format!("https://{host}/")),
+                Err(Error::PrivateAddress)
+            );
         }
     }
 
@@ -244,20 +315,36 @@
             "1.1.1.1:443".parse().unwrap(),
             "10.0.0.1:443".parse().unwrap(),
         ];
-        assert_eq!(validate_resolved_addresses(&addresses), Err(Error::PrivateAddress));
-        assert_eq!(validate_resolved_addresses(&[]), Err(Error::DnsResolutionFailed));
+        assert_eq!(
+            validate_resolved_addresses(&addresses),
+            Err(Error::PrivateAddress)
+        );
+        assert_eq!(
+            validate_resolved_addresses(&[]),
+            Err(Error::DnsResolutionFailed)
+        );
     }
 
     #[test]
     fn parses_bounded_search_results_and_unwraps_redirects() {
         let html = r#"<a class="result__a" href="//duckduckgo.com/l/?uddg=https%3A%2F%2Fexample.com%2F&amp;rut=x">Example &amp; Site</a><a class="result__snippet" href="x">A <b>snippet</b>.</a>"#;
-        assert_eq!(parse_search_html(html, 2), vec![SearchResult {
-            title: "Example & Site".into(), url: "https://example.com/".into(), snippet: "A snippet .".into()
-        }]);
+        assert_eq!(
+            parse_search_html(html, 2),
+            vec![SearchResult {
+                title: "Example & Site".into(),
+                url: "https://example.com/".into(),
+                snippet: "A snippet .".into()
+            }]
+        );
     }
 
     #[test]
     fn extracts_text_without_script_or_style_content() {
-        assert_eq!(extract_html_text("<style>x{}</style><h1>Hello</h1><script>alert(1)</script><p>world</p>"), "Hello world");
-    }
-}
+        assert_eq!(
+            extract_html_text(
+                "<style>x{}</style><h1>Hello</h1><script>alert(1)</script><p>world</p>"
+            ),
+            "Hello world"
+        );
+    }
+}
--- a/crates/airouter-tools/src/worker.rs
+++ b/crates/airouter-tools/src/worker.rs
@@ -16,12 +16,12 @@
     process::Stdio,
     time::{Duration, SystemTime, UNIX_EPOCH},
 };
-use uuid::Uuid;
 use tokio::{
     io::{AsyncRead, AsyncReadExt, AsyncWriteExt},
     process::Command,
     sync::watch,
 };
+use uuid::Uuid;
 
 #[derive(Deserialize)]
 #[serde(deny_unknown_fields)]
@@ -32,8 +32,12 @@
     #[serde(default = "default_excerpt")]
     max_excerpt_bytes: usize,
 }
-fn default_sources() -> usize { 3 }
-fn default_excerpt() -> usize { 8192 }
+fn default_sources() -> usize {
+    3
+}
+fn default_excerpt() -> usize {
+    8192
+}
 const RESEARCH_CACHE_MAX_BYTES: u64 = 512 * 1024 * 1024;
 const RESEARCH_CACHE_SWEEP_INTERVAL: Duration = Duration::from_secs(15 * 60);
 const RESEARCH_CACHE_TEMP_MAX_AGE: Duration = Duration::from_secs(5 * 60);
@@ -306,7 +310,8 @@
         };
         match self.fetch_web_bytes(&raw, seconds, cancelled).await {
             Ok((content_type, bytes, final_url)) => {
-                let mut data = format!("URL: {final_url}\nContent-Type: {content_type}\n\n").into_bytes();
+                let mut data =
+                    format!("URL: {final_url}\nContent-Type: {content_type}\n\n").into_bytes();
                 data.extend_from_slice(&bytes);
                 Execution {
                     error: None,
@@ -348,7 +353,11 @@
                     Ok(data) => data,
                     Err(_) => return Execution::failed("web_parse_failed"),
                 };
-                Execution { error: None, events: None, outputs: vec![("search.json".into(), "application/json".into(), data)] }
+                Execution {
+                    error: None,
+                    events: None,
+                    outputs: vec![("search.json".into(), "application/json".into(), data)],
+                }
             }
             Err(error) => Execution::failed(error),
         }
@@ -364,9 +373,13 @@
             total_timeout: Duration::from_secs(seconds.min(60)),
             ..Policy::default()
         };
-        fetch_with_controls(policy, raw.to_owned(), cancelled, |policy, raw| async move {
-            fetch_url(&policy, &raw).await
-        }).await
+        fetch_with_controls(
+            policy,
+            raw.to_owned(),
+            cancelled,
+            |policy, raw| async move { fetch_url(&policy, &raw).await },
+        )
+        .await
     }
 
     async fn extract_web(
@@ -404,7 +417,11 @@
                     Ok(data) => data,
                     Err(_) => return Execution::failed("web_parse_failed"),
                 };
-                Execution { error: None, events: None, outputs: vec![("extracted.json".into(), "application/json".into(), data)] }
+                Execution {
+                    error: None,
+                    events: None,
+                    outputs: vec![("extracted.json".into(), "application/json".into(), data)],
+                }
             }
             Err(error) => Execution::failed(error),
         }
@@ -443,9 +460,11 @@
             ) => result.map_err(|_| "web_timeout".to_string()).and_then(|r| r),
         };
         match result {
-            Ok(data) => {
-                Execution { error: None, events: None, outputs: vec![("research.json".into(), "application/json".into(), data)] }
-            }
+            Ok(data) => Execution {
+                error: None,
+                events: None,
+                outputs: vec![("research.json".into(), "application/json".into(), data)],
+            },
             Err(error) => Execution::failed(error),
         }
     }
@@ -457,21 +476,50 @@
         seconds: u64,
         cancelled: watch::Receiver<bool>,
     ) -> Execution {
-        if *cancelled.borrow() { return Execution::failed("cancelled"); }
+        if *cancelled.borrow() {
+            return Execution::failed("cancelled");
+        }
         let bytes = match tokio::fs::read(&input).await {
             Ok(bytes) if !bytes.is_empty() && bytes.len() <= 16 * 1024 * 1024 => bytes,
             _ => return Execution::failed("input_size_limit"),
         };
         if bytes.starts_with(b"%PDF-") {
-            let operation = if ocr { Operation::OcrPdf { language: "eng".into() } } else { Operation::PdfExtract { layout: false } };
-            return self.run(input, operation, seconds, cancelled).await.unwrap_or_else(Execution::failed);
-        }
-        if bytes.starts_with(b"\x89PNG") || bytes.starts_with(&[0xff, 0xd8, 0xff]) || bytes.starts_with(b"GIF8") || bytes.get(0..4) == Some(b"RIFF") && bytes.get(8..12) == Some(b"WEBP") {
-            return self.run(input, Operation::OcrImage { language: "eng".into() }, seconds, cancelled).await.unwrap_or_else(Execution::failed);
+            let operation = if ocr {
+                Operation::OcrPdf {
+                    language: "eng".into(),
+                }
+            } else {
+                Operation::PdfExtract { layout: false }
+            };
+            return self
+                .run(input, operation, seconds, cancelled)
+                .await
+                .unwrap_or_else(Execution::failed);
+        }
+        if bytes.starts_with(b"\x89PNG")
+            || bytes.starts_with(&[0xff, 0xd8, 0xff])
+            || bytes.starts_with(b"GIF8")
+            || bytes.get(0..4) == Some(b"RIFF") && bytes.get(8..12) == Some(b"WEBP")
+        {
+            return self
+                .run(
+                    input,
+                    Operation::OcrImage {
+                        language: "eng".into(),
+                    },
+                    seconds,
+                    cancelled,
+                )
+                .await
+                .unwrap_or_else(Execution::failed);
         }
         let raw = String::from_utf8_lossy(&bytes);
         let html = raw.trim_start().starts_with("<") || raw.to_ascii_lowercase().contains("<html");
-        let text = if html { web::extract_html_text(&raw) } else { raw.split_whitespace().collect::<Vec<_>>().join(" ") };
+        let text = if html {
+            web::extract_html_text(&raw)
+        } else {
+            raw.split_whitespace().collect::<Vec<_>>().join(" ")
+        };
         let data = match serde_json::to_vec(&serde_json::json!({
             "format": if html { "html" } else { "text" },
             "bytes": bytes.len(),
@@ -482,7 +530,11 @@
             Ok(data) => data,
             Err(_) => return Execution::failed("document_parse_failed"),
         };
-        Execution { error: None, events: None, outputs: vec![("document.json".into(), "application/json".into(), data)] }
+        Execution {
+            error: None,
+            events: None,
+            outputs: vec![("document.json".into(), "application/json".into(), data)],
+        }
     }
 
     async fn research_inner(
@@ -491,17 +543,24 @@
         seconds: u64,
         cancelled: watch::Receiver<bool>,
     ) -> Result<Vec<u8>, String> {
-        let encoded: String = url::form_urlencoded::byte_serialize(request.question.as_bytes()).collect();
+        let encoded: String =
+            url::form_urlencoded::byte_serialize(request.question.as_bytes()).collect();
         let endpoint = format!("https://html.duckduckgo.com/html/?q={encoded}");
-        let (_, search_bytes, search_url) = self.fetch_web_bytes(&endpoint, seconds, cancelled.clone()).await?;
+        let (_, search_bytes, search_url) = self
+            .fetch_web_bytes(&endpoint, seconds, cancelled.clone())
+            .await?;
         let search_html = String::from_utf8_lossy(&search_bytes);
         let candidates = web::parse_search_html(&search_html, request.max_sources * 2);
         let mut seen = std::collections::HashSet::new();
         let mut sources = Vec::new();
         let mut errors = Vec::new();
         for candidate in candidates {
-            if sources.len() >= request.max_sources || !seen.insert(candidate.url.clone()) { continue; }
-            let fetched = self.fetch_web_bytes(&candidate.url, seconds, cancelled.clone()).await;
+            if sources.len() >= request.max_sources || !seen.insert(candidate.url.clone()) {
+                continue;
+            }
+            let fetched = self
+                .fetch_web_bytes(&candidate.url, seconds, cancelled.clone())
+                .await;
             match fetched {
                 Ok((content_type, bytes, final_url)) if is_supported_text_content_type(&content_type) => {
                     let text = if content_type == "text/html" { web::extract_html_text(&String::from_utf8_lossy(&bytes)) } else { String::from_utf8_lossy(&bytes).split_whitespace().collect::<Vec<_>>().join(" ") };
@@ -523,7 +582,6 @@
         }
         encode_research_result(&request.question, &search_url, sources, errors)
     }
-
 }
 
 async fn fetch_with_controls<F, Fut>(
@@ -573,13 +631,23 @@
 }
 
 fn truncate_utf8(value: &str, max_bytes: usize) -> String {
-    if value.len() <= max_bytes { return value.to_owned(); }
-    let end = value.char_indices().take_while(|(i, _)| *i < max_bytes.saturating_sub(1)).last().map(|(i, c)| i + c.len_utf8()).unwrap_or(0);
+    if value.len() <= max_bytes {
+        return value.to_owned();
+    }
+    let end = value
+        .char_indices()
+        .take_while(|(i, _)| *i < max_bytes.saturating_sub(1))
+        .last()
+        .map(|(i, c)| i + c.len_utf8())
+        .unwrap_or(0);
     format!("{}…", &value[..end])
 }
 
 fn is_supported_text_content_type(content_type: &str) -> bool {
-    matches!(content_type, "text/html" | "text/plain" | "application/json" | "application/xml" | "text/xml")
+    matches!(
+        content_type,
+        "text/html" | "text/plain" | "application/json" | "application/xml" | "text/xml"
+    )
 }
 
 fn encode_research_result(
@@ -595,11 +663,15 @@
         "sources": sources,
         "errors": errors,
         "untrusted_content": true,
-    })).map_err(|_| "web_parse_failed".into())
+    }))
+    .map_err(|_| "web_parse_failed".into())
 }
 
 fn now_seconds() -> u64 {
-    SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0)
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_secs())
+        .unwrap_or(0)
 }
 
 async fn research_cache_get_or_run<F, Fut>(
@@ -611,7 +683,9 @@
     F: FnOnce() -> Fut,
     Fut: Future<Output = Result<Vec<u8>, String>>,
 {
-    let hash = key.strip_prefix("sha256:").ok_or("research_cache_key_invalid")?;
+    let hash = key
+        .strip_prefix("sha256:")
+        .ok_or("research_cache_key_invalid")?;
     let path = cache_dir.join(hash).with_extension("json");
     if let Some(data) = read_research_cache_entry(&path, now_seconds()).await {
         return Ok(data);
@@ -638,7 +712,9 @@
 async fn read_research_cache_entry(path: &Path, now: u64) -> Option<Vec<u8>> {
     let discard = || {
         let path = path.to_owned();
-        async move { let _ = tokio::fs::remove_file(path).await; }
+        async move {
+            let _ = tokio::fs::remove_file(path).await;
+        }
     };
     let Some(value) = tokio::fs::read(path)
         .await
@@ -648,13 +724,23 @@
         discard().await;
         return None;
     };
-    if value.get("expires_at").and_then(serde_json::Value::as_u64).is_none_or(|expires| expires <= now) {
+    if value
+        .get("expires_at")
+        .and_then(serde_json::Value::as_u64)
+        .is_none_or(|expires| expires <= now)
+    {
         discard().await;
         return None;
     }
-    let data = value.get("data_base64").and_then(serde_json::Value::as_str)
+    let data = value
+        .get("data_base64")
+        .and_then(serde_json::Value::as_str)
         .and_then(|encoded| STANDARD.decode(encoded).ok())
-        .or_else(|| value.get("data").and_then(|data| serde_json::to_vec(data).ok()));
+        .or_else(|| {
+            value
+                .get("data")
+                .and_then(|data| serde_json::to_vec(data).ok())
+        });
     let Some(data) = data else {
         discard().await;
         return None;
@@ -667,24 +753,29 @@
     }
     let path_for_touch = path.to_owned();
     let _ = tokio::task::spawn_blocking(move || {
-        std::fs::File::open(path_for_touch)?.set_times(
-            std::fs::FileTimes::new().set_modified(SystemTime::now()),
-        )
+        std::fs::File::open(path_for_touch)?
+            .set_times(std::fs::FileTimes::new().set_modified(SystemTime::now()))
     })
     .await;
     Some(data)
 }
 
 async fn prune_research_cache_dir(path: &Path, now: u64, max_bytes: u64) {
-    let Ok(mut entries) = tokio::fs::read_dir(path).await else { return };
+    let Ok(mut entries) = tokio::fs::read_dir(path).await else {
+        return;
+    };
     let mut files = Vec::new();
     let mut total = 0u64;
     while let Ok(Some(entry)) = entries.next_entry().await {
-        let Ok(file_type) = entry.file_type().await else { continue };
+        let Ok(file_type) = entry.file_type().await else {
+            continue;
+        };
         if !file_type.is_file() {
             continue;
         }
-        let Ok(metadata) = entry.metadata().await else { continue };
+        let Ok(metadata) = entry.metadata().await else {
+            continue;
+        };
         let file_path = entry.path();
         if file_path.extension().is_some_and(|ext| ext == "tmp") {
             let stale = metadata
@@ -722,7 +813,9 @@
         // eviction order while keeping the cached payload immutable.
         files.sort_by_key(|(_, _, modified)| *modified);
         for (file_path, size, _) in files {
-            if total <= max_bytes { break; }
+            if total <= max_bytes {
+                break;
+            }
             if tokio::fs::remove_file(file_path).await.is_ok() {
                 total = total.saturating_sub(size);
             }
@@ -730,13 +823,11 @@
     }
 }
 
-async fn fetch_url(
-    policy: &Policy,
-    raw: &str,
-) -> Result<(String, Vec<u8>, String), String> {
+async fn fetch_url(policy: &Policy, raw: &str) -> Result<(String, Vec<u8>, String), String> {
     fetch_url_with_resolver(policy, raw, |raw| async move {
         web::resolve_public(&raw).await.map_err(|e| e.to_string())
-    }).await
+    })
+    .await
 }
 
 async fn fetch_url_with_resolver<F, Fut>(
@@ -942,7 +1033,9 @@
         })
     }
 
-    fn fixture_resolver(port: u16) -> impl FnMut(String) -> std::future::Ready<Result<(url::Url, Vec<SocketAddr>), String>> {
+    fn fixture_resolver(
+        port: u16,
+    ) -> impl FnMut(String) -> std::future::Ready<Result<(url::Url, Vec<SocketAddr>), String>> {
         move |raw| {
             let parsed = url::Url::parse(&raw).map_err(|_| "invalid_url".to_string());
             std::future::ready(parsed.and_then(|url| {
@@ -963,7 +1056,8 @@
             "HTTP/1.1 302 Found\r\nLocation: http://127.0.0.1:9/private\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".into(),
         ).await;
         let raw = format!("http://fetch-fixture.invalid:{port}/start");
-        let result = fetch_url_with_resolver(&Policy::default(), &raw, fixture_resolver(port)).await;
+        let result =
+            fetch_url_with_resolver(&Policy::default(), &raw, fixture_resolver(port)).await;
         assert_eq!(result, Err("private_address".into()));
         server.await.unwrap();
     }
@@ -972,9 +1066,10 @@
     async fn web_fetch_uses_the_address_resolved_before_dns_rebinding() {
         let safe_listener = TcpListener::bind(("127.0.0.1", 0)).await.unwrap();
         let safe_addr = safe_listener.local_addr().unwrap();
-        let rebound_listener = TcpListener::bind(SocketAddr::from(([127, 0, 0, 2], safe_addr.port())))
-            .await
-            .unwrap();
+        let rebound_listener =
+            TcpListener::bind(SocketAddr::from(([127, 0, 0, 2], safe_addr.port())))
+                .await
+                .unwrap();
         let safe_server = serve_once(safe_listener,
             "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: 4\r\nConnection: close\r\n\r\nsafe".into());
         let mut rebound_server = serve_once(rebound_listener,
@@ -985,9 +1080,15 @@
             // First answer is safe; any subsequent lookup observes the simulated
             // rebinding to 127.0.0.2. The request must use the first pinned answer.
             let lookup = count_for_resolver.fetch_add(1, Ordering::SeqCst);
-            let ip = if lookup == 0 { [127, 0, 0, 1] } else { [127, 0, 0, 2] };
+            let ip = if lookup == 0 {
+                [127, 0, 0, 1]
+            } else {
+                [127, 0, 0, 2]
+            };
             let parsed = url::Url::parse(&raw).map_err(|_| "invalid_url".to_string());
-            std::future::ready(parsed.map(|url| (url, vec![SocketAddr::from((ip, safe_addr.port()))])))
+            std::future::ready(
+                parsed.map(|url| (url, vec![SocketAddr::from((ip, safe_addr.port()))])),
+            )
         };
         let raw = format!("http://rebinding-fixture.invalid:{}/page", safe_addr.port());
         let (content_type, body, _) = fetch_url_with_resolver(&Policy::default(), &raw, resolver)
@@ -997,7 +1098,11 @@
         assert_eq!(content_type, "text/plain");
         assert_eq!(body, b"safe");
         assert_eq!(resolution_count.load(Ordering::SeqCst), 1);
-        assert!(tokio::time::timeout(Duration::from_millis(25), &mut rebound_server).await.is_err());
+        assert!(
+            tokio::time::timeout(Duration::from_millis(25), &mut rebound_server)
+                .await
+                .is_err()
+        );
         rebound_server.abort();
         safe_server.await.unwrap();
     }
@@ -1010,7 +1115,10 @@
             body.len(), String::from_utf8_lossy(body)
         )).await;
         let raw = format!("http://fetch-fixture.invalid:{port}/large");
-        let policy = Policy { max_bytes: 4, ..Policy::default() };
+        let policy = Policy {
+            max_bytes: 4,
+            ..Policy::default()
+        };
         let result = fetch_url_with_resolver(&policy, &raw, fixture_resolver(port)).await;
         assert_eq!(result, Err("web_size_limit".into()));
         server.await.unwrap();
@@ -1021,7 +1129,13 @@
         for content_type in ["application/pdf", "image/png", "application/octet-stream"] {
             assert!(!is_supported_text_content_type(content_type));
         }
-        for content_type in ["text/html", "text/plain", "application/json", "application/xml", "text/xml"] {
+        for content_type in [
+            "text/html",
+            "text/plain",
+            "application/json",
+            "application/xml",
+            "text/xml",
+        ] {
             assert!(is_supported_text_content_type(content_type));
         }
     }
@@ -1038,7 +1152,8 @@
                 "untrusted_content": true,
             })],
             vec![],
-        ).unwrap();
+        )
+        .unwrap();
         let result: serde_json::Value = serde_json::from_slice(&encoded).unwrap();
         assert_eq!(result["untrusted_content"], true);
         assert_eq!(result["sources"][0]["untrusted_content"], true);
@@ -1047,13 +1162,20 @@
 
     #[tokio::test]
     async fn web_fetch_timeout_returns_bounded_error() {
-        let policy = Policy { total_timeout: Duration::from_millis(10), ..Policy::default() };
-        let result = fetch_with_controls(policy, "https://fixture.invalid/slow".into(),
+        let policy = Policy {
+            total_timeout: Duration::from_millis(10),
+            ..Policy::default()
+        };
+        let result = fetch_with_controls(
+            policy,
+            "https://fixture.invalid/slow".into(),
             watch::channel(false).1,
             |_policy, _raw| async {
                 tokio::time::sleep(Duration::from_secs(5)).await;
                 Err("unexpected_success".into())
-            }).await;
+            },
+        )
+        .await;
         assert_eq!(result, Err("web_timeout".into()));
     }
 
@@ -1061,11 +1183,16 @@
     async fn web_fetch_cancellation_returns_without_waiting_for_request() {
         let (cancel_tx, cancel_rx) = watch::channel(false);
         let task = tokio::spawn(async move {
-            fetch_with_controls(Policy::default(), "https://fixture.invalid/slow".into(), cancel_rx,
+            fetch_with_controls(
+                Policy::default(),
+                "https://fixture.invalid/slow".into(),
+                cancel_rx,
                 |_policy, _raw| async {
                     tokio::time::sleep(Duration::from_secs(30)).await;
                     Err("unexpected_success".into())
-                }).await
+                },
+            )
+            .await
         });
         cancel_tx.send(true).unwrap();
         assert_eq!(task.await.unwrap(), Err("cancelled".into()));
@@ -1100,7 +1227,10 @@
 
         prune_research_cache_dir(dir.path(), 100, 1024 * 1024).await;
 
-        assert!(!expired.exists(), "expired source excerpts should be deleted");
+        assert!(
+            !expired.exists(),
+            "expired source excerpts should be deleted"
+        );
         assert!(current.exists(), "unexpired cache entries should remain");
     }
 
@@ -1116,7 +1246,9 @@
             let result = research_cache_get_or_run(dir.path(), &key, || async {
                 fetches.fetch_add(1, Ordering::SeqCst);
                 Ok(expected.clone())
-            }).await.unwrap();
+            })
+            .await
+            .unwrap();
             assert_eq!(result, expected);
         }
 
@@ -1126,30 +1258,44 @@
     async fn research_cache_removes_abandoned_temp_files() {
         let dir = tempfile::tempdir().unwrap();
         let stale = dir.path().join(".entry.crashed.tmp");
-        tokio::fs::write(&stale, b"partial cached result").await.unwrap();
-        std::fs::File::open(&stale).unwrap().set_times(
-            std::fs::FileTimes::new().set_modified(SystemTime::UNIX_EPOCH),
-        ).unwrap();
+        tokio::fs::write(&stale, b"partial cached result")
+            .await
+            .unwrap();
+        std::fs::File::open(&stale)
+            .unwrap()
+            .set_times(std::fs::FileTimes::new().set_modified(SystemTime::UNIX_EPOCH))
+            .unwrap();
 
         prune_research_cache_dir(dir.path(), 1_000, 1024 * 1024).await;
 
-        assert!(!stale.exists(), "stale temp payloads should not escape quota/retention cleanup");
+        assert!(
+            !stale.exists(),
+            "stale temp payloads should not escape quota/retention cleanup"
+        );
     }
     #[tokio::test]
     async fn research_cache_rejects_and_deletes_payload_hash_mismatch() {
         let dir = tempfile::tempdir().unwrap();
         let path = dir.path().join(format!("{}.json", "c".repeat(64)));
         let original = serde_json::json!({"excerpt":"expected"});
-        let digest = format!("sha256:{:x}", Sha256::digest(serde_json::to_vec(&original).unwrap()));
+        let digest = format!(
+            "sha256:{:x}",
+            Sha256::digest(serde_json::to_vec(&original).unwrap())
+        );
         let changed = serde_json::json!({
             "expires_at": 200,
             "data_sha256": digest,
             "data": {"excerpt":"tampered"},
         });
-        tokio::fs::write(&path, serde_json::to_vec(&changed).unwrap()).await.unwrap();
+        tokio::fs::write(&path, serde_json::to_vec(&changed).unwrap())
+            .await
+            .unwrap();
 
         assert!(read_research_cache_entry(&path, 100).await.is_none());
-        assert!(!path.exists(), "a corrupt cache result should not be retained");
+        assert!(
+            !path.exists(),
+            "a corrupt cache result should not be retained"
+        );
     }
     #[tokio::test]
     async fn research_cache_hits_refresh_lru_timestamp() {
@@ -1162,11 +1308,14 @@
             "data_sha256": format!("sha256:{:x}", Sha256::digest(&encoded)),
             "data": data,
         });
-        tokio::fs::write(&path, serde_json::to_vec(&entry).unwrap()).await.unwrap();
+        tokio::fs::write(&path, serde_json::to_vec(&entry).unwrap())
+            .await
+            .unwrap();
         let old = SystemTime::UNIX_EPOCH + Duration::from_secs(1_000);
-        std::fs::File::open(&path).unwrap().set_times(
-            std::fs::FileTimes::new().set_modified(old),
-        ).unwrap();
+        std::fs::File::open(&path)
+            .unwrap()
+            .set_times(std::fs::FileTimes::new().set_modified(old))
+            .unwrap();
 
         assert_eq!(read_research_cache_entry(&path, 100).await, Some(encoded));
         assert!(std::fs::metadata(path).unwrap().modified().unwrap() > old);
--- /dev/null
+++ b/crates/airouter-tools/tests/mcp.rs
@@ -0,0 +1,282 @@
+#[path = "support/mcp_fixture.rs"]
+mod fixture;
+use airouter_tools::api::{self, Peer, Runtime};
+use axum::{
+    Extension,
+    body::{Body, to_bytes},
+    extract::ConnectInfo,
+    http::{Request, StatusCode},
+};
+use fixture::*;
+use serde_json::{Value, json};
+use std::{sync::Arc, time::Duration};
+use tower::ServiceExt;
+use uuid::Uuid;
+async fn call(
+    rt: Arc<Runtime>,
+    method: &str,
+    path: &str,
+    body: Value,
+    subject: &str,
+    context: &str,
+) -> (StatusCode, Value) {
+    let r = api::router(rt)
+        .layer(Extension(ConnectInfo(Peer(Some(
+            nix::unistd::geteuid().as_raw(),
+        )))))
+        .oneshot(
+            Request::builder()
+                .method(method)
+                .uri(path)
+                .header("authorization", format!("Bearer {TOKEN}"))
+                .header("content-type", "application/json")
+                .header("x-airouter-mcp-subject", subject)
+                .header("x-airouter-mcp-context", context)
+                .body(Body::from(body.to_string()))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    let status = r.status();
+    let bytes = to_bytes(r.into_body(), 100000).await.unwrap();
+    (
+        status,
+        serde_json::from_slice(&bytes).unwrap_or(json!(null)),
+    )
+}
+async fn done(rt: Arc<Runtime>, path: &str) -> Value {
+    tokio::time::timeout(Duration::from_secs(8), async {
+        loop {
+            let (s, v) = call(
+                rt.clone(),
+                "GET",
+                path,
+                json!(null),
+                "citadel:user:7",
+                &"a".repeat(64),
+            )
+            .await;
+            assert_eq!(s, 200, "{v}");
+            if v["complete"] == true {
+                break v;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap()
+}
+#[tokio::test]
+#[ignore = "requires WARDEN_TEST_ADAPTER (qualified real RMCP adapter)"]
+async fn real_adapter_recovery_revocation_isolation_cancellation_and_errors() {
+    let adapter = std::env::var("WARDEN_TEST_ADAPTER").expect("adapter path");
+    let dir = tempfile::tempdir().unwrap();
+    let f = Fixture::start(dir.path(), std::path::Path::new(&adapter)).await;
+    assert_eq!(f.root, dir.path());
+    let ctx = "a".repeat(64);
+    let path = format!("/v1/scopes/{WB}/mcp");
+    let (s, c) = call(
+        f.runtime.clone(),
+        "GET",
+        &format!("{path}/catalog"),
+        json!(null),
+        "citadel:user:7",
+        &ctx,
+    )
+    .await;
+    assert_eq!(s, 200, "{c}");
+    assert_eq!(c["servers"][0]["tools"].as_array().unwrap().len(), 3);
+    assert_eq!(
+        call(
+            f.runtime.clone(),
+            "GET",
+            &format!("{path}/catalog"),
+            json!(null),
+            "citadel:user:8",
+            &ctx
+        )
+        .await
+        .0,
+        403
+    );
+    let id = Uuid::new_v4();
+    let req = json!({"request_id":id,"server":"warden","name":"search_logs","arguments":{"source_id":"syslog-alpha-smsd"}});
+    let submit = format!("{path}/runs");
+    let lookup = format!("{submit}/{id}");
+    let (s, j) = call(
+        f.runtime.clone(),
+        "POST",
+        &submit,
+        req.clone(),
+        "citadel:user:7",
+        &ctx,
+    )
+    .await;
+    assert_eq!(s, 202, "{j}");
+    let result = done(f.runtime.clone(), &lookup).await;
+    assert_eq!(result["succeeded"], true, "{result}");
+    assert!(
+        result["result"]["structuredContent"]["evidence"][0]["text"]
+            .as_str()
+            .unwrap()
+            .contains('�')
+    );
+    let replay = call(
+        f.runtime.clone(),
+        "POST",
+        &submit,
+        req,
+        "citadel:user:7",
+        &ctx,
+    )
+    .await;
+    assert_eq!(replay.0, 200);
+    assert!(replay.1["result"].is_null());
+    assert_eq!(
+        std::fs::read_to_string(dir.path().join("calls")).unwrap(),
+        "1"
+    );
+    assert_eq!(
+        call(
+            f.runtime.clone(),
+            "GET",
+            &lookup,
+            json!(null),
+            "citadel:user:7",
+            &"b".repeat(64)
+        )
+        .await
+        .0,
+        403
+    );
+    let ledger: Value =
+        serde_json::from_slice(&std::fs::read(dir.path().join("ledger/ledger.json")).unwrap())
+            .unwrap();
+    for artifact in ledger["artifacts"].as_object().unwrap().keys() {
+        assert_eq!(
+            call(
+                f.runtime.clone(),
+                "GET",
+                &format!("/v1/scopes/{WB}/artifacts/{artifact}"),
+                json!(null),
+                "citadel:user:7",
+                &ctx
+            )
+            .await
+            .0,
+            403
+        );
+        let req = json!({"request_id":Uuid::new_v4(),"input":artifact,"operation":{"tool":"pdf_extract","options":{}},"timeout_seconds":20});
+        assert_eq!(
+            call(
+                f.runtime.clone(),
+                "POST",
+                &format!("/v1/scopes/{WB}/jobs"),
+                req,
+                "citadel:user:7",
+                &ctx
+            )
+            .await
+            .0,
+            403
+        );
+    }
+    assert_eq!(
+        call(
+            f.runtime.clone(),
+            "GET",
+            &format!("/v1/scopes/{WB}/jobs/{}", j["job_id"].as_str().unwrap()),
+            json!(null),
+            "citadel:user:7",
+            &ctx
+        )
+        .await
+        .0,
+        403
+    );
+    f.runtime.shutdown().await;
+    let restart = tempfile::tempdir().unwrap();
+    std::fs::create_dir(restart.path().join("ledger")).unwrap();
+    use std::os::unix::fs::PermissionsExt;
+    std::fs::set_permissions(restart.path(), std::fs::Permissions::from_mode(0o700)).unwrap();
+    std::fs::set_permissions(
+        restart.path().join("ledger"),
+        std::fs::Permissions::from_mode(0o700),
+    )
+    .unwrap();
+    private(
+        &restart.path().join("registry.json"),
+        &std::fs::read(dir.path().join("registry.json")).unwrap(),
+    );
+    for file in std::fs::read_dir(dir.path().join("ledger")).unwrap() {
+        let file = file.unwrap();
+        if file.file_name() != ".lock" {
+            std::fs::copy(
+                file.path(),
+                restart.path().join("ledger").join(file.file_name()),
+            )
+            .unwrap();
+        }
+    }
+    let recovered = Fixture::runtime(restart.path());
+    assert_eq!(done(recovered.clone(), &lookup).await["succeeded"], true);
+    assert_eq!(
+        std::fs::read_to_string(dir.path().join("calls")).unwrap(),
+        "1"
+    );
+    private(&dir.path().join("control.json"), br#"{"revision":2}"#);
+    assert_eq!(
+        call(
+            recovered.clone(),
+            "GET",
+            &lookup,
+            json!(null),
+            "citadel:user:7",
+            &ctx
+        )
+        .await
+        .0,
+        403
+    );
+    private(&dir.path().join("control.json"), br#"{"revoked":true}"#);
+    assert_eq!(call(recovered.clone(),"POST",&submit,json!({"request_id":Uuid::new_v4(),"server":"warden","name":"search_logs","arguments":{"source_id":"syslog-alpha-smsd"}}),"citadel:user:7",&ctx).await.0,403);
+    private(&dir.path().join("control.json"), b"{}");
+    for (text, host, error) in [
+        ("", "beta", "permission_denied"),
+        ("fixture_timeout", "alpha", "query_timeout"),
+        ("fixture_oversize", "alpha", "response_limit"),
+    ] {
+        let id = Uuid::new_v4();
+        let(s,j)=call(recovered.clone(),"POST",&submit,json!({"request_id":id,"server":"warden","name":"search_logs","arguments":{"source_id":"syslog-alpha-smsd","host":host,"text":text}}),"citadel:user:7",&ctx).await;
+        assert_eq!(s, 202, "{j}");
+        let v = done(recovered.clone(), &format!("{submit}/{id}")).await;
+        assert_eq!(v["succeeded"], false);
+        assert!(v.to_string().contains(error), "{v}");
+    }
+    let id = Uuid::new_v4();
+    let(s,j)=call(recovered.clone(),"POST",&submit,json!({"request_id":id,"server":"warden","name":"search_logs","arguments":{"source_id":"syslog-alpha-smsd","text":"fixture_wait"}}),"citadel:user:7",&ctx).await;
+    assert_eq!(s, 202, "{j}");
+    let before = std::fs::read_to_string(dir.path().join("calls")).unwrap();
+    tokio::time::timeout(Duration::from_secs(5), async {
+        while std::fs::read_to_string(dir.path().join("calls")).unwrap() == before {
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap();
+    let (s, _) = call(
+        recovered.clone(),
+        "POST",
+        &format!("{submit}/{id}/cancel"),
+        json!({}),
+        "citadel:user:7",
+        &ctx,
+    )
+    .await;
+    assert_eq!(s, 200);
+    assert_eq!(
+        done(recovered.clone(), &format!("{submit}/{id}")).await["phase"],
+        "cancelled"
+    );
+    recovered.shutdown().await;
+}
--- a/crates/airouter-tools/tests/store.rs
+++ b/crates/airouter-tools/tests/store.rs
@@ -252,3 +252,48 @@
     assert!(store.job("a", scope, j.id).is_err());
     assert!(!store.blob_path(a.id).exists());
 }
+
+#[test]
+fn diagnostic_outputs_expire_with_their_policy_even_when_the_output_is_newer() {
+    let dir = private_dir();
+    let mut store = Store::open(dir.path()).unwrap();
+    let scope = Uuid::new_v4();
+    let a = store.upload("a", scope, b"{}").unwrap();
+    let mut r = request(a.id);
+    r.operation = Operation::McpRead {
+        server: "warden".into(),
+        name: "search_logs".into(),
+        arguments: serde_json::json!({}),
+        authority: airouter_tools::mcp::Authority {
+            subject: "fixture".into(),
+            context: "a".repeat(64),
+            registration: "b".repeat(64),
+            data_revision: "c".repeat(64),
+        },
+    };
+    let j = store.admit("a", scope, r).unwrap();
+    let done = store
+        .complete(
+            j.id,
+            None,
+            None,
+            vec![(
+                "evidence.json".into(),
+                "application/json".into(),
+                b"private evidence".to_vec(),
+            )],
+        )
+        .unwrap();
+    let output = done.artifacts[0].id;
+    assert!(store.read_artifact("a", scope, output).is_err());
+    drop(store);
+    let path = dir.path().join("ledger.json");
+    let mut ledger: serde_json::Value =
+        serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
+    ledger["jobs"][j.id.to_string()]["created_at"] = serde_json::json!(0);
+    std::fs::write(&path, serde_json::to_vec(&ledger).unwrap()).unwrap();
+    let store = Store::open(dir.path()).unwrap();
+    assert!(store.artifact("a", scope, output).is_err());
+    assert!(store.artifact("a", scope, a.id).is_err());
+    assert!(!store.blob_path(output).exists());
+}
--- /dev/null
+++ b/crates/airouter-tools/tests/support/mcp_fixture.rs
@@ -0,0 +1,225 @@
+//! Model-free cross-application fixture: real toolsd routes, ledger, AIRouter MCP
+//! connector and Warden adapter; only the upstream Warden HTTP data is scripted.
+use airouter_tools::{
+    api::{self, Application, Peer, Runtime},
+    catalog::Models,
+    mcp::{Binding, Broker, Registry},
+    protocol::Operation,
+    store::{Store, digest, now},
+    worker::{Execution, Executor},
+};
+use async_trait::async_trait;
+use axum::{
+    Json, Router,
+    extract::{Query, State},
+    http::{HeaderMap, StatusCode},
+    response::IntoResponse,
+    routing::get,
+};
+use serde_json::{Value, json};
+use std::{
+    collections::HashMap,
+    os::unix::fs::PermissionsExt,
+    path::{Path, PathBuf},
+    sync::{
+        Arc,
+        atomic::{AtomicUsize, Ordering},
+    },
+};
+use tokio::{net::UnixListener, sync::watch, task::JoinHandle};
+use uuid::Uuid;
+pub const TOKEN: &str = "fixture-only-frontend-token-at-least-32-bytes";
+pub const SMITH_TOKEN: &str = "fixture-only-smith-frontend-token-at-least-32-bytes";
+pub const DATA_TOKEN: &str = "fixture-only-warden-token-at-least-32-bytes";
+pub const WB: &str = "9d03b45c-495f-4d35-b3fa-f938102174a8";
+pub const SMITH: &str = "580daaa0-1100-4e07-8c7a-2732329aaf91";
+pub struct NoWorker;
+#[async_trait]
+impl Executor for NoWorker {
+    fn unavailable_reason(&self, _: &Models, _: &str) -> Option<String> {
+        None
+    }
+    async fn execute(
+        &self,
+        _: PathBuf,
+        _: Operation,
+        _: u64,
+        _: watch::Receiver<bool>,
+    ) -> Execution {
+        panic!("diagnostic reached utility worker")
+    }
+}
+#[derive(Clone)]
+struct Data {
+    root: PathBuf,
+    calls: Arc<AtomicUsize>,
+}
+fn control(data: &Data) -> Value {
+    serde_json::from_slice(&std::fs::read(data.root.join("control.json")).unwrap()).unwrap()
+}
+async fn authority(State(data): State<Data>, headers: HeaderMap) -> axum::response::Response {
+    if headers.get("authorization").and_then(|v| v.to_str().ok())
+        != Some(&format!("Bearer {DATA_TOKEN}"))
+        || control(&data)["revoked"] == true
+    {
+        return StatusCode::FORBIDDEN.into_response();
+    }
+    let revision = control(&data)["revision"].as_u64().unwrap_or(1);
+    (
+        [("x-warden-authority", format!("{revision:064x}"))],
+        Json(json!({"ok":true})),
+    )
+        .into_response()
+}
+async fn logs(
+    State(data): State<Data>,
+    Query(args): Query<HashMap<String, String>>,
+    headers: HeaderMap,
+) -> axum::response::Response {
+    if !authority(State(data.clone()), headers)
+        .await
+        .status()
+        .is_success()
+    {
+        return StatusCode::FORBIDDEN.into_response();
+    }
+    if args.get("source_id").map(String::as_str) != Some("syslog-alpha-smsd")
+        || args.get("host").is_some_and(|v| v != "alpha")
+        || args.get("service").is_some_and(|v| v != "smsd")
+    {
+        return StatusCode::FORBIDDEN.into_response();
+    }
+    let count = data.calls.fetch_add(1, Ordering::SeqCst) + 1;
+    std::fs::write(data.root.join("calls"), count.to_string()).unwrap();
+    if args.get("text").is_some_and(|v| v == "fixture_wait") {
+        std::future::pending::<()>().await;
+    }
+    if args.get("text").is_some_and(|v| v == "fixture_timeout") {
+        return StatusCode::GATEWAY_TIMEOUT.into_response();
+    }
+    if args.get("text").is_some_and(|v| v == "fixture_oversize") {
+        return "x".repeat(17000).into_response();
+    }
+    Json(json!({"schema":"warden.diagnostics.v1","query_id":"fixture","complete":true,"sources":[{"source_id":"syslog-alpha-smsd","host":"alpha","service":"smsd"}],"evidence":[{"id":"syslog:3","text":"fixture|line\nnext\t� token=<redacted>","untrusted":true}],"pagination":{"has_more":false}})).into_response()
+}
+async fn sources(State(data): State<Data>, headers: HeaderMap) -> axum::response::Response {
+    if !authority(State(data), headers).await.status().is_success() {
+        return StatusCode::FORBIDDEN.into_response();
+    }
+    Json(json!({"schema":"warden.diagnostics.v1","complete":true,"evidence":[],"sources":[{"source_id":"syslog-alpha-smsd","host":"alpha","service":"smsd"}]})).into_response()
+}
+pub struct Fixture {
+    pub root: PathBuf,
+    pub runtime: Arc<Runtime>,
+    tasks: Vec<JoinHandle<()>>,
+}
+impl Drop for Fixture {
+    fn drop(&mut self) {
+        for task in &self.tasks {
+            task.abort();
+        }
+    }
+}
+impl Fixture {
+    pub async fn start(root: &Path, adapter: &Path) -> Self {
+        std::fs::create_dir_all(root).unwrap();
+        std::fs::set_permissions(root, std::fs::Permissions::from_mode(0o700)).unwrap();
+        std::fs::create_dir(root.join("ledger")).unwrap();
+        std::fs::set_permissions(root.join("ledger"), std::fs::Permissions::from_mode(0o700))
+            .unwrap();
+        let command = root.join("warden-mcp");
+        std::fs::copy(adapter, &command).unwrap();
+        std::fs::set_permissions(&command, std::fs::Permissions::from_mode(0o700)).unwrap();
+        private(&root.join("data.token"), DATA_TOKEN.as_bytes());
+        private(&root.join("frontend.token"), TOKEN.as_bytes());
+        private(&root.join("smith.token"), SMITH_TOKEN.as_bytes());
+        private(&root.join("control.json"), b"{}");
+        let socket = root.join("warden.sock");
+        let listener = UnixListener::bind(&socket).unwrap();
+        let data = Data {
+            root: root.into(),
+            calls: Arc::new(AtomicUsize::new(0)),
+        };
+        let router = Router::new()
+            .route("/api/v1/diagnostics/authority", get(authority))
+            .route("/api/v1/diagnostics/logs", get(logs))
+            .route("/api/v1/diagnostics/sources", get(sources))
+            .with_state(data);
+        let backend = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() });
+        let registry = Registry {
+            version: 1,
+            bindings: [
+                ("workbench", WB, "citadel:user:7"),
+                ("smith", SMITH, "smith:operator"),
+            ]
+            .into_iter()
+            .map(|(application, scope, subject)| Binding {
+                application: application.into(),
+                scope: scope.parse().unwrap(),
+                subject: subject.into(),
+                server: "warden".into(),
+                enabled: true,
+                revision: 1,
+                expires_at: now() + 3600,
+                command: command.clone(),
+                command_sha256: digest(&std::fs::read(&command).unwrap()),
+                socket: socket.clone(),
+                credential_file: root.join("data.token"),
+                authority_path: "/api/v1/diagnostics/authority".into(),
+                revision_header: "x-warden-authority".into(),
+                tools: vec![
+                    "list_log_sources".into(),
+                    "search_logs".into(),
+                    "get_diagnostic_context".into(),
+                ],
+            })
+            .collect(),
+        };
+        private(
+            &root.join("registry.json"),
+            &serde_json::to_vec(&registry).unwrap(),
+        );
+        let runtime = Self::runtime(root);
+        let listener = UnixListener::bind(root.join("tools.sock")).unwrap();
+        let app = api::router(runtime.clone());
+        let server = tokio::spawn(async move {
+            axum::serve(listener, app.into_make_service_with_connect_info::<Peer>())
+                .await
+                .unwrap()
+        });
+        Self {
+            root: root.into(),
+            runtime,
+            tasks: vec![backend, server],
+        }
+    }
+    pub fn runtime(root: &Path) -> Arc<Runtime> {
+        let uid = nix::unistd::geteuid().as_raw();
+        let apps = [("workbench", WB, TOKEN), ("smith", SMITH, SMITH_TOKEN)]
+            .into_iter()
+            .map(|(name, id, token)| {
+                Application::new(
+                    name.into(),
+                    uid,
+                    token,
+                    vec![id.parse::<Uuid>().unwrap()],
+                    vec!["mcp_read".into(), "pdf_extract".into()],
+                )
+                .unwrap()
+            })
+            .collect();
+        Arc::new(
+            Runtime::new(
+                Store::open(&root.join("ledger")).unwrap(),
+                apps,
+                Models::default(),
+                Arc::new(NoWorker),
+            )
+            .with_mcp(Arc::new(Broker::new(root.join("registry.json")))),
+        )
+    }
+}
+pub fn private(path: &Path, bytes: &[u8]) {
+    std::fs::write(path, bytes).unwrap();
+    std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600)).unwrap();
+}
--- a/crates/airouter-mcp/src/lib.rs
+++ b/crates/airouter-mcp/src/lib.rs
@@ -1127,7 +1127,7 @@
         );
     }
 
-    #[cfg(feature = "http")]
+    #[cfg(any(feature = "http", feature = "stdio"))]
     fn registration(server_id: &str) -> EndpointRegistration {
         EndpointRegistration {
             server_id: server_id.into(),