proofborne 0.1.0-alpha.2

use std::{
    collections::{BTreeMap, BTreeSet, HashMap},
    fs,
    path::{Path, PathBuf},
    process::ExitCode,
    sync::{Arc, Mutex},
};

use anyhow::{Context, Result, anyhow, bail};
use async_trait::async_trait;
use proofborne_adapters::{
    McpConnectionReport, McpServerAttachment, McpTransportMaterial, register_mcp_attachments,
};
use proofborne_core::{
    ActionClass, CriterionState, EventEnvelope, RunOutcome, SCHEMA_VERSION, TaskContract, hash_json,
};
use proofborne_runtime::{
    AppConfig, ApprovalHandler, EventSink, PolicyPreset, PolicyRequest, RunControl, RunOptions,
    RunResult, RuntimePaths, SessionStore, ToolRegistry,
};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use tokio::{
    io::{AsyncBufRead, AsyncBufReadExt, AsyncWriteExt, BufReader, BufWriter},
    sync::{mpsc, oneshot},
};
use uuid::Uuid;

use super::{prepare_engine_with_tools, tui::SurfaceSnapshot};

const MAX_RPC_LINE: usize = 4 * 1024 * 1024;
const MAX_MCP_SERVERS: usize = 32;
const MAX_MCP_FIELDS: usize = 64;
const MAX_MCP_ARGS: usize = 128;
const MAX_MCP_VALUE: usize = 64 * 1024;

struct ActivePrompt {
    control: RunControl,
    task: Option<tokio::task::JoinHandle<()>>,
}

struct AcpSession {
    metadata: PersistedAcpSession,
    active: Option<ActivePrompt>,
    mcp: AcpMcpRuntime,
}

type Sessions = Arc<Mutex<HashMap<String, AcpSession>>>;
type PendingClientRequests = Arc<Mutex<HashMap<String, PendingClientRequest>>>;

struct PendingClientRequest {
    session_id: String,
    response: oneshot::Sender<Value>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
struct PersistedAcpSession {
    schema_version: String,
    session_id: String,
    workspace: PathBuf,
    runtime_session_ids: Vec<Uuid>,
    #[serde(default)]
    mcp_servers: Vec<PersistedMcpServer>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    tool_authority_hash: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct PersistedMcpServer {
    name: String,
    transport: String,
    endpoint: String,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    args: Vec<String>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    credential_fields: Vec<String>,
}

#[derive(Clone)]
struct AcpMcpServer {
    descriptor: PersistedMcpServer,
    transport: AcpMcpTransport,
}

#[derive(Clone)]
enum AcpMcpTransport {
    Stdio {
        command: String,
        args: Vec<String>,
        env: Vec<(String, String)>,
    },
    Http {
        url: String,
        headers: Vec<(String, String)>,
    },
}

struct AcpMcpRuntime {
    tools: ToolRegistry,
    secrets: Vec<String>,
    reports: Vec<McpConnectionReport>,
}

impl AcpMcpRuntime {
    fn empty() -> Self {
        Self {
            tools: ToolRegistry::default(),
            secrets: Vec::new(),
            reports: Vec::new(),
        }
    }
}

#[derive(Deserialize)]
#[serde(untagged)]
enum AcpMcpServerWire {
    Stdio(AcpMcpStdioWire),
    Http(AcpMcpHttpWire),
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct AcpMcpStdioWire {
    name: String,
    command: PathBuf,
    #[serde(default)]
    args: Vec<String>,
    #[serde(default)]
    env: Vec<AcpNameValue>,
    #[serde(default, rename = "_meta")]
    _meta: Option<Value>,
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct AcpMcpHttpWire {
    #[serde(rename = "type")]
    transport: String,
    name: String,
    url: String,
    #[serde(default)]
    headers: Vec<AcpNameValue>,
    #[serde(default, rename = "_meta")]
    _meta: Option<Value>,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct AcpNameValue {
    name: String,
    value: String,
}

#[derive(Debug, Clone)]
struct AcpSessionRegistry {
    directory: PathBuf,
}

struct FinalizedRun {
    result: RunResult,
    events: Vec<EventEnvelope>,
}

impl AcpSessionRegistry {
    fn open(default_workspace: &Path) -> Result<Self> {
        let paths = RuntimePaths::for_workspace(default_workspace)?;
        let directory = paths.user_data.join("acp").join("sessions");
        fs::create_dir_all(&directory)
            .with_context(|| format!("failed to create {}", directory.display()))?;
        Ok(Self { directory })
    }

    #[cfg(test)]
    fn at(directory: PathBuf) -> Result<Self> {
        fs::create_dir_all(&directory)?;
        Ok(Self { directory })
    }

    fn save(&self, metadata: &PersistedAcpSession) -> Result<()> {
        validate_acp_session_id(&metadata.session_id)?;
        let path = self.path(&metadata.session_id)?;
        let bytes = serde_json::to_vec_pretty(metadata)?;
        fs::write(&path, bytes).with_context(|| {
            format!(
                "failed to persist ACP session metadata at {}",
                path.display()
            )
        })
    }

    fn load(&self, session_id: &str) -> Result<PersistedAcpSession> {
        let path = self.path(session_id)?;
        let bytes = fs::read(&path)
            .with_context(|| format!("unknown persisted ACP session {session_id}"))?;
        let metadata: PersistedAcpSession = serde_json::from_slice(&bytes)
            .with_context(|| format!("invalid ACP session metadata at {}", path.display()))?;
        if metadata.schema_version != SCHEMA_VERSION {
            bail!(
                "ACP session {} uses unsupported schema version {}",
                session_id,
                metadata.schema_version
            );
        }
        if metadata.session_id != session_id {
            bail!("ACP session metadata identity does not match its file name");
        }
        Ok(metadata)
    }

    fn path(&self, session_id: &str) -> Result<PathBuf> {
        validate_acp_session_id(session_id)?;
        Ok(self.directory.join(format!("{session_id}.json")))
    }
}

/// Serves stable ACP v1 over newline-delimited JSON-RPC on stdio.
pub async fn serve(
    default_workspace: PathBuf,
    profile: Option<String>,
    policy: Option<PolicyPreset>,
    allow_client_mcp: bool,
) -> Result<ExitCode> {
    let registry = AcpSessionRegistry::open(&default_workspace)?;
    let (output_tx, mut output_rx) = mpsc::unbounded_channel::<Value>();
    let writer = tokio::spawn(async move {
        let mut stdout = BufWriter::new(tokio::io::stdout());
        while let Some(message) = output_rx.recv().await {
            let bytes = serde_json::to_vec(&message).context("failed to encode ACP message")?;
            stdout.write_all(&bytes).await?;
            stdout.write_all(b"\n").await?;
            stdout.flush().await?;
        }
        Ok::<_, anyhow::Error>(())
    });

    let sessions: Sessions = Arc::new(Mutex::new(HashMap::new()));
    let pending_client_requests: PendingClientRequests = Arc::new(Mutex::new(HashMap::new()));
    let mut initialized = false;
    let mut stdin = BufReader::new(tokio::io::stdin());
    loop {
        let line = match read_bounded_rpc_line(&mut stdin, MAX_RPC_LINE).await? {
            RpcLine::Eof => break,
            RpcLine::Oversized => {
                output_tx
                    .send(error_response(
                        Value::Null,
                        -32600,
                        "ACP message exceeds 4 MiB",
                    ))
                    .map_err(|_| anyhow!("ACP stdout writer stopped"))?;
                continue;
            }
            RpcLine::Message(line) => line,
        };
        if line.len() > MAX_RPC_LINE {
            output_tx
                .send(error_response(
                    Value::Null,
                    -32600,
                    "ACP message exceeds 4 MiB",
                ))
                .map_err(|_| anyhow!("ACP stdout writer stopped"))?;
            continue;
        }
        let message: Value = match serde_json::from_slice(&line) {
            Ok(message) => message,
            Err(error) => {
                output_tx
                    .send(error_response(
                        Value::Null,
                        -32700,
                        &format!("parse error: {error}"),
                    ))
                    .map_err(|_| anyhow!("ACP stdout writer stopped"))?;
                continue;
            }
        };
        if route_client_response(&message, &pending_client_requests)? {
            continue;
        }
        let method = message
            .get("method")
            .and_then(Value::as_str)
            .unwrap_or_default();
        let id = message.get("id").cloned();
        let params = message.get("params").cloned().unwrap_or_else(|| json!({}));
        match method {
            "initialize" => {
                let Some(id) = id else {
                    send_error(
                        &output_tx,
                        Value::Null,
                        -32600,
                        "initialize must be a request",
                    )?;
                    continue;
                };
                initialized = true;
                send_result(&output_tx, id, initialize_result(allow_client_mcp))?;
            }
            _ if !initialized => {
                if let Some(id) = id {
                    send_error(&output_tx, id, -32002, "connection is not initialized")?;
                }
            }
            "session/new" => {
                let Some(id) = id else {
                    continue;
                };
                if let Err(error) = validate_session_setup(&params, allow_client_mcp) {
                    send_error(&output_tx, id, -32602, &error.to_string())?;
                    continue;
                }
                let mcp_servers = match parse_acp_mcp_servers(&params) {
                    Ok(servers) => servers,
                    Err(error) => {
                        send_error(&output_tx, id, -32602, &error.to_string())?;
                        continue;
                    }
                };
                let mcp_descriptors = mcp_descriptors(&mcp_servers);
                let workspace = match acp_workspace(&params, &default_workspace) {
                    Ok(workspace) => workspace,
                    Err(error) => {
                        send_error(&output_tx, id, -32602, &error.to_string())?;
                        continue;
                    }
                };
                if let Err(error) = validate_acp_mcp_policy(&workspace, &mcp_servers) {
                    send_error(&output_tx, id, -32003, &error.to_string())?;
                    continue;
                }
                let mcp = match connect_acp_mcp(&mcp_servers).await {
                    Ok(runtime) => runtime,
                    Err(error) => {
                        send_error(
                            &output_tx,
                            id,
                            -32000,
                            &redacted_mcp_setup_error(&mcp_servers, &error),
                        )?;
                        continue;
                    }
                };
                let mcp_reports = mcp.reports.clone();
                let tool_authority_hash = mcp_authority_hash(&mcp_descriptors, &mcp);
                let session_id = format!("sess_{}", Uuid::now_v7().simple());
                let metadata = PersistedAcpSession {
                    schema_version: SCHEMA_VERSION.to_owned(),
                    session_id: session_id.clone(),
                    workspace,
                    runtime_session_ids: Vec::new(),
                    mcp_servers: mcp_descriptors,
                    tool_authority_hash,
                };
                registry.save(&metadata)?;
                sessions
                    .lock()
                    .map_err(|_| anyhow!("ACP session lock poisoned"))?
                    .insert(
                        session_id.clone(),
                        AcpSession {
                            metadata,
                            active: None,
                            mcp,
                        },
                    );
                send_result(
                    &output_tx,
                    id,
                    json!({
                        "sessionId": session_id,
                        "_meta": {"proofborne": {"mcpConnections": mcp_reports}}
                    }),
                )?;
            }
            "session/load" => {
                let Some(id) = id else {
                    continue;
                };
                if let Err(error) = validate_session_setup(&params, allow_client_mcp) {
                    send_error(&output_tx, id, -32602, &error.to_string())?;
                    continue;
                }
                let mcp_servers = match parse_acp_mcp_servers(&params) {
                    Ok(servers) => servers,
                    Err(error) => {
                        send_error(&output_tx, id, -32602, &error.to_string())?;
                        continue;
                    }
                };
                let mcp_descriptors = mcp_descriptors(&mcp_servers);
                let Some(session_id) = params.get("sessionId").and_then(Value::as_str) else {
                    send_error(&output_tx, id, -32602, "sessionId is required")?;
                    continue;
                };
                let active = sessions
                    .lock()
                    .map_err(|_| anyhow!("ACP session lock poisoned"))?
                    .get(session_id)
                    .is_some_and(|session| session.active.is_some());
                if active {
                    send_error(
                        &output_tx,
                        id,
                        -32003,
                        "session already has an active prompt",
                    )?;
                    continue;
                }
                let requested_workspace = match acp_workspace(&params, &default_workspace) {
                    Ok(workspace) => workspace,
                    Err(error) => {
                        send_error(&output_tx, id, -32602, &error.to_string())?;
                        continue;
                    }
                };
                let mut metadata = match registry.load(session_id) {
                    Ok(metadata) => metadata,
                    Err(error) => {
                        send_error(&output_tx, id, -32001, &error.to_string())?;
                        continue;
                    }
                };
                if !same_path(&metadata.workspace, &requested_workspace) {
                    send_error(
                        &output_tx,
                        id,
                        -32602,
                        "session cwd differs from its persisted canonical workspace",
                    )?;
                    continue;
                }
                if let Err(error) = validate_acp_mcp_policy(&requested_workspace, &mcp_servers) {
                    send_error(&output_tx, id, -32003, &error.to_string())?;
                    continue;
                }
                let mcp = match connect_acp_mcp(&mcp_servers).await {
                    Ok(runtime) => runtime,
                    Err(error) => {
                        send_error(
                            &output_tx,
                            id,
                            -32000,
                            &redacted_mcp_setup_error(&mcp_servers, &error),
                        )?;
                        continue;
                    }
                };
                let mcp_reports = mcp.reports.clone();
                let tool_authority_hash = mcp_authority_hash(&mcp_descriptors, &mcp);
                if let Err(error) = recoverable_runtime_session(&metadata) {
                    send_error(&output_tx, id, -32001, &error.to_string())?;
                    continue;
                }
                metadata.mcp_servers = mcp_descriptors;
                metadata.tool_authority_hash = tool_authority_hash;
                registry.save(&metadata)?;
                let runs = match load_finalized_runs(&metadata) {
                    Ok(runs) => runs,
                    Err(error) => {
                        send_error(&output_tx, id, -32001, &error.to_string())?;
                        continue;
                    }
                };
                sessions
                    .lock()
                    .map_err(|_| anyhow!("ACP session lock poisoned"))?
                    .insert(
                        session_id.to_owned(),
                        AcpSession {
                            metadata,
                            active: None,
                            mcp,
                        },
                    );
                replay_runs(session_id, &runs, &output_tx)?;
                send_result(
                    &output_tx,
                    id,
                    json!({"_meta": {"proofborne": {"mcpConnections": mcp_reports}}}),
                )?;
            }
            "session/prompt" => {
                let Some(id) = id else {
                    continue;
                };
                let Some(session_id) = params
                    .get("sessionId")
                    .and_then(Value::as_str)
                    .map(ToOwned::to_owned)
                else {
                    send_error(&output_tx, id, -32602, "sessionId is required")?;
                    continue;
                };
                let contract = match prompt_contract(&params) {
                    Ok(contract) => contract,
                    Err(error) => {
                        send_error(&output_tx, id, -32602, &error.to_string())?;
                        continue;
                    }
                };
                let (
                    workspace,
                    resume_session_id,
                    attached_tools,
                    attached_secrets,
                    tool_authority_hash,
                    mcp_reports,
                ) = {
                    let mut sessions_guard = sessions
                        .lock()
                        .map_err(|_| anyhow!("ACP session lock poisoned"))?;
                    let Some(session) = sessions_guard.get_mut(&session_id) else {
                        send_error(&output_tx, id, -32001, "unknown ACP session")?;
                        continue;
                    };
                    if session.active.is_some() {
                        send_error(
                            &output_tx,
                            id,
                            -32003,
                            "session already has an active prompt",
                        )?;
                        continue;
                    }
                    let resume_session_id = match recoverable_runtime_session(&session.metadata) {
                        Ok(id) => id,
                        Err(error) => {
                            send_error(&output_tx, id, -32001, &error.to_string())?;
                            continue;
                        }
                    };
                    (
                        session.metadata.workspace.clone(),
                        resume_session_id,
                        session.mcp.tools.clone(),
                        session.mcp.secrets.clone(),
                        session.metadata.tool_authority_hash.clone(),
                        session.mcp.reports.clone(),
                    )
                };

                let (start_tx, start_rx) = oneshot::channel();
                let task_sessions = sessions.clone();
                let task_output = output_tx.clone();
                let task_session_id = session_id.clone();
                let task_request_id = id.clone();
                let task_profile = profile.clone();
                let task_pending = pending_client_requests.clone();
                let task_registry = registry.clone();
                let control = RunControl::default();
                let task_control = control.clone();
                let task = tokio::spawn(async move {
                    let _ = start_rx.await;
                    let response = run_prompt(
                        workspace,
                        task_session_id.clone(),
                        contract,
                        task_profile,
                        policy,
                        task_output.clone(),
                        task_pending,
                        task_control,
                        resume_session_id,
                        attached_tools,
                        attached_secrets,
                        tool_authority_hash,
                        mcp_reports,
                    )
                    .await;
                    match response {
                        Ok((result, root_hash, surface)) => {
                            let persistence = persist_runtime_session(
                                &task_sessions,
                                &task_registry,
                                &task_session_id,
                                result.session_id,
                            );
                            match persistence {
                                Ok(()) => {
                                    let _ = task_output.send(json!({
                                        "jsonrpc": "2.0",
                                        "id": task_request_id,
                                        "result": {
                                            "stopReason": if result.outcome == RunOutcome::Cancelled { "cancelled" } else { "end_turn" },
                                            "_meta": {
                                                "proofborne": {
                                                    "runtimeSessionId": result.session_id,
                                                    "eventRootHash": root_hash,
                                                    "proofFinalized": true,
                                                    "surface": surface,
                                                }
                                            }
                                        }
                                    }));
                                }
                                Err(error) => {
                                    let _ = task_output.send(error_response(
                                        task_request_id,
                                        -32000,
                                        &format!(
                                            "Proofborne session persistence failed: {error:#}"
                                        ),
                                    ));
                                }
                            }
                        }
                        Err(error) => {
                            if let Ok(Some(runtime_session_id)) =
                                sink_runtime_session_id_from_error(&error)
                            {
                                let _ = persist_runtime_session(
                                    &task_sessions,
                                    &task_registry,
                                    &task_session_id,
                                    runtime_session_id,
                                );
                            }
                            let _ = task_output.send(error_response(
                                task_request_id,
                                -32000,
                                &format!("Proofborne run failed: {error:#}"),
                            ));
                        }
                    }
                    if let Ok(mut guard) = task_sessions.lock()
                        && let Some(session) = guard.get_mut(&task_session_id)
                    {
                        session.active = None;
                    }
                });
                {
                    let mut guard = sessions
                        .lock()
                        .map_err(|_| anyhow!("ACP session lock poisoned"))?;
                    if let Some(session) = guard.get_mut(&session_id) {
                        session.active = Some(ActivePrompt {
                            control,
                            task: Some(task),
                        });
                    }
                }
                let _ = start_tx.send(());
            }
            "session/cancel" => {
                if let Some(session_id) = params.get("sessionId").and_then(Value::as_str) {
                    cancel_prompt(&sessions, session_id, &output_tx, &pending_client_requests)?;
                }
            }
            "session/close" => {
                let Some(id) = id else {
                    continue;
                };
                let Some(session_id) = params.get("sessionId").and_then(Value::as_str) else {
                    send_error(&output_tx, id, -32602, "sessionId is required")?;
                    continue;
                };
                let active = {
                    let mut guard = sessions
                        .lock()
                        .map_err(|_| anyhow!("ACP session lock poisoned"))?;
                    let Some(session) = guard.get_mut(session_id) else {
                        send_error(&output_tx, id, -32001, "unknown active ACP session")?;
                        continue;
                    };
                    session.active.take()
                };
                if let Some(active) = active {
                    active.control.cancel();
                    cancel_pending_permissions(&pending_client_requests, session_id)?;
                    if let Err(error) = await_prompt_termination(active).await {
                        sessions
                            .lock()
                            .map_err(|_| anyhow!("ACP session lock poisoned"))?
                            .remove(session_id);
                        send_error(&output_tx, id, -32000, &error.to_string())?;
                        continue;
                    }
                }
                let removed = sessions
                    .lock()
                    .map_err(|_| anyhow!("ACP session lock poisoned"))?
                    .remove(session_id)
                    .is_some();
                if !removed {
                    send_error(&output_tx, id, -32001, "unknown active ACP session")?;
                    continue;
                }
                send_result(&output_tx, id, json!({}))?;
            }
            "" => {
                if let Some(id) = id {
                    send_error(&output_tx, id, -32600, "missing method")?;
                }
            }
            _ => {
                if let Some(id) = id {
                    send_error(&output_tx, id, -32601, "method not found")?;
                }
            }
        }
    }

    let active_prompts = {
        let mut guard = sessions
            .lock()
            .map_err(|_| anyhow!("ACP session lock poisoned"))?;
        guard
            .iter_mut()
            .filter_map(|(session_id, session)| {
                session
                    .active
                    .take()
                    .map(|active| (session_id.clone(), active))
            })
            .collect::<Vec<_>>()
    };
    for (session_id, active) in active_prompts {
        active.control.cancel();
        cancel_pending_permissions(&pending_client_requests, &session_id)?;
        let _ = await_prompt_termination(active).await;
    }
    drop(output_tx);
    writer.await.context("ACP writer task failed")??;
    Ok(ExitCode::SUCCESS)
}

enum RpcLine {
    Eof,
    Message(Vec<u8>),
    Oversized,
}

async fn read_bounded_rpc_line<R>(reader: &mut R, limit: usize) -> std::io::Result<RpcLine>
where
    R: AsyncBufRead + Unpin,
{
    let mut line = Vec::new();
    let mut oversized = false;
    loop {
        let available = reader.fill_buf().await?;
        if available.is_empty() {
            if line.is_empty() && !oversized {
                return Ok(RpcLine::Eof);
            }
            return Ok(if oversized {
                RpcLine::Oversized
            } else {
                RpcLine::Message(line)
            });
        }
        let newline = available.iter().position(|byte| *byte == b'\n');
        let consumed = newline.map_or(available.len(), |position| position + 1);
        if !oversized && line.len().saturating_add(consumed) <= limit {
            line.extend_from_slice(&available[..consumed]);
        } else {
            oversized = true;
            line.clear();
        }
        reader.consume(consumed);
        if newline.is_some() {
            return Ok(if oversized {
                RpcLine::Oversized
            } else {
                RpcLine::Message(line)
            });
        }
    }
}

fn initialize_result(allow_client_mcp: bool) -> Value {
    json!({
        "protocolVersion": 1,
        "agentCapabilities": {
            "loadSession": true,
            "promptCapabilities": {
                "image": false,
                "audio": false,
                "embeddedContext": false
            },
            "mcpCapabilities": {
                "http": allow_client_mcp,
                "sse": false
            },
            "sessionCapabilities": {
                "close": {}
            }
        },
        "agentInfo": {
            "name": "proofborne",
            "title": "Proofborne",
            "version": env!("CARGO_PKG_VERSION")
        },
        "authMethods": []
    })
}

async fn run_prompt(
    workspace: PathBuf,
    acp_session_id: String,
    contract: TaskContract,
    profile: Option<String>,
    policy: Option<PolicyPreset>,
    output: mpsc::UnboundedSender<Value>,
    pending: PendingClientRequests,
    control: RunControl,
    resume_session_id: Option<Uuid>,
    attached_tools: ToolRegistry,
    attached_secrets: Vec<String>,
    tool_authority_hash: Option<String>,
    mcp_reports: Vec<McpConnectionReport>,
) -> Result<(RunResult, String, SurfaceSnapshot)> {
    let prepared = if let Some(runtime_session_id) = resume_session_id {
        prepare_engine_for_resume(
            &workspace,
            runtime_session_id,
            policy.unwrap_or(PolicyPreset::Review),
            &attached_tools,
            &attached_secrets,
        )
        .await?
    } else {
        prepare_engine_with_tools(
            &workspace,
            profile.as_deref(),
            Some(policy.unwrap_or(PolicyPreset::Review)),
            Some(&attached_tools),
            &attached_secrets,
        )
        .await?
    };
    let mut options = RunOptions::new(workspace, &prepared.model, contract);
    options.policy = prepared.policy;
    options.redactor = prepared.redactor;
    options.interactive = true;
    options.approvals = Arc::new(AcpApproval {
        session_id: acp_session_id.clone(),
        output: output.clone(),
        pending,
        control: control.clone(),
    });
    options.control = control;
    options.resume_session_id = resume_session_id;
    options.tool_authority_hash = tool_authority_hash;
    options.mcp_connections = mcp_reports
        .into_iter()
        .map(|report| {
            let mut value = serde_json::to_value(report)?;
            value["source"] = json!("acp");
            Ok::<Value, serde_json::Error>(value)
        })
        .collect::<std::result::Result<Vec<_>, _>>()?;
    let sink = AcpEventSink::new(acp_session_id.clone(), output.clone());
    let result = prepared
        .engine
        .run_chain(prepared.provider_chain, options, &sink)
        .await
        .map_err(|error| {
            let runtime_session_id = sink
                .events()
                .ok()
                .and_then(|events| events.first().map(|event| event.session_id));
            match runtime_session_id {
                Some(runtime_session_id) => anyhow!(
                    "runtime session {runtime_session_id} is recoverable or failed: {error}"
                ),
                None => anyhow!(error),
            }
        })?;
    let surface = SurfaceSnapshot::from_result(&result, &sink.events()?);
    output
        .send(session_update(
            &acp_session_id,
            json!({
                "sessionUpdate": "agent_message_chunk",
                "messageId": sink.message_id,
                "content": {
                    "type": "text",
                    "text": proof_summary(&surface, result.session_id)
                },
                "_meta": {"proofborne": {"surface": &surface}}
            }),
        ))
        .map_err(|_| anyhow!("ACP client disconnected"))?;
    Ok((result, sink.last_hash()?, surface))
}

fn sink_runtime_session_id_from_error(error: &anyhow::Error) -> Result<Option<Uuid>> {
    let message = error.to_string();
    let Some(rest) = message.strip_prefix("runtime session ") else {
        return Ok(None);
    };
    let Some((id, _)) = rest.split_once(' ') else {
        return Ok(None);
    };
    Ok(Some(
        Uuid::parse_str(id).context("invalid runtime session id in run error")?,
    ))
}

/// Resolves the exact configured provider/model that created a recoverable
/// runtime session. Resuming through another profile would make the durable
/// proof chain claim a provider/model that did not actually continue it.
async fn prepare_engine_for_resume(
    workspace: &Path,
    runtime_session_id: Uuid,
    policy: PolicyPreset,
    attached_tools: &ToolRegistry,
    attached_secrets: &[String],
) -> Result<super::PreparedEngine> {
    let paths = RuntimePaths::for_workspace(workspace)?;
    let store = SessionStore::open(paths.workspace_data.join("sessions.sqlite3"))?;
    store.load_resumable_checkpoint(runtime_session_id)?;
    let record = store.load_session(runtime_session_id)?;
    let config = proofborne_runtime::AppConfig::load(workspace)?;
    let profile = config
        .providers
        .iter()
        .find_map(|(name, profile)| {
            (profile.provider == record.provider && profile.model == record.model)
                .then_some(name.as_str())
        })
        .ok_or_else(|| anyhow!(
            "resume requires the recorded provider/model {}/{}; no matching configured profile exists",
            record.provider,
            record.model
        ))?;
    prepare_engine_with_tools(
        workspace,
        Some(profile),
        Some(policy),
        Some(attached_tools),
        attached_secrets,
    )
    .await
}

fn proof_summary(surface: &SurfaceSnapshot, runtime_session_id: Uuid) -> String {
    let reason = surface
        .reason
        .as_deref()
        .map_or_else(String::new, |reason| format!(" · reason={reason}"));
    format!(
        "\n\nProof gate: {:?} · claimScope={:?} · assurance={:?} · {} evidence nodes · runtime session {}{}",
        surface.outcome,
        surface.claim_scope,
        surface.assurance_level,
        surface.evidence_count,
        runtime_session_id,
        reason,
    )
}

struct AcpApproval {
    session_id: String,
    output: mpsc::UnboundedSender<Value>,
    pending: PendingClientRequests,
    control: RunControl,
}

#[async_trait]
impl ApprovalHandler for AcpApproval {
    async fn approve(&self, request: &PolicyRequest, reason: &str) -> bool {
        let request_id = format!("proofborne_permission_{}", Uuid::now_v7().simple());
        let tool_call_id = format!("approval_{}", Uuid::now_v7().simple());
        let (response_tx, response_rx) = oneshot::channel();
        let registered = self.pending.lock().map_err(|_| ()).map(|mut pending| {
            if self.control.is_cancelled() {
                false
            } else {
                pending.insert(
                    request_id.clone(),
                    PendingClientRequest {
                        session_id: self.session_id.clone(),
                        response: response_tx,
                    },
                );
                true
            }
        });
        if registered != Ok(true) {
            return false;
        }
        let action_scope = serde_json::to_value(request).unwrap_or(Value::Null);
        let permission_request = json!({
            "jsonrpc": "2.0",
            "id": request_id,
            "method": "session/request_permission",
            "params": {
                "sessionId": self.session_id,
                "toolCall": {
                    "toolCallId": tool_call_id,
                    "title": format!("Approve exact {} action", request.tool),
                    "kind": tool_kind_for_action(request.action_class),
                    "status": "pending",
                    "rawInput": action_scope,
                    "_meta": {
                        "proofborne": {
                            "reason": reason,
                            "scope": "exact_action_once"
                        }
                    }
                },
                "options": [
                    {"optionId": "allow_once", "name": "Allow once", "kind": "allow_once"},
                    {"optionId": "reject_once", "name": "Reject", "kind": "reject_once"}
                ]
            }
        });
        if self.output.send(permission_request).is_err() {
            if let Ok(mut pending) = self.pending.lock() {
                pending.remove(&request_id);
            }
            return false;
        }
        let approved = match response_rx.await {
            Ok(response) => permission_selected(&response, "allow_once"),
            Err(_) => false,
        };
        if let Ok(mut pending) = self.pending.lock() {
            pending.remove(&request_id);
        }
        approved && !self.control.is_cancelled()
    }
}

fn permission_selected(response: &Value, expected_option: &str) -> bool {
    response
        .pointer("/result/outcome/outcome")
        .and_then(Value::as_str)
        == Some("selected")
        && response
            .pointer("/result/outcome/optionId")
            .and_then(Value::as_str)
            == Some(expected_option)
}

fn tool_kind_for_action(action: ActionClass) -> &'static str {
    match action {
        ActionClass::Read => "read",
        ActionClass::Diagnose | ActionClass::Execute => "execute",
        ActionClass::WorkspaceWrite => "edit",
        ActionClass::Network => "fetch",
        ActionClass::Sensitive => "other",
    }
}

struct AcpEventSink {
    session_id: String,
    output: mpsc::UnboundedSender<Value>,
    last_hash: Arc<Mutex<Option<String>>>,
    captured: Arc<Mutex<Vec<EventEnvelope>>>,
    message_id: String,
    streamed_text: Arc<Mutex<bool>>,
    streamed_tool_arguments: Arc<Mutex<HashMap<String, String>>>,
}

impl AcpEventSink {
    fn new(session_id: String, output: mpsc::UnboundedSender<Value>) -> Self {
        Self {
            message_id: format!("msg_{}", Uuid::now_v7().simple()),
            session_id,
            output,
            last_hash: Arc::new(Mutex::new(None)),
            captured: Arc::new(Mutex::new(Vec::new())),
            streamed_text: Arc::new(Mutex::new(false)),
            streamed_tool_arguments: Arc::new(Mutex::new(HashMap::new())),
        }
    }

    fn last_hash(&self) -> Result<String> {
        Ok(self
            .last_hash
            .lock()
            .map_err(|_| anyhow!("ACP event hash lock poisoned"))?
            .clone()
            .unwrap_or_default())
    }

    fn events(&self) -> Result<Vec<EventEnvelope>> {
        Ok(self
            .captured
            .lock()
            .map_err(|_| anyhow!("ACP event collection lock poisoned"))?
            .clone())
    }

    fn send_update(&self, update: Value) -> std::result::Result<(), String> {
        self.output
            .send(session_update(&self.session_id, update))
            .map_err(|_| "ACP client disconnected".to_owned())
    }

    fn send_agent_text(
        &self,
        text: &str,
        event: &EventEnvelope,
    ) -> std::result::Result<(), String> {
        if text.is_empty() {
            return Ok(());
        }
        self.send_update(json!({
            "sessionUpdate": "agent_message_chunk",
            "messageId": self.message_id,
            "content": {"type": "text", "text": text},
            "_meta": {"proofborneEventHash": event.hash}
        }))
    }
}

impl EventSink for AcpEventSink {
    fn emit(&self, event: &EventEnvelope) -> std::result::Result<(), String> {
        *self
            .last_hash
            .lock()
            .map_err(|_| "ACP event hash lock poisoned".to_owned())? = Some(event.hash.clone());
        self.captured
            .lock()
            .map_err(|_| "ACP event collection lock poisoned".to_owned())?
            .push(event.clone());
        match event.kind.as_str() {
            "provider.text.delta" | "provider.output.delta" => {
                let text = event
                    .payload
                    .get("delta")
                    .or_else(|| event.payload.get("text"))
                    .and_then(Value::as_str)
                    .unwrap_or_default();
                *self
                    .streamed_text
                    .lock()
                    .map_err(|_| "ACP stream state lock poisoned".to_owned())? = true;
                self.send_agent_text(text, event)?;
            }
            "provider.stream.event" => match event.payload.get("type").and_then(Value::as_str) {
                Some("textDelta") => {
                    let text = event
                        .payload
                        .get("delta")
                        .and_then(Value::as_str)
                        .unwrap_or_default();
                    *self
                        .streamed_text
                        .lock()
                        .map_err(|_| "ACP stream state lock poisoned".to_owned())? = true;
                    self.send_agent_text(text, event)?;
                }
                Some("toolCallStarted") => self.send_streamed_tool_start(event)?,
                Some("toolCallArgumentsDelta") => self.send_streamed_tool_arguments(event)?,
                Some("toolCallCompleted") => self.send_streamed_tool_complete(event)?,
                Some("completed") => {
                    let already_streamed = *self
                        .streamed_text
                        .lock()
                        .map_err(|_| "ACP stream state lock poisoned".to_owned())?;
                    if !already_streamed {
                        let text = event
                            .payload
                            .pointer("/turn/outputText")
                            .and_then(Value::as_str)
                            .unwrap_or_default();
                        self.send_agent_text(text, event)?;
                        *self
                            .streamed_text
                            .lock()
                            .map_err(|_| "ACP stream state lock poisoned".to_owned())? = true;
                    }
                }
                _ => {}
            },
            "provider.turn.completed" => {
                let already_streamed = *self
                    .streamed_text
                    .lock()
                    .map_err(|_| "ACP stream state lock poisoned".to_owned())?;
                if !already_streamed {
                    self.send_agent_text(
                        event
                            .payload
                            .get("outputText")
                            .and_then(Value::as_str)
                            .unwrap_or_default(),
                        event,
                    )?;
                }
            }
            "tool.requested" => {
                let tool_call_id = event
                    .payload
                    .get("id")
                    .and_then(Value::as_str)
                    .unwrap_or("proofborne-tool");
                let title = event
                    .payload
                    .get("name")
                    .and_then(Value::as_str)
                    .unwrap_or("Proofborne tool");
                let streamed = self
                    .streamed_tool_arguments
                    .lock()
                    .map_err(|_| "ACP streamed tool lock poisoned".to_owned())?
                    .contains_key(tool_call_id);
                let session_update = if streamed {
                    "tool_call_update"
                } else {
                    "tool_call"
                };
                self.send_update(json!({
                    "sessionUpdate": session_update,
                    "toolCallId": tool_call_id,
                    "title": title,
                    "kind": "other",
                    "status": "pending",
                    "rawInput": event.payload.get("arguments").cloned().unwrap_or(Value::Null),
                    "_meta": {"proofborneEventHash": event.hash}
                }))?;
            }
            "action.started" => {
                if let Some(tool_call_id) = event.payload.get("toolCallId").and_then(Value::as_str)
                {
                    self.send_update(json!({
                        "sessionUpdate": "tool_call_update",
                        "toolCallId": tool_call_id,
                        "title": event.payload.get("tool").cloned().unwrap_or_else(|| json!("Proofborne action")),
                        "kind": action_kind_from_event(event),
                        "status": "in_progress",
                        "_meta": {"proofborneEventHash": event.hash}
                    }))?;
                }
            }
            "tool.completed" | "verification.completed" => {
                self.send_tool_result(event, true)?;
            }
            "tool.failed" | "policy.denied" => self.send_tool_result(event, false)?,
            "verification.started" => {
                let call = event.payload.get("call").unwrap_or(&Value::Null);
                if let Some(tool_call_id) = call.get("id").and_then(Value::as_str) {
                    self.send_update(json!({
                        "sessionUpdate": "tool_call",
                        "toolCallId": tool_call_id,
                        "title": "Verify acceptance criterion",
                        "kind": "execute",
                        "status": "in_progress",
                        "rawInput": call.get("arguments").cloned().unwrap_or(Value::Null),
                        "_meta": {"proofborneEventHash": event.hash}
                    }))?;
                }
            }
            _ => {}
        }
        Ok(())
    }
}

impl AcpEventSink {
    fn send_streamed_tool_start(&self, event: &EventEnvelope) -> std::result::Result<(), String> {
        let Some(tool_call_id) = event.payload.get("id").and_then(Value::as_str) else {
            return Ok(());
        };
        self.streamed_tool_arguments
            .lock()
            .map_err(|_| "ACP streamed tool lock poisoned".to_owned())?
            .entry(tool_call_id.to_owned())
            .or_default();
        self.send_update(json!({
            "sessionUpdate": "tool_call",
            "toolCallId": tool_call_id,
            "title": event.payload.get("name").cloned().unwrap_or_else(|| json!("Proofborne tool")),
            "kind": "other",
            "status": "pending",
            "_meta": {"proofborneEventHash": event.hash}
        }))
    }

    fn send_streamed_tool_arguments(
        &self,
        event: &EventEnvelope,
    ) -> std::result::Result<(), String> {
        let Some(tool_call_id) = event.payload.get("id").and_then(Value::as_str) else {
            return Ok(());
        };
        let delta = event
            .payload
            .get("delta")
            .and_then(Value::as_str)
            .unwrap_or_default();
        let accumulated = {
            let mut arguments = self
                .streamed_tool_arguments
                .lock()
                .map_err(|_| "ACP streamed tool lock poisoned".to_owned())?;
            let accumulated = arguments.entry(tool_call_id.to_owned()).or_default();
            accumulated.push_str(delta);
            accumulated.clone()
        };
        self.send_update(json!({
            "sessionUpdate": "tool_call_update",
            "toolCallId": tool_call_id,
            "status": "pending",
            "rawInput": {"partialJson": accumulated},
            "_meta": {"proofborneEventHash": event.hash}
        }))
    }

    fn send_streamed_tool_complete(
        &self,
        event: &EventEnvelope,
    ) -> std::result::Result<(), String> {
        let call = event.payload.get("call").unwrap_or(&Value::Null);
        let Some(tool_call_id) = call.get("id").and_then(Value::as_str) else {
            return Ok(());
        };
        self.send_update(json!({
            "sessionUpdate": "tool_call_update",
            "toolCallId": tool_call_id,
            "title": call.get("name").cloned().unwrap_or_else(|| json!("Proofborne tool")),
            "status": "pending",
            "rawInput": call.get("arguments").cloned().unwrap_or(Value::Null),
            "_meta": {"proofborneEventHash": event.hash}
        }))
    }

    fn send_tool_result(
        &self,
        event: &EventEnvelope,
        default_success: bool,
    ) -> std::result::Result<(), String> {
        let result = event.payload.get("result").unwrap_or(&event.payload);
        let Some(tool_call_id) = result.get("callId").and_then(Value::as_str) else {
            return Ok(());
        };
        let success = result
            .get("success")
            .and_then(Value::as_bool)
            .unwrap_or(default_success);
        self.send_update(json!({
            "sessionUpdate": "tool_call_update",
            "toolCallId": tool_call_id,
            "status": if success { "completed" } else { "failed" },
            "rawOutput": result.get("output").cloned().unwrap_or_else(|| result.clone()),
            "_meta": {"proofborneEventHash": event.hash}
        }))
    }
}

fn action_kind_from_event(event: &EventEnvelope) -> &'static str {
    match event.payload.get("actionClass").and_then(Value::as_str) {
        Some("read") => "read",
        Some("diagnose" | "execute") => "execute",
        Some("workspace_write") => "edit",
        Some("network") => "fetch",
        _ => "other",
    }
}

fn validate_session_setup(params: &Value, allow_client_mcp: bool) -> Result<()> {
    let servers = params
        .get("mcpServers")
        .and_then(Value::as_array)
        .ok_or_else(|| anyhow!("mcpServers must be an array"))?;
    if !servers.is_empty() && !allow_client_mcp {
        bail!("client-supplied MCP requires the operator flag --allow-client-mcp");
    }
    if params
        .get("additionalDirectories")
        .and_then(Value::as_array)
        .is_some_and(|directories| !directories.is_empty())
    {
        bail!("this build does not advertise ACP additional workspace directories");
    }
    Ok(())
}

fn parse_acp_mcp_servers(params: &Value) -> Result<Vec<AcpMcpServer>> {
    let values = params
        .get("mcpServers")
        .and_then(Value::as_array)
        .ok_or_else(|| anyhow!("mcpServers must be an array"))?;
    if values.len() > MAX_MCP_SERVERS {
        bail!("mcpServers exceeds the limit of {MAX_MCP_SERVERS}");
    }
    let mut names = BTreeSet::new();
    let mut servers = Vec::with_capacity(values.len());
    for value in values {
        let wire: AcpMcpServerWire = serde_json::from_value(value.clone())
            .context("invalid ACP v1 MCP server configuration")?;
        let server = match wire {
            AcpMcpServerWire::Stdio(wire) => parse_stdio_mcp(wire)?,
            AcpMcpServerWire::Http(wire) => parse_http_mcp(wire)?,
        };
        if !names.insert(server.descriptor.name.clone()) {
            bail!("duplicate MCP server name: {}", server.descriptor.name);
        }
        servers.push(server);
    }
    servers.sort_by(|left, right| left.descriptor.name.cmp(&right.descriptor.name));
    Ok(servers)
}

fn validate_acp_mcp_policy(workspace: &Path, servers: &[AcpMcpServer]) -> Result<()> {
    if servers.is_empty() {
        return Ok(());
    }
    let config = AppConfig::load(workspace)?;
    validate_mcp_aliases(servers, &config.policy.mcp_allow)
}

fn validate_mcp_aliases(servers: &[AcpMcpServer], allowed_aliases: &[String]) -> Result<()> {
    for server in servers {
        let allowed = allowed_aliases
            .iter()
            .any(|alias| alias == "*" || alias == &server.descriptor.name);
        if !allowed {
            bail!(
                "ACP MCP server '{}' is not authorized by policy.mcp_allow",
                server.descriptor.name
            );
        }
    }
    Ok(())
}

fn parse_stdio_mcp(wire: AcpMcpStdioWire) -> Result<AcpMcpServer> {
    validate_mcp_name(&wire.name)?;
    if !wire.command.is_absolute() {
        bail!("ACP stdio MCP command must be absolute");
    }
    if wire.args.len() > MAX_MCP_ARGS
        || wire
            .args
            .iter()
            .any(|argument| argument.len() > MAX_MCP_VALUE || argument.contains('\0'))
    {
        bail!("ACP stdio MCP arguments exceed safe limits");
    }
    let env = validate_name_values(wire.env, NameValueKind::Environment)?;
    let command = wire.command.to_string_lossy().into_owned();
    if command.len() > MAX_MCP_VALUE || command.contains('\0') {
        bail!("ACP stdio MCP command exceeds safe limits");
    }
    let mut credential_fields = env.iter().map(|(name, _)| name.clone()).collect::<Vec<_>>();
    credential_fields.sort();
    Ok(AcpMcpServer {
        descriptor: PersistedMcpServer {
            name: wire.name,
            transport: "stdio".to_owned(),
            endpoint: command.clone(),
            args: wire.args.clone(),
            credential_fields,
        },
        transport: AcpMcpTransport::Stdio {
            command,
            args: wire.args,
            env,
        },
    })
}

fn parse_http_mcp(wire: AcpMcpHttpWire) -> Result<AcpMcpServer> {
    validate_mcp_name(&wire.name)?;
    if wire.transport != "http" {
        bail!("unsupported ACP MCP transport: {}", wire.transport);
    }
    let parsed = url::Url::parse(&wire.url).context("invalid ACP HTTP MCP URL")?;
    if !matches!(parsed.scheme(), "http" | "https") {
        bail!("ACP HTTP MCP URL must use http or https");
    }
    if !parsed.username().is_empty()
        || parsed.password().is_some()
        || parsed.query().is_some()
        || parsed.fragment().is_some()
    {
        bail!("ACP HTTP MCP URL must not contain userinfo, query, or fragment credentials");
    }
    let headers = validate_name_values(wire.headers, NameValueKind::Header)?;
    let mut credential_fields = headers
        .iter()
        .map(|(name, _)| name.to_ascii_lowercase())
        .collect::<Vec<_>>();
    credential_fields.sort();
    Ok(AcpMcpServer {
        descriptor: PersistedMcpServer {
            name: wire.name,
            transport: "http".to_owned(),
            endpoint: parsed.to_string(),
            args: Vec::new(),
            credential_fields,
        },
        transport: AcpMcpTransport::Http {
            url: parsed.to_string(),
            headers,
        },
    })
}

#[derive(Clone, Copy)]
enum NameValueKind {
    Environment,
    Header,
}

fn validate_name_values(
    values: Vec<AcpNameValue>,
    kind: NameValueKind,
) -> Result<Vec<(String, String)>> {
    if values.len() > MAX_MCP_FIELDS {
        bail!("ACP MCP credentials exceed the limit of {MAX_MCP_FIELDS}");
    }
    let mut names = BTreeSet::new();
    let mut result = Vec::with_capacity(values.len());
    for value in values {
        let valid_name = match kind {
            NameValueKind::Environment => valid_environment_name(&value.name),
            NameValueKind::Header => valid_header_name(&value.name),
        };
        if !valid_name {
            bail!("invalid ACP MCP credential field name");
        }
        if (!value.value.is_empty() && value.value.len() < 4)
            || value.value.len() > MAX_MCP_VALUE
            || value.value.contains('\0')
            || value
                .value
                .chars()
                .any(|character| character.is_control() && character != '\t')
        {
            bail!("ACP MCP credential values must be empty or 4-65536 safe characters");
        }
        let comparison = match kind {
            NameValueKind::Environment => value.name.clone(),
            NameValueKind::Header => value.name.to_ascii_lowercase(),
        };
        if !names.insert(comparison) {
            bail!("duplicate ACP MCP credential field name");
        }
        result.push((value.name, value.value));
    }
    Ok(result)
}

fn validate_mcp_name(name: &str) -> Result<()> {
    if name.is_empty()
        || name.len() > 48
        || !name
            .chars()
            .all(|character| character.is_ascii_alphanumeric() || matches!(character, '_' | '-'))
    {
        bail!("MCP server name must use 1-48 ASCII letters, digits, '_' or '-'");
    }
    Ok(())
}

fn valid_environment_name(name: &str) -> bool {
    let mut characters = name.chars();
    characters
        .next()
        .is_some_and(|character| character.is_ascii_alphabetic() || character == '_')
        && characters.all(|character| character.is_ascii_alphanumeric() || character == '_')
}

fn valid_header_name(name: &str) -> bool {
    !name.is_empty()
        && name.bytes().all(|byte| {
            byte.is_ascii_alphanumeric()
                || matches!(
                    byte,
                    b'!' | b'#'
                        | b'$'
                        | b'%'
                        | b'&'
                        | b'\''
                        | b'*'
                        | b'+'
                        | b'-'
                        | b'.'
                        | b'^'
                        | b'_'
                        | b'`'
                        | b'|'
                        | b'~'
                )
        })
}

fn mcp_descriptors(servers: &[AcpMcpServer]) -> Vec<PersistedMcpServer> {
    servers
        .iter()
        .map(|server| server.descriptor.clone())
        .collect()
}

fn mcp_authority_hash(
    descriptors: &[PersistedMcpServer],
    runtime: &AcpMcpRuntime,
) -> Option<String> {
    if descriptors.is_empty() {
        None
    } else {
        Some(hash_json(&json!({
            "schemaVersion": SCHEMA_VERSION,
            "kind": "proofborne.mcp-authority",
            "servers": descriptors,
            "connections": runtime.reports,
            "toolDefinitions": runtime.tools.definitions(),
        })))
    }
}

async fn connect_acp_mcp(servers: &[AcpMcpServer]) -> Result<AcpMcpRuntime> {
    if servers.is_empty() {
        return Ok(AcpMcpRuntime::empty());
    }
    let attachments = servers
        .iter()
        .map(|server| McpServerAttachment {
            alias: server.descriptor.name.clone(),
            transport: match &server.transport {
                AcpMcpTransport::Stdio { command, args, env } => McpTransportMaterial::Stdio {
                    command: command.clone(),
                    args: args.clone(),
                    env: env.iter().cloned().collect::<BTreeMap<_, _>>(),
                },
                AcpMcpTransport::Http { url, headers } => McpTransportMaterial::Http {
                    url: url.clone(),
                    headers: headers.iter().cloned().collect::<BTreeMap<_, _>>(),
                },
            },
        })
        .collect::<Vec<_>>();
    let secrets = acp_mcp_secret_values(servers);
    let mut tools = ToolRegistry::default();
    let reports = register_mcp_attachments(&attachments, &mut tools)
        .await
        .context("failed to attach ACP-provided MCP servers")?;
    Ok(AcpMcpRuntime {
        tools,
        secrets,
        reports,
    })
}

fn acp_mcp_secret_values(servers: &[AcpMcpServer]) -> Vec<String> {
    let mut secrets = Vec::new();
    for server in servers {
        let values = match &server.transport {
            AcpMcpTransport::Stdio { env, .. } => env,
            AcpMcpTransport::Http { headers, .. } => headers,
        };
        secrets.extend(
            values
                .iter()
                .map(|(_, value)| value)
                .filter(|value| !value.is_empty())
                .cloned(),
        );
    }
    secrets
}

fn redacted_mcp_setup_error(servers: &[AcpMcpServer], error: &anyhow::Error) -> String {
    let redactor = proofborne_runtime::Redactor::with_secrets(acp_mcp_secret_values(servers));
    let (message, _) = redactor.redact_str(&format!("MCP setup failed: {error:#}"));
    message
}

fn acp_workspace(params: &Value, default_workspace: &Path) -> Result<PathBuf> {
    let supplied = params
        .get("cwd")
        .and_then(Value::as_str)
        .map_or_else(|| default_workspace.to_path_buf(), PathBuf::from);
    if !supplied.is_absolute() {
        bail!("ACP cwd must be absolute");
    }
    supplied
        .canonicalize()
        .with_context(|| format!("ACP cwd does not exist: {}", supplied.display()))
}

fn prompt_text(params: &Value) -> Result<String> {
    let blocks = params
        .get("prompt")
        .and_then(Value::as_array)
        .ok_or_else(|| anyhow!("prompt must be an array"))?;
    let text = blocks
        .iter()
        .filter(|block| block.get("type").and_then(Value::as_str) == Some("text"))
        .filter_map(|block| block.get("text").and_then(Value::as_str))
        .collect::<Vec<_>>()
        .join("\n");
    if text.trim().is_empty() {
        bail!("at least one non-empty text content block is required");
    }
    Ok(text)
}

fn prompt_contract(params: &Value) -> Result<TaskContract> {
    let prompt = prompt_text(params)?;
    let Some(value) = params.pointer("/_meta/proofborne/contract") else {
        return Ok(TaskContract::automatic(prompt));
    };
    let mut contract: TaskContract = serde_json::from_value(value.clone())
        .context("_meta.proofborne.contract is not a valid proofborne.v1 TaskContract")?;
    if contract.goal != prompt {
        bail!("ACP prompt text differs from the goal pinned in the TaskContract");
    }
    if !contract.confirmed {
        bail!("ACP TaskContract must be explicitly confirmed");
    }
    for criterion in &mut contract.criteria {
        if criterion.waiver.is_none() {
            criterion.state = CriterionState::Pending;
            criterion.evidence_ids.clear();
        }
    }
    contract.validate()?;
    Ok(contract)
}

fn persist_runtime_session(
    sessions: &Sessions,
    registry: &AcpSessionRegistry,
    acp_session_id: &str,
    runtime_session_id: Uuid,
) -> Result<()> {
    let metadata = {
        let mut guard = sessions
            .lock()
            .map_err(|_| anyhow!("ACP session lock poisoned"))?;
        let session = guard
            .get_mut(acp_session_id)
            .ok_or_else(|| anyhow!("ACP session disappeared during prompt"))?;
        if !session
            .metadata
            .runtime_session_ids
            .contains(&runtime_session_id)
        {
            session
                .metadata
                .runtime_session_ids
                .push(runtime_session_id);
        }
        session.metadata.clone()
    };
    registry.save(&metadata)
}

fn load_finalized_runs(metadata: &PersistedAcpSession) -> Result<Vec<FinalizedRun>> {
    if metadata.runtime_session_ids.is_empty() {
        bail!("ACP session has no finalized prompt turns to load");
    }
    let paths = RuntimePaths::for_workspace(&metadata.workspace)?;
    let store = SessionStore::open(paths.workspace_data.join("sessions.sqlite3"))?;
    metadata
        .runtime_session_ids
        .iter()
        .map(|id| {
            let record = store.load_session(*id)?;
            if record.checkpoint.is_some() {
                // Recoverable turns have no final proof to replay. They remain
                // attached to the ACP session and are continued by its next
                // prompt using the original runtime UUID.
                return Ok(None);
            }
            let mut contract = record
                .contract
                .ok_or_else(|| anyhow!("runtime session {id} has no final contract"))?;
            let proof = record
                .proof
                .ok_or_else(|| anyhow!("runtime session {id} has no final proof"))?;
            let evaluation = proof.evaluate_detailed(&mut contract)?;
            let stored_outcome: RunOutcome = serde_json::from_value(json!(record.status))
                .with_context(|| format!("runtime session {id} has a non-final status"))?;
            if stored_outcome != evaluation.outcome {
                bail!("runtime session {id} outcome disagrees with its proof");
            }
            let output_text = record
                .events
                .iter()
                .rev()
                .find(|event| event.kind == "provider.turn.completed")
                .and_then(|event| event.payload.get("outputText"))
                .and_then(Value::as_str)
                .unwrap_or_default()
                .to_owned();
            Ok(Some(FinalizedRun {
                events: record.events,
                result: RunResult {
                    session_id: *id,
                    outcome: evaluation.outcome,
                    evaluation,
                    contract,
                    proof,
                    output_text,
                },
            }))
        })
        .filter_map(std::result::Result::transpose)
        .collect()
}

fn recoverable_runtime_session(metadata: &PersistedAcpSession) -> Result<Option<Uuid>> {
    let paths = RuntimePaths::for_workspace(&metadata.workspace)?;
    let store = SessionStore::open(paths.workspace_data.join("sessions.sqlite3"))?;
    let mut recoverable = None;
    for id in &metadata.runtime_session_ids {
        let record = store.load_session(*id)?;
        if record.checkpoint.is_some() {
            store.load_resumable_checkpoint(*id)?;
            if recoverable.replace(*id).is_some() {
                bail!("ACP session has more than one recoverable runtime turn");
            }
        }
    }
    Ok(recoverable)
}

fn replay_runs(
    acp_session_id: &str,
    runs: &[FinalizedRun],
    output: &mpsc::UnboundedSender<Value>,
) -> Result<()> {
    for finalized in runs {
        let run = &finalized.result;
        output
            .send(session_update(
                acp_session_id,
                json!({
                    "sessionUpdate": "user_message_chunk",
                    "messageId": format!("user_{}", run.session_id.simple()),
                    "content": {"type": "text", "text": run.contract.goal}
                }),
            ))
            .map_err(|_| anyhow!("ACP stdout writer stopped"))?;
        let replay_sink = AcpEventSink::new(acp_session_id.to_owned(), output.clone());
        for event in &finalized.events {
            replay_sink.emit(event).map_err(|error| anyhow!(error))?;
        }
        let surface = SurfaceSnapshot::from_result(run, &finalized.events);
        output
            .send(session_update(
                acp_session_id,
                json!({
                    "sessionUpdate": "agent_message_chunk",
                    "messageId": format!("proof_{}", run.session_id.simple()),
                    "content": {"type": "text", "text": proof_summary(&surface, run.session_id)},
                    "_meta": {"proofborne": {"surface": surface, "replayed": true}}
                }),
            ))
            .map_err(|_| anyhow!("ACP stdout writer stopped"))?;
    }
    Ok(())
}

fn validate_acp_session_id(session_id: &str) -> Result<()> {
    let suffix = session_id
        .strip_prefix("sess_")
        .ok_or_else(|| anyhow!("invalid Proofborne ACP session id"))?;
    Uuid::parse_str(suffix).context("invalid Proofborne ACP session id")?;
    Ok(())
}

fn same_path(left: &Path, right: &Path) -> bool {
    #[cfg(windows)]
    {
        left.to_string_lossy()
            .eq_ignore_ascii_case(&right.to_string_lossy())
    }
    #[cfg(not(windows))]
    {
        left == right
    }
}

fn route_client_response(message: &Value, pending: &PendingClientRequests) -> Result<bool> {
    if message.get("method").is_some()
        || (message.get("result").is_none() && message.get("error").is_none())
    {
        return Ok(false);
    }
    let Some(id) = message.get("id").and_then(Value::as_str) else {
        return Ok(true);
    };
    let request = pending
        .lock()
        .map_err(|_| anyhow!("ACP pending request lock poisoned"))?
        .remove(id);
    if let Some(request) = request {
        let _ = request.response.send(message.clone());
    }
    Ok(true)
}

fn cancel_prompt(
    sessions: &Sessions,
    session_id: &str,
    _output: &mpsc::UnboundedSender<Value>,
    pending: &PendingClientRequests,
) -> Result<()> {
    let control = sessions
        .lock()
        .map_err(|_| anyhow!("ACP session lock poisoned"))?
        .get(session_id)
        .and_then(|session| session.active.as_ref())
        .map(|active| active.control.clone());
    if let Some(control) = control {
        control.cancel();
    }
    cancel_pending_permissions(pending, session_id)?;
    Ok(())
}

async fn await_prompt_termination(mut active: ActivePrompt) -> Result<()> {
    let Some(mut task) = active.task.take() else {
        return Ok(());
    };
    match tokio::time::timeout(std::time::Duration::from_secs(5), &mut task).await {
        Ok(Ok(())) => Ok(()),
        Ok(Err(error)) => Err(anyhow!("ACP prompt task failed during close: {error}")),
        Err(_) => {
            task.abort();
            let _ = task.await;
            Err(anyhow!(
                "ACP prompt did not stop within 5 seconds and was terminated"
            ))
        }
    }
}

fn cancel_pending_permissions(pending: &PendingClientRequests, session_id: &str) -> Result<()> {
    let cancelled = {
        let mut guard = pending
            .lock()
            .map_err(|_| anyhow!("ACP pending request lock poisoned"))?;
        let ids = guard
            .iter()
            .filter(|(_, request)| request.session_id == session_id)
            .map(|(id, _)| id.clone())
            .collect::<Vec<_>>();
        ids.into_iter()
            .filter_map(|id| guard.remove(&id))
            .collect::<Vec<_>>()
    };
    for request in cancelled {
        let _ = request.response.send(json!({
            "result": {"outcome": {"outcome": "cancelled"}}
        }));
    }
    Ok(())
}

fn session_update(session_id: &str, update: Value) -> Value {
    json!({
        "jsonrpc": "2.0",
        "method": "session/update",
        "params": {"sessionId": session_id, "update": update}
    })
}

fn send_result(output: &mpsc::UnboundedSender<Value>, id: Value, result: Value) -> Result<()> {
    output
        .send(json!({"jsonrpc": "2.0", "id": id, "result": result}))
        .map_err(|_| anyhow!("ACP stdout writer stopped"))
}

fn send_error(
    output: &mpsc::UnboundedSender<Value>,
    id: Value,
    code: i64,
    message: &str,
) -> Result<()> {
    output
        .send(error_response(id, code, message))
        .map_err(|_| anyhow!("ACP stdout writer stopped"))
}

fn error_response(id: Value, code: i64, message: &str) -> Value {
    json!({
        "jsonrpc": "2.0",
        "id": id,
        "error": {"code": code, "message": message}
    })
}

#[cfg(test)]
mod tests {
    use std::{
        thread,
        time::{Duration, Instant},
    };

    use proofborne_core::{
        Criterion, EvidenceFreshness, EvidenceKind, ProviderCapabilities, ProviderError,
        ProviderRequest, ProviderTurn, StopReason, TokenUsage,
    };
    use proofborne_runtime::{
        CasStore, Provider, ProviderChain, RunOptions, SessionEngine, ToolRegistry,
    };
    use tempfile::tempdir;

    use super::*;

    struct FailingProvider;

    struct HealthyProvider;

    fn test_capabilities() -> ProviderCapabilities {
        ProviderCapabilities {
            streaming: false,
            tools: true,
            structured_output: true,
            multimodal_input: false,
            context_tokens: Some(8_192),
        }
    }

    #[async_trait]
    impl Provider for FailingProvider {
        fn name(&self) -> &str {
            "acp-test-provider"
        }

        fn capabilities(&self) -> ProviderCapabilities {
            test_capabilities()
        }

        async fn turn(&self, _request: ProviderRequest) -> Result<ProviderTurn, ProviderError> {
            Err(ProviderError::Transport(
                "injected provider outage".to_owned(),
            ))
        }
    }

    #[async_trait]
    impl Provider for HealthyProvider {
        fn name(&self) -> &str {
            "acp-test-provider"
        }

        fn capabilities(&self) -> ProviderCapabilities {
            test_capabilities()
        }

        async fn turn(&self, request: ProviderRequest) -> Result<ProviderTurn, ProviderError> {
            Ok(ProviderTurn {
                provider: self.name().to_owned(),
                model: request.model,
                response_id: Some("recovered-response".to_owned()),
                output_text: "recovered".to_owned(),
                tool_calls: Vec::new(),
                stop_reason: StopReason::EndTurn,
                usage: TokenUsage::default(),
                opaque: json!({}),
            })
        }
    }

    fn resumable_contract() -> TaskContract {
        let mut criterion = Criterion::required("verified", "verified");
        criterion.evidence_requirement.allowed_kinds =
            [EvidenceKind::Process].into_iter().collect();
        criterion.evidence_requirement.allowed_producers =
            ["verify.exec".to_owned()].into_iter().collect();
        criterion.evidence_requirement.freshness = EvidenceFreshness::FinalWorkspaceState;
        let mut contract = TaskContract::new("complete the task", vec![criterion]);
        contract.confirmed = true;
        contract
    }

    #[test]
    fn extracts_only_supported_text_blocks() {
        let params = json!({
            "prompt": [
                {"type": "text", "text": "first"},
                {"type": "resource_link", "uri": "file:///tmp/a"},
                {"type": "text", "text": "second"}
            ]
        });
        assert_eq!(prompt_text(&params).unwrap(), "first\nsecond");
    }

    #[test]
    fn initialization_advertises_only_implemented_optional_capabilities() {
        let message = initialize_result(true);
        assert_eq!(message["protocolVersion"], 1);
        assert_eq!(message["agentCapabilities"]["loadSession"], true);
        assert!(
            message["agentCapabilities"]["sessionCapabilities"]
                .get("resume")
                .is_none()
        );
        assert!(
            message["agentCapabilities"]["sessionCapabilities"]
                .get("close")
                .is_some()
        );
        assert_eq!(
            message["agentCapabilities"]["mcpCapabilities"]["http"],
            true
        );
    }

    #[test]
    fn parses_stdio_and_http_mcp_without_persisting_secrets() {
        let command = std::env::current_exe().unwrap();
        let params = json!({
            "mcpServers": [
                {
                    "name": "local-tools",
                    "command": command,
                    "args": ["--fixture"],
                    "env": [{"name": "MCP_TOKEN", "value": "stdio-secret-value"}]
                },
                {
                    "type": "http",
                    "name": "remote-docs",
                    "url": "https://example.invalid/mcp",
                    "headers": [{"name": "Authorization", "value": "http-secret-value"}]
                }
            ]
        });
        let servers = parse_acp_mcp_servers(&params).unwrap();
        let descriptors = mcp_descriptors(&servers);
        let authority = mcp_authority_hash(&descriptors, &AcpMcpRuntime::empty());
        assert_eq!(servers.len(), 2);
        assert!(authority.is_some());
        let persisted = serde_json::to_string(&descriptors).unwrap();
        assert!(!persisted.contains("stdio-secret-value"));
        assert!(!persisted.contains("http-secret-value"));
        assert!(persisted.contains("MCP_TOKEN"));
        assert!(persisted.contains("authorization"));
    }

    #[test]
    fn rejects_unsupported_or_ambiguous_mcp_boundaries() {
        for params in [
            json!({"mcpServers": [{"type": "sse", "name": "old", "url": "https://example.invalid", "headers": []}]}),
            json!({"mcpServers": [{"name": "relative", "command": "server", "args": [], "env": []}]}),
            json!({"mcpServers": [
                {"type": "http", "name": "same", "url": "https://one.invalid", "headers": []},
                {"type": "http", "name": "same", "url": "https://two.invalid", "headers": []}
            ]}),
            json!({"mcpServers": [{"type": "http", "name": "bad name", "url": "https://example.invalid", "headers": []}]}),
            json!({"mcpServers": [{"type": "http", "name": "headers", "url": "https://example.invalid", "headers": [{"name": "X-Test", "value": "ok\r\nInjected: true"}]}]}),
            json!({"mcpServers": [{"type": "http", "name": "url-secret", "url": "https://user:secret@example.invalid/mcp?token=secret", "headers": []}]}),
        ] {
            assert!(parse_acp_mcp_servers(&params).is_err());
        }
    }

    #[test]
    fn client_mcp_aliases_require_explicit_project_authority() {
        let command = std::env::current_exe().unwrap();
        let servers = parse_acp_mcp_servers(&json!({
            "mcpServers": [{"name": "fixture", "command": command, "args": [], "env": []}]
        }))
        .unwrap();
        assert!(validate_mcp_aliases(&servers, &[]).is_err());
        assert!(validate_mcp_aliases(&servers, &["other".to_owned()]).is_err());
        assert!(validate_mcp_aliases(&servers, &["fixture".to_owned()]).is_ok());
        assert!(validate_mcp_aliases(&servers, &["*".to_owned()]).is_ok());
    }

    #[test]
    fn registry_round_trips_only_valid_namespaced_ids() {
        let directory = tempdir().unwrap();
        let registry = AcpSessionRegistry::at(directory.path().to_path_buf()).unwrap();
        let session_id = format!("sess_{}", Uuid::now_v7().simple());
        let metadata = PersistedAcpSession {
            schema_version: SCHEMA_VERSION.to_owned(),
            session_id: session_id.clone(),
            workspace: directory.path().to_path_buf(),
            runtime_session_ids: vec![Uuid::now_v7()],
            mcp_servers: Vec::new(),
            tool_authority_hash: None,
        };
        registry.save(&metadata).unwrap();
        assert_eq!(registry.load(&session_id).unwrap(), metadata);
        assert!(registry.load("../escape").is_err());
    }

    #[tokio::test]
    async fn permission_bridge_approves_only_the_exact_allow_once_option() {
        let (output, mut messages) = mpsc::unbounded_channel();
        let pending = PendingClientRequests::default();
        let approval = Arc::new(AcpApproval {
            session_id: "sess_test".to_owned(),
            output,
            pending: pending.clone(),
            control: RunControl::default(),
        });
        let request = PolicyRequest {
            tool: "patch_file".to_owned(),
            action_class: ActionClass::WorkspaceWrite,
            target: Some(PathBuf::from("C:/workspace/file.rs")),
            executable: None,
            interactive: true,
        };
        let task = tokio::spawn({
            let approval = approval.clone();
            async move { approval.approve(&request, "workspace is not trusted").await }
        });
        let message = messages.recv().await.unwrap();
        assert_eq!(message["method"], "session/request_permission");
        assert_eq!(
            message["params"]["toolCall"]["rawInput"]["tool"],
            "patch_file"
        );
        let id = message["id"].as_str().unwrap();
        route_client_response(
            &json!({
                "jsonrpc": "2.0",
                "id": id,
                "result": {"outcome": {"outcome": "selected", "optionId": "allow_once"}}
            }),
            &pending,
        )
        .unwrap();
        assert!(task.await.unwrap());
    }

    #[tokio::test]
    async fn event_sink_emits_the_complete_tool_status_lifecycle() {
        let (output, mut messages) = mpsc::unbounded_channel();
        let sink = AcpEventSink::new("sess_test".to_owned(), output);
        let runtime_session = Uuid::now_v7();
        let events = [
            (
                0,
                "tool.requested",
                json!({"id": "call-1", "name": "patch_file", "arguments": {}}),
            ),
            (
                1,
                "action.started",
                json!({"toolCallId": "call-1", "tool": "patch_file", "actionClass": "workspace_write"}),
            ),
            (
                2,
                "tool.completed",
                json!({"callId": "call-1", "success": true, "output": {"ok": true}}),
            ),
        ]
        .into_iter()
        .map(|(seq, kind, payload)| {
            EventEnvelope::new(
                runtime_session,
                seq,
                kind,
                payload,
                None,
            )
        })
        .collect::<Vec<_>>();
        for event in &events {
            sink.emit(event).unwrap();
        }
        assert_eq!(sink.events().unwrap(), events);
        let updates = [
            messages.recv().await.unwrap(),
            messages.recv().await.unwrap(),
            messages.recv().await.unwrap(),
        ];
        assert_eq!(updates[0]["params"]["update"]["status"], "pending");
        assert_eq!(updates[1]["params"]["update"]["status"], "in_progress");
        assert_eq!(updates[2]["params"]["update"]["status"], "completed");
    }

    #[tokio::test]
    async fn cancellation_is_delegated_to_the_runtime_control() {
        let (output, _messages) = mpsc::unbounded_channel();
        let pending = PendingClientRequests::default();
        let sessions = Sessions::default();
        let control = RunControl::default();
        let session_id = format!("sess_{}", Uuid::now_v7().simple());
        sessions.lock().unwrap().insert(
            session_id.clone(),
            AcpSession {
                metadata: PersistedAcpSession {
                    schema_version: SCHEMA_VERSION.to_owned(),
                    session_id: session_id.clone(),
                    workspace: PathBuf::from("C:/workspace"),
                    runtime_session_ids: Vec::new(),
                    mcp_servers: Vec::new(),
                    tool_authority_hash: None,
                },
                active: Some(ActivePrompt {
                    control: control.clone(),
                    task: None,
                }),
                mcp: AcpMcpRuntime::empty(),
            },
        );
        cancel_prompt(&sessions, &session_id, &output, &pending).unwrap();
        assert!(control.is_cancelled());
    }

    #[tokio::test]
    async fn approval_cannot_register_after_cancel_drains_an_empty_pending_map() {
        let (output, mut messages) = mpsc::unbounded_channel();
        let pending = PendingClientRequests::default();
        let sessions = Sessions::default();
        let control = RunControl::default();
        let session_id = format!("sess_{}", Uuid::now_v7().simple());
        sessions.lock().unwrap().insert(
            session_id.clone(),
            AcpSession {
                metadata: PersistedAcpSession {
                    schema_version: SCHEMA_VERSION.to_owned(),
                    session_id: session_id.clone(),
                    workspace: PathBuf::from("C:/workspace"),
                    runtime_session_ids: Vec::new(),
                    mcp_servers: Vec::new(),
                    tool_authority_hash: None,
                },
                active: Some(ActivePrompt {
                    control: control.clone(),
                    task: None,
                }),
                mcp: AcpMcpRuntime::empty(),
            },
        );

        cancel_prompt(&sessions, &session_id, &output, &pending).unwrap();
        assert!(control.is_cancelled());
        assert!(pending.lock().unwrap().is_empty());

        let approval = AcpApproval {
            session_id,
            output,
            pending: pending.clone(),
            control,
        };
        let request = PolicyRequest {
            tool: "patch_file".to_owned(),
            action_class: ActionClass::WorkspaceWrite,
            target: Some(PathBuf::from("C:/workspace/file.rs")),
            executable: None,
            interactive: true,
        };
        let approved = tokio::time::timeout(
            Duration::from_millis(100),
            approval.approve(&request, "workspace is not trusted"),
        )
        .await
        .expect("cancelled approval must not wait for a client reply");

        assert!(!approved);
        assert!(pending.lock().unwrap().is_empty());
        assert!(messages.try_recv().is_err());
    }

    #[test]
    fn cancellation_marks_control_before_releasing_pending_permission() {
        let (output, _messages) = mpsc::unbounded_channel();
        let pending = PendingClientRequests::default();
        let sessions = Sessions::default();
        let control = RunControl::default();
        let session_id = format!("sess_{}", Uuid::now_v7().simple());
        let (reply, response) = oneshot::channel();
        pending.lock().unwrap().insert(
            "permission-1".to_owned(),
            PendingClientRequest {
                session_id: session_id.clone(),
                response: reply,
            },
        );
        sessions.lock().unwrap().insert(
            session_id.clone(),
            AcpSession {
                metadata: PersistedAcpSession {
                    schema_version: SCHEMA_VERSION.to_owned(),
                    session_id: session_id.clone(),
                    workspace: PathBuf::from("C:/workspace"),
                    runtime_session_ids: Vec::new(),
                    mcp_servers: Vec::new(),
                    tool_authority_hash: None,
                },
                active: Some(ActivePrompt {
                    control: control.clone(),
                    task: None,
                }),
                mcp: AcpMcpRuntime::empty(),
            },
        );

        let pending_guard = pending.lock().unwrap();
        let cancellation = thread::spawn({
            let sessions = sessions.clone();
            let pending = pending.clone();
            let output = output.clone();
            let session_id = session_id.clone();
            move || cancel_prompt(&sessions, &session_id, &output, &pending).unwrap()
        });
        let deadline = Instant::now() + Duration::from_secs(1);
        while !control.is_cancelled() && Instant::now() < deadline {
            thread::yield_now();
        }
        assert!(control.is_cancelled());
        drop(pending_guard);
        cancellation.join().unwrap();
        assert_eq!(
            response.blocking_recv().unwrap(),
            json!({"result": {"outcome": {"outcome": "cancelled"}}})
        );
    }

    #[tokio::test]
    async fn recoverable_provider_failure_continues_through_acp_with_the_same_runtime_session() {
        let workspace = tempdir().unwrap();
        let paths = RuntimePaths::for_workspace(workspace.path()).unwrap();
        paths.ensure_data_directories().unwrap();
        let store = SessionStore::open(paths.workspace_data.join("sessions.sqlite3")).unwrap();
        let engine = SessionEngine::new(
            store.clone(),
            CasStore::new(paths.workspace_data.join("cas")).unwrap(),
            ToolRegistry::with_builtins(),
        );
        let (output, _messages) = mpsc::unbounded_channel();
        let first_sink = AcpEventSink::new("sess_test".to_owned(), output.clone());
        assert!(
            engine
                .run_chain(
                    ProviderChain::single("recorded-model", Arc::new(FailingProvider)),
                    RunOptions::new(
                        workspace.path().to_path_buf(),
                        "recorded-model",
                        resumable_contract()
                    ),
                    &first_sink,
                )
                .await
                .is_err()
        );
        let first_events = first_sink.events().unwrap();
        let runtime_session_id = first_events[0].session_id;
        let metadata = PersistedAcpSession {
            schema_version: SCHEMA_VERSION.to_owned(),
            session_id: format!("sess_{}", Uuid::now_v7().simple()),
            workspace: workspace.path().canonicalize().unwrap(),
            runtime_session_ids: vec![runtime_session_id],
            mcp_servers: Vec::new(),
            tool_authority_hash: None,
        };
        assert_eq!(
            recoverable_runtime_session(&metadata).unwrap(),
            Some(runtime_session_id)
        );
        assert!(load_finalized_runs(&metadata).unwrap().is_empty());

        let resumed_sink = AcpEventSink::new(metadata.session_id.clone(), output);
        let mut options = RunOptions::new(
            workspace.path().to_path_buf(),
            "recorded-model",
            resumable_contract(),
        );
        options.resume_session_id = Some(runtime_session_id);
        let resumed = engine
            .run_chain(
                ProviderChain::single("recorded-model", Arc::new(HealthyProvider)),
                options,
                &resumed_sink,
            )
            .await
            .unwrap();
        assert_eq!(resumed.session_id, runtime_session_id);
        let stored = store.load_session(runtime_session_id).unwrap();
        assert!(stored.checkpoint.is_none());
        let mut acp_events = first_events;
        acp_events.extend(resumed_sink.events().unwrap());
        assert_eq!(
            acp_events, stored.events,
            "ACP must capture the exact canonical envelopes serialized by JSONL"
        );
    }

    #[test]
    fn runtime_only_prompt_contract_is_visibly_non_task_scoped() {
        let contract = prompt_contract(&json!({
            "prompt": [{"type": "text", "text": "do the work"}]
        }))
        .unwrap();
        assert_eq!(contract.claim_scope, proofborne_core::ClaimScope::Runtime);
        assert_eq!(contract.criteria[0].id, "runtime_completed");
    }
}