bitrouter 1.0.0-alpha.31

//! `bro acp` subcommands — headless ACP session surface.
//!
//! Two entry points:
//!
//! - [`serve`] — launch a session and expose it as a vanilla ACP Agent over
//!   **stdio** until the ACP client disconnects. Used by GUIs and orchestrating
//!   agents that speak ACP.
//!
//! - [`prompt`] — launch a session, subscribe to updates, send one prompt, and
//!   stream each event as a self-describing **NDJSON** line to `out`. Exits
//!   after the prompt resolves (or immediately after submission when `no_wait`
//!   is true).
//!
//! ## Output formats
//!
//! `--format ndjson` (the default; `json` remains an alias) is NDJSON. Update lines carry the
//! [`SessionUpdateKind`] directly — the `type` tag value is the snake_case
//! variant name (`message_chunk`, `thought_chunk`, `tool_call`,
//! `tool_call_update`). A permission the headless policy answered is one line:
//!
//! ```json
//! {"type":"permission","decision":"denied","title":"Write src/main.rs","kind":"edit"}
//! ```
//!
//! The terminal result line is:
//!
//! ```json
//! {"type":"result","stop_reason":"end_turn"}
//! ```
//!
//! The `stop_reason` value is the ACP wire form (snake_case, via serde) so a
//! downstream parser sees the same spelling the protocol uses.
//!
//! In `--no-wait` mode the only line emitted is:
//!
//! ```json
//! {"type":"submitted"}
//! ```
//!
//! `--format text` prints the transcript exactly as piped `bro chat <agent>` prints
//! it to a pipe; `--format quiet` prints the assistant's text and nothing else.
//!
//! ## Permissions
//!
//! Nobody is at a headless terminal to broker a permission, so the caller
//! states the rule up front: `--deny-all` (the default), `--approve-reads`,
//! `--approve-all`, and a per-tool `--permission-policy`. The decision is made
//! by the same [`bitrouter_tui::permission::Policy`] the piped `chat` runs
//! under and reaches the agent through the same wire as a keystroke would. A
//! run that denied at least one request and approved none exits 5.
//!
//! Every entry point here — `serve`, `prompt`, `commands`, and piped `chat` —
//! loads its `Config` via the standard resolution chain (see
//! `bitrouter::paths`) and launches the agent named under `config.agents`.

use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;

use anyhow::{Context, Result};
use bitrouter_sdk::acp::transport::{AcpAgentConfig, AcpTransport};
use bitrouter_sdk::config::Config;
use bitrouter_sdk::invocation;
use futures::{FutureExt, StreamExt};
use serde::Serialize;
use tokio::io::{AsyncWrite, AsyncWriteExt};
use tokio_util::sync::CancellationToken;

use agent_client_protocol::schema::v1::{Cost, LlmProtocol, SessionUpdate};
use bitrouter_sdk::acp::client::{AcpClient, ClientOptions, PendingPermission};
use bitrouter_sdk::acp::controller::{
    RouteControl as AcpRouteControl, RouteControlError, RouteControlState,
    SessionCost as AcpSessionCost,
};
use bitrouter_sdk::acp::translate::{SessionUpdateKind, translate};
use bitrouter_tui::permission::{Decision, Mode, Policy, Prompt as PermissionPrompt};

use crate::chat::effects::Wire;
use crate::paths::ConfigSource;

// ── routing (spawn --via-daemon by default) ─────────────────────────────────────

/// Per-invocation routing decision for a spawned sub-agent. Routing is on by
/// default; `direct` opts out. See `docs/SPAWN_SPEC.md` §5.
#[derive(clap::Args, Debug, Clone, Default, PartialEq, Eq)]
pub struct RoutingOptions {
    /// Do NOT route this session's LLM traffic through the daemon — let the
    /// harness use its own provider auth. Routing is attempted by default
    /// when the harness supports headless redirection.
    #[arg(long)]
    pub direct: bool,
    /// Override the gateway base URL (else derived from `server.listen`).
    #[arg(long)]
    pub base_url: Option<String>,
    /// Pin the harness's model (via its model env var / `-c model=`).
    #[arg(long)]
    pub model: Option<String>,
    /// Never auto-start a local daemon when none is running — fail fast.
    #[arg(long)]
    pub no_start: bool,
}

/// Resolve a user-facing ACP agent name. Configured ids are authoritative;
/// friendly built-in aliases are considered only when no exact config entry
/// shadows them.
pub fn resolve_agent_id(config: &Config, requested: &str) -> Result<String> {
    if config.agents.contains_key(requested) {
        return Ok(requested.to_string());
    }
    let canonical = match requested {
        "claude" | "claude-code" => "claude-acp",
        "codex" => "codex-acp",
        other => other,
    };
    if config.agents.contains_key(canonical)
        || crate::harness::by_id(canonical).is_some_and(|harness| harness.acp_command.is_some())
    {
        Ok(canonical.to_string())
    } else {
        Err(anyhow::anyhow!(
            "'{requested}' is neither a configured ACP agent nor an unambiguous catalog alias"
        ))
    }
}

#[cfg(test)]
mod agent_resolution_tests {
    use bitrouter_sdk::acp::transport::{AcpAgentConfig, AcpTransport};
    use bitrouter_sdk::config::Config;

    use super::resolve_agent_id;

    #[test]
    fn configured_name_wins_over_builtin_alias() -> anyhow::Result<()> {
        let mut config = Config::default();
        config.agents.insert(
            "codex".to_string(),
            AcpAgentConfig {
                name: "codex".to_string(),
                transport: AcpTransport::Stdio {
                    command: "custom-codex".to_string(),
                    args: Vec::new(),
                    env: Default::default(),
                },
            },
        );
        assert_eq!(resolve_agent_id(&config, "codex")?, "codex");
        Ok(())
    }

    #[test]
    fn friendly_alias_resolves_to_acp_adapter() -> anyhow::Result<()> {
        assert_eq!(
            resolve_agent_id(&Config::default(), "claude")?,
            "claude-acp"
        );
        assert_eq!(resolve_agent_id(&Config::default(), "codex")?, "codex-acp");
        Ok(())
    }
}

#[cfg(all(test, unix))]
mod session_host_tests {
    use std::collections::HashMap;
    use std::path::Path;
    use std::time::Duration;

    use futures::StreamExt;
    use tokio_util::sync::CancellationToken;

    use super::{
        RoutingOptions, SessionHost, SessionSelection, SpawnContext, is_lifecycle_cancelled,
        launch_options,
    };
    use bitrouter_sdk::acp::transport::{AcpAgentConfig, AcpTransport};
    use bitrouter_sdk::config::Config;

    const STUB: &str = r#"
while read line; do
  id=$(printf '%s' "$line" | sed -n 's/.*"id":"\([^"]*\)".*/\1/p')
  case "$line" in
    *initialize*) printf '{"jsonrpc":"2.0","id":"%s","result":{"protocolVersion":1,"agentCapabilities":{"loadSession":true,"sessionCapabilities":{"list":{},"resume":{},"close":{}}}}}\n' "$id";;
    *session/new*) printf '{"jsonrpc":"2.0","id":"%s","result":{"sessionId":"native-1"}}\n' "$id";;
    *session/load*) printf '{"jsonrpc":"2.0","id":"%s","error":{"code":-32000,"message":"native load failed"}}\n' "$id";;
    *session/prompt*) printf '{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"native-1","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"hello"}}}}\n';
                      printf '{"jsonrpc":"2.0","id":"%s","result":{"stopReason":"end_turn"}}\n' "$id";;
  esac
done
"#;

    const HANG_INITIALIZE_STUB: &str = r#"
while read line; do
  case "$line" in
    *initialize*) printf 'initialize\n' > "$READY_PATH";;
  esac
done
"#;

    const HANG_NEW_STUB: &str = r#"
while read line; do
  id=$(printf '%s' "$line" | sed -n 's/.*"id":"\([^"]*\)".*/\1/p')
  case "$line" in
    *initialize*) printf '{"jsonrpc":"2.0","id":"%s","result":{"protocolVersion":1,"agentCapabilities":{}}}\n' "$id";;
    *session/new*) printf 'session-new\n' > "$READY_PATH";;
  esac
done
"#;

    const HANG_LOAD_STUB: &str = r#"
while read line; do
  id=$(printf '%s' "$line" | sed -n 's/.*"id":"\([^"]*\)".*/\1/p')
  case "$line" in
    *initialize*) printf '{"jsonrpc":"2.0","id":"%s","result":{"protocolVersion":1,"agentCapabilities":{"loadSession":true}}}\n' "$id";;
    *session/new*) printf '{"jsonrpc":"2.0","id":"%s","result":{"sessionId":"native-1"}}\n' "$id";;
    *session/load*) printf 'session-load\n' > "$READY_PATH";;
    *session/prompt*) printf '{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"native-1","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"hello"}}}}\n';
                      printf '{"jsonrpc":"2.0","id":"%s","result":{"stopReason":"end_turn"}}\n' "$id";;
  esac
done
"#;

    async fn wait_for_ready(path: &Path) -> anyhow::Result<()> {
        tokio::time::timeout(Duration::from_secs(5), async {
            while !path.exists() {
                tokio::time::sleep(Duration::from_millis(5)).await;
            }
        })
        .await
        .map_err(|_| anyhow::anyhow!("stub did not reach its expected lifecycle phase"))?;
        Ok(())
    }

    async fn host_with_stub(
        source: &crate::paths::ConfigSource,
        script: &str,
        ready_path: &Path,
    ) -> anyhow::Result<SessionHost> {
        let mut config = Config::default();
        let mut env = HashMap::new();
        env.insert(
            "READY_PATH".to_string(),
            ready_path.to_string_lossy().into_owned(),
        );
        config.agents.insert(
            "stub".to_string(),
            AcpAgentConfig {
                name: "stub".to_string(),
                transport: AcpTransport::Stdio {
                    command: "bash".to_string(),
                    args: vec!["-c".to_string(), script.to_string()],
                    env,
                },
            },
        );
        SessionHost::prepare(
            SpawnContext {
                source,
                config,
                agent_id: "stub",
                options: launch_options(None),
                routing: RoutingOptions {
                    direct: true,
                    ..RoutingOptions::default()
                },
            },
            false,
        )
        .await
    }

    #[tokio::test]
    async fn full_screen_host_returns_routing_notes_to_its_presenter() -> anyhow::Result<()> {
        let directory = tempfile::tempdir()?;
        let source = crate::paths::ConfigSource::Default {
            home: directory.path().to_path_buf(),
        };
        let mut config = Config::default();
        config.agents.insert(
            "stub".to_string(),
            AcpAgentConfig {
                name: "stub".to_string(),
                transport: AcpTransport::Stdio {
                    command: "bash".to_string(),
                    args: vec!["-c".to_string(), STUB.to_string()],
                    env: HashMap::new(),
                },
            },
        );
        let mut diagnostics = Vec::new();
        let host = SessionHost::prepare_with_diagnostics(
            SpawnContext {
                source: &source,
                config,
                agent_id: "stub",
                options: launch_options(None),
                routing: RoutingOptions {
                    direct: true,
                    model: Some("ignored-model".to_string()),
                    ..RoutingOptions::default()
                },
            },
            false,
            &mut |message| diagnostics.push(message),
        )
        .await?;

        assert_eq!(
            diagnostics,
            ["note: --model 'ignored-model' ignored — running --direct"]
        );
        assert_eq!(host.agent_id, "stub");
        Ok(())
    }

    #[tokio::test]
    async fn shared_host_negotiates_lifecycle_and_drives_one_turn() -> anyhow::Result<()> {
        let directory = tempfile::tempdir()?;
        let source = crate::paths::ConfigSource::Default {
            home: directory.path().to_path_buf(),
        };
        let mut config = Config::default();
        config.agents.insert(
            "stub".to_string(),
            AcpAgentConfig {
                name: "stub".to_string(),
                transport: AcpTransport::Stdio {
                    command: "bash".to_string(),
                    args: vec!["-c".to_string(), STUB.to_string()],
                    env: HashMap::new(),
                },
            },
        );
        let host = SessionHost::prepare(
            SpawnContext {
                source: &source,
                config,
                agent_id: "stub",
                options: launch_options(None),
                routing: RoutingOptions {
                    direct: true,
                    ..RoutingOptions::default()
                },
            },
            true,
        )
        .await?;
        let mut handle = host
            .open(&SessionSelection::New, std::env::current_dir()?)
            .await?;
        assert_eq!(handle.session_id, "native-1");
        assert!(handle.capabilities.load);
        assert!(handle.capabilities.list);
        assert!(handle.capabilities.resume);
        assert!(handle.capabilities.close);
        assert!(!handle.capabilities.delete);
        let mut updates = handle.take_updates();
        let response = handle.client.prompt(&handle.session_id, "hi").await?;
        assert_eq!(format!("{:?}", response.stop_reason), "EndTurn");
        let update = tokio::time::timeout(std::time::Duration::from_secs(2), updates.next())
            .await?
            .ok_or_else(|| anyhow::anyhow!("stub emitted no update"))?;
        assert!(matches!(
            update,
            agent_client_protocol::schema::v1::SessionUpdate::AgentMessageChunk(_)
        ));
        assert!(handle.shutdown().await);
        Ok(())
    }

    #[tokio::test]
    async fn failed_load_retains_the_active_session_identity_and_stream() -> anyhow::Result<()> {
        let directory = tempfile::tempdir()?;
        let source = crate::paths::ConfigSource::Default {
            home: directory.path().to_path_buf(),
        };
        let mut config = Config::default();
        config.agents.insert(
            "stub".to_string(),
            AcpAgentConfig {
                name: "stub".to_string(),
                transport: AcpTransport::Stdio {
                    command: "bash".to_string(),
                    args: vec!["-c".to_string(), STUB.to_string()],
                    env: HashMap::new(),
                },
            },
        );
        let host = SessionHost::prepare(
            SpawnContext {
                source: &source,
                config,
                agent_id: "stub",
                options: launch_options(None),
                routing: RoutingOptions {
                    direct: true,
                    ..RoutingOptions::default()
                },
            },
            false,
        )
        .await?;
        let mut handle = host
            .open(&SessionSelection::New, std::env::current_dir()?)
            .await?;
        let active_id = handle.session_id.clone();
        let active_agent_id = handle.agent_session_id.clone();
        let mut updates = handle.take_updates();

        let load = handle
            .select_with_cancel(
                &SessionSelection::Load("missing".to_string()),
                &CancellationToken::new(),
            )
            .await;
        assert!(load.is_err());
        assert_eq!(handle.session_id, active_id);
        assert_eq!(handle.agent_session_id, active_agent_id);

        let response = handle.client.prompt(&handle.session_id, "hi").await?;
        assert_eq!(format!("{:?}", response.stop_reason), "EndTurn");
        let update = tokio::time::timeout(std::time::Duration::from_secs(2), updates.next())
            .await?
            .ok_or_else(|| anyhow::anyhow!("active session emitted no update after failed load"))?;
        assert!(matches!(
            update,
            agent_client_protocol::schema::v1::SessionUpdate::AgentMessageChunk(_)
        ));
        assert!(handle.shutdown().await);
        Ok(())
    }

    #[tokio::test]
    async fn cancelled_initialize_reaps_the_owned_controller() -> anyhow::Result<()> {
        let directory = tempfile::tempdir()?;
        let source = crate::paths::ConfigSource::Default {
            home: directory.path().to_path_buf(),
        };
        let ready = directory.path().join("initialize-ready");
        let host = host_with_stub(&source, HANG_INITIALIZE_STUB, &ready).await?;
        let cancel = CancellationToken::new();
        let opening =
            host.open_with_cancel(&SessionSelection::New, std::env::current_dir()?, &cancel);
        tokio::pin!(opening);
        tokio::select! {
            result = &mut opening => {
                return if result.is_ok() {
                    Err(anyhow::anyhow!("unresponsive initialize unexpectedly opened a session"))
                } else {
                    Err(anyhow::anyhow!("unresponsive initialize failed before cancellation"))
                };
            }
            ready = wait_for_ready(&ready) => ready?,
        }
        cancel.cancel();
        let result = tokio::time::timeout(Duration::from_secs(20), &mut opening)
            .await
            .map_err(|_| anyhow::anyhow!("cancelled initialize did not settle"))?;
        let error = result.err().ok_or_else(|| {
            anyhow::anyhow!("cancelled initialize unexpectedly returned a session")
        })?;
        assert!(is_lifecycle_cancelled(&error));
        Ok(())
    }

    #[tokio::test]
    async fn cancelled_new_session_reaps_the_owned_controller() -> anyhow::Result<()> {
        let directory = tempfile::tempdir()?;
        let source = crate::paths::ConfigSource::Default {
            home: directory.path().to_path_buf(),
        };
        let ready = directory.path().join("new-ready");
        let host = host_with_stub(&source, HANG_NEW_STUB, &ready).await?;
        let cancel = CancellationToken::new();
        let opening =
            host.open_with_cancel(&SessionSelection::New, std::env::current_dir()?, &cancel);
        tokio::pin!(opening);
        tokio::select! {
            result = &mut opening => {
                return if result.is_ok() {
                    Err(anyhow::anyhow!("unresponsive session/new unexpectedly opened a session"))
                } else {
                    Err(anyhow::anyhow!("unresponsive session/new failed before cancellation"))
                };
            }
            ready = wait_for_ready(&ready) => ready?,
        }
        cancel.cancel();
        let result = tokio::time::timeout(Duration::from_secs(20), &mut opening)
            .await
            .map_err(|_| anyhow::anyhow!("cancelled session/new did not settle"))?;
        let error = result.err().ok_or_else(|| {
            anyhow::anyhow!("cancelled session/new unexpectedly returned a session")
        })?;
        assert!(is_lifecycle_cancelled(&error));
        Ok(())
    }

    #[tokio::test]
    async fn cancelled_load_keeps_the_active_session_and_stream() -> anyhow::Result<()> {
        let directory = tempfile::tempdir()?;
        let source = crate::paths::ConfigSource::Default {
            home: directory.path().to_path_buf(),
        };
        let ready = directory.path().join("load-ready");
        let host = host_with_stub(&source, HANG_LOAD_STUB, &ready).await?;
        let mut handle = host
            .open(&SessionSelection::New, std::env::current_dir()?)
            .await?;
        let active_id = handle.session_id.clone();
        let active_agent_id = handle.agent_session_id.clone();
        let mut updates = handle.take_updates();
        let cancel = CancellationToken::new();

        {
            let selection = SessionSelection::Load("native-other".to_string());
            let loading = handle.select_with_cancel(&selection, &cancel);
            tokio::pin!(loading);
            tokio::select! {
                result = &mut loading => {
                    return if result.is_ok() {
                        Err(anyhow::anyhow!("unresponsive session/load unexpectedly completed"))
                    } else {
                        Err(anyhow::anyhow!("unresponsive session/load failed before cancellation"))
                    };
                }
                ready = wait_for_ready(&ready) => ready?,
            }
            cancel.cancel();
            let result = tokio::time::timeout(Duration::from_secs(5), &mut loading)
                .await
                .map_err(|_| anyhow::anyhow!("cancelled session/load did not settle"))?;
            let error = result
                .err()
                .ok_or_else(|| anyhow::anyhow!("cancelled session/load unexpectedly completed"))?;
            assert!(is_lifecycle_cancelled(&error));
        }

        assert_eq!(handle.session_id, active_id);
        assert_eq!(handle.agent_session_id, active_agent_id);
        let response = handle.client.prompt(&handle.session_id, "hi").await?;
        assert_eq!(format!("{:?}", response.stop_reason), "EndTurn");
        let update = tokio::time::timeout(Duration::from_secs(2), updates.next())
            .await?
            .ok_or_else(|| anyhow::anyhow!("active session stream ended after cancelled load"))?;
        assert!(matches!(
            update,
            agent_client_protocol::schema::v1::SessionUpdate::AgentMessageChunk(_)
        ));
        assert!(handle.shutdown().await);
        Ok(())
    }
}

// ── headless presentation and permissions ────────────────────────────────────

/// What a headless `prompt` prints.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, clap::ValueEnum)]
pub enum PromptFormat {
    /// One self-describing JSON object per line.
    #[default]
    #[value(alias = "json")]
    Ndjson,
    /// The transcript as the piped `bro chat <agent>` compatibility path prints it.
    Text,
    /// The assistant's text and nothing else.
    Quiet,
}

/// The flags a headless caller states instead of sitting at the terminal:
/// how permission requests are answered, and what is printed.
#[derive(clap::Args, Debug, Clone, Default, PartialEq, Eq)]
pub struct HeadlessOptions {
    /// Approve every permission request the harness makes.
    #[arg(long, conflicts_with_all = ["approve_reads", "deny_all"])]
    pub approve_all: bool,
    /// Approve tool calls the harness labels `read` or `search` (the ACP tool
    /// kind); deny everything else, including unlabelled calls.
    #[arg(long, conflicts_with_all = ["approve_all", "deny_all"])]
    pub approve_reads: bool,
    /// Deny every permission request. This is the default.
    #[arg(long, conflicts_with_all = ["approve_all", "approve_reads"])]
    pub deny_all: bool,
    /// Per-tool policy, inline JSON or `@path`:
    /// `{"autoApprove":[…],"autoDeny":[…],"defaultAction":"approve"|"deny"}`.
    /// Entries match the ACP tool kind (`read`, `edit`, `execute`, …), the
    /// tool-call title, or its first word. `autoDeny` wins over `autoApprove`;
    /// an unmatched request uses `defaultAction`, else the mode flag.
    #[arg(long, value_name = "JSON|@PATH")]
    pub permission_policy: Option<String>,
    /// Output: `ndjson` (the default), `text` (the transcript as
    /// `code <agent>` prints it to a pipe), or `quiet` (assistant text only).
    #[arg(long, value_enum, default_value_t)]
    pub format: PromptFormat,
}

/// The per-tool half of [`HeadlessOptions::permission_policy`], as written.
///
/// `escalate` — acpx's fourth action — is deliberately not a field: nobody is
/// at this terminal to escalate to, and accepting it would silently mean
/// `deny`. `deny_unknown_fields` is what turns it into a parse error instead.
#[derive(serde::Deserialize, Default)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct PolicyFile {
    #[serde(default)]
    auto_approve: Vec<String>,
    #[serde(default)]
    auto_deny: Vec<String>,
    default_action: Option<PolicyAction>,
}

#[derive(serde::Deserialize, Clone, Copy)]
#[serde(rename_all = "lowercase")]
enum PolicyAction {
    Approve,
    Deny,
}

impl HeadlessOptions {
    /// The policy these flags describe. A malformed `--permission-policy`
    /// fails here, before any session side effect.
    pub fn policy(&self) -> Result<Policy> {
        let mode = if self.approve_all {
            Mode::ApproveAll
        } else if self.approve_reads {
            Mode::ApproveReads
        } else {
            Mode::DenyAll
        };
        let file = match self.permission_policy.as_deref() {
            None => PolicyFile::default(),
            Some(flag) => {
                let text = match flag.strip_prefix('@') {
                    Some(path) => std::fs::read_to_string(path)
                        .with_context(|| format!("reading --permission-policy from {path}"))?,
                    None => flag.to_string(),
                };
                serde_json::from_str(&text).context("--permission-policy is not a valid policy")?
            }
        };
        Ok(Policy {
            mode,
            auto_approve: file.auto_approve,
            auto_deny: file.auto_deny,
            default_action: file.default_action.map(|action| match action {
                PolicyAction::Approve => Decision::Approve,
                PolicyAction::Deny => Decision::Deny,
            }),
        })
    }
}

/// Everything one headless [`prompt`] is told beyond its text.
#[derive(Default)]
pub struct PromptOptions {
    /// Return as soon as the prompt is submitted.
    pub no_wait: bool,
    /// The `--result-schema` contract, when there is one.
    pub contract: Option<crate::result_contract::ResultContract>,
    /// How permission requests are answered.
    pub policy: Policy,
    /// What is printed.
    pub format: PromptFormat,
    /// Create a new session, load one with replay, or resume one without replay.
    pub session: SessionSelection,
    /// Working directory supplied to the native session lifecycle request.
    pub cwd: Option<PathBuf>,
}

/// Which harness-native session a one-shot driver opens.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub enum SessionSelection {
    /// Create a new harness-native session.
    #[default]
    New,
    /// Load a session and replay its history.
    Load(String),
    /// Resume a session without replaying its history.
    Resume(String),
}

/// Negotiated facts shared by every BitRouter-owned ACP client surface.
///
/// Drivers render or reject controls from this value; they do not infer
/// support from an agent name or optimistically send draft methods.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct CapabilitySnapshot {
    pub load: bool,
    pub resume: bool,
    pub list: bool,
    pub close: bool,
    pub delete: bool,
    pub fork: bool,
    pub mcp_stdio: bool,
    pub mcp_http: bool,
    pub mcp_sse: bool,
    pub authentication_methods: Vec<String>,
    pub terminal_auth: bool,
    pub route_list: bool,
    pub route_set: bool,
    pub route_reset: bool,
}

impl CapabilitySnapshot {
    fn from_client(client: &AcpClient, terminal_auth: bool) -> Self {
        let agent = client.agent_capabilities();
        let sessions = &agent.session_capabilities;
        let route = client.route_control();
        Self {
            load: agent.load_session,
            resume: sessions.resume.is_some(),
            list: sessions.list.is_some(),
            close: sessions.close.is_some(),
            delete: sessions.delete.is_some(),
            fork: sessions.fork.is_some(),
            // ACP stdio MCP descriptors are the baseline representation. HTTP
            // and SSE must be explicitly negotiated.
            mcp_stdio: true,
            mcp_http: agent.mcp_capabilities.http,
            mcp_sse: agent.mcp_capabilities.sse,
            authentication_methods: client
                .auth_methods()
                .iter()
                .map(|method| method.id().0.to_string())
                .collect(),
            terminal_auth,
            route_list: route.allows(bitrouter_sdk::acp::client::RouteMethod::List),
            route_set: route.allows(bitrouter_sdk::acp::client::RouteMethod::Set),
            route_reset: route.allows(bitrouter_sdk::acp::client::RouteMethod::Reset),
        }
    }

    /// Compact lifecycle explanation used by the Code TUI.
    pub fn lifecycle_summary(&self) -> String {
        let mut supported = vec!["new"];
        for (name, enabled) in [
            ("list", self.list),
            ("load", self.load),
            ("resume", self.resume),
            ("close", self.close),
            ("fork*", self.fork),
            ("delete*", self.delete),
        ] {
            if enabled {
                supported.push(name);
            }
        }
        supported.join(", ")
    }
}

/// What the headless policy decided over one run, for the exit status.
///
/// Only decisions the policy made are counted. A request still outstanding at
/// teardown is denied by the client's own ledger (invariant I1) and is not
/// tallied, because no policy answered it.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct PermissionTally {
    approved: usize,
    denied: usize,
}

impl PermissionTally {
    fn record(&mut self, decision: Decision) {
        match decision {
            Decision::Approve => self.approved += 1,
            Decision::Deny => self.denied += 1,
        }
    }

    /// The process exit status this tally calls for: 5 when at least one
    /// request was denied and none approved (acpx's reading of "the agent was
    /// refused"), else 0.
    pub fn exit_code(&self) -> i32 {
        if self.denied > 0 && self.approved == 0 {
            5
        } else {
            0
        }
    }
}

/// The inputs shared by the two sub-agent launch paths ([`serve`] and
/// [`prompt`]): where the config came from, the loaded config, which agent,
/// the session options, and the routing decision. Bundled so each entry point
/// keeps a small, readable signature.
pub struct SpawnContext<'a> {
    /// Where the config was resolved from (daemon socket / auto-start).
    pub source: &'a ConfigSource,
    /// The loaded config (routing overlays its agent entry in place).
    pub config: Config,
    /// The agent id to launch (catalog id or configured entry).
    pub agent_id: &'a str,
    /// Session options (turn timeout).
    pub options: LaunchOptions,
    /// The routing decision (via-daemon by default, or `--direct`).
    pub routing: RoutingOptions,
}

/// A fail-fast routing failure, surfaced BEFORE any session side effect
/// (`docs/SPAWN_SPEC.md` §8). Rendered as a structured NDJSON `error` line in
/// `prompt` mode, or to stderr in `serve` mode.
#[derive(Debug)]
pub enum RoutingError {
    /// The daemon behind `via` did not answer `/health` after auto-start.
    DaemonUnreachable {
        /// The gateway base URL that was probed.
        via: String,
    },
    /// The daemon requires auth and no `BITROUTER_API_KEY` is available.
    AuthRequired {
        /// The gateway base URL that would have been used.
        via: String,
    },
    /// The pinned adapter fallback could not be rendered before launch.
    EndpointConfiguration {
        /// The gateway base URL that would have been used.
        via: String,
        /// Sanitized configuration failure. Credentials are never included.
        message: String,
    },
}

impl std::fmt::Display for RoutingError {
    /// Message *and* hint, because this is what a caller renders when the
    /// failure leaves the process as an error value rather than as an
    /// already-printed line. The `ndjson` form keeps them as separate fields
    /// and is unaffected.
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(formatter, "{}\n  hint: {}", self.message(), self.hint())
    }
}

impl std::error::Error for RoutingError {}

impl RoutingError {
    /// Machine-readable `code` for the NDJSON `error` line.
    fn code(&self) -> &'static str {
        match self {
            RoutingError::DaemonUnreachable { .. } => "daemon_unreachable",
            RoutingError::AuthRequired { .. } => "auth_required",
            RoutingError::EndpointConfiguration { .. } => "endpoint_configuration",
        }
    }

    /// The gateway base URL this failure concerns.
    fn via(&self) -> &str {
        match self {
            RoutingError::DaemonUnreachable { via }
            | RoutingError::AuthRequired { via }
            | RoutingError::EndpointConfiguration { via, .. } => via,
        }
    }

    /// One-line remediation hint.
    fn hint(&self) -> String {
        match self {
            RoutingError::DaemonUnreachable { .. } => {
                format!("run `{} start`, or pass --direct", invocation::name())
            }
            RoutingError::AuthRequired { .. } => {
                "export BITROUTER_API_KEY (or create a key), or pass --direct".to_string()
            }
            RoutingError::EndpointConfiguration { .. } => {
                "check the pinned ACP adapter configuration, or pass --direct".to_string()
            }
        }
    }

    /// Human message for stderr (`serve`) and the NDJSON `message` field.
    fn message(&self) -> String {
        match self {
            RoutingError::DaemonUnreachable { via } => {
                format!("BitRouter daemon unreachable at {via}")
            }
            RoutingError::AuthRequired { via } => {
                format!("daemon at {via} requires auth but no BITROUTER_API_KEY is set")
            }
            RoutingError::EndpointConfiguration { message, .. } => message.clone(),
        }
    }

    /// The structured NDJSON `error` line for this failure.
    fn ndjson(&self) -> serde_json::Value {
        serde_json::json!({
            "type": "error",
            "code": self.code(),
            "via": self.via(),
            "hint": self.hint(),
            "message": self.message(),
        })
    }
}

/// What routing decided for one session.
#[derive(Debug, Clone, Default)]
pub struct Routed {
    /// The base URL the session's LLM traffic goes through, or `None` when it
    /// runs direct.
    pub via: Option<String>,
    /// The per-session attribution token, when this session's traffic can be
    /// told apart from every other caller's.
    ///
    /// `None` when the caller supplied their own credential: that is real
    /// authentication, and rewriting it to attach attribution would break
    /// `skip_auth: false` — the same rule `spawn.rs` follows. A session
    /// without one has no per-launch attribution; on the controlled paths a
    /// claimed controller namespace attributes by controller instance instead.
    pub launch_id: Option<String>,
    /// One controller process / harness connection correlation id. This is
    /// never an ACP session id and is absent for direct or legacy harnesses.
    pub controller_instance_id: Option<String>,
    /// Opaque route namespace derived from the ordinary API credential, or
    /// `local` under `skip_auth`. It is not a bearer credential.
    pub api_principal: Option<String>,
    /// The one process-scoped endpoint plan used for both launch fallback and
    /// post-initialize ACP provider configuration.
    pub endpoint_plan: Option<crate::harness::HarnessEndpointPlan>,
}

/// Resolve routing and overlay it onto `config`'s entry for `agent_id`,
/// inserting the bundled-catalog invocation when the id is catalog-known but
/// unconfigured (so `bro spawn claude-acp` works with no YAML edit).
///
/// Returns the "via" base URL when routing is active, or `None` when the
/// session runs direct (`--direct`, an unknown/custom agent, or an
/// unroutable harness — each warned to stderr). Fails fast — before the
/// caller spawns any agent process — on an unreachable daemon or a missing
/// required credential.
pub async fn apply_routing(
    source: &ConfigSource,
    config: &mut Config,
    agent_id: &str,
    opts: &RoutingOptions,
) -> std::result::Result<Routed, RoutingError> {
    let cloud_credentials = crate::cloud::StandaloneCloudCredentials::new();
    apply_routing_with_cloud_credentials(source, config, agent_id, opts, &cloud_credentials).await
}

async fn apply_routing_with_cloud_credentials(
    source: &ConfigSource,
    config: &mut Config,
    agent_id: &str,
    opts: &RoutingOptions,
    cloud_credentials: &crate::cloud::StandaloneCloudCredentials,
) -> std::result::Result<Routed, RoutingError> {
    let mut diagnostics = |message| eprintln!("{message}");
    apply_routing_with_diagnostics(
        source,
        config,
        agent_id,
        opts,
        cloud_credentials,
        &mut diagnostics,
    )
    .await
}

async fn apply_routing_with_diagnostics(
    source: &ConfigSource,
    config: &mut Config,
    agent_id: &str,
    opts: &RoutingOptions,
    cloud_credentials: &crate::cloud::StandaloneCloudCredentials,
    diagnostics: &mut dyn FnMut(String),
) -> std::result::Result<Routed, RoutingError> {
    // Every controlled entrypoint shares this preparation. Preserve the config
    // home's database location even when the coding working directory differs.
    config.database.url = crate::db::anchor_url(&config.database.url, source.home());
    // A catalog-known id needs no `agents:` entry — synthesize its invocation.
    if !config.agents.contains_key(agent_id)
        && let Some(h) = crate::harness::by_id(agent_id)
        // Interactive-only harnesses (grok, antigravity) have no ACP
        // adapter to synthesize — the id falls through to not-found.
        && let Some(command) = h.acp_command
    {
        config.agents.insert(
            agent_id.to_string(),
            AcpAgentConfig {
                name: agent_id.to_string(),
                transport: AcpTransport::Stdio {
                    command: command.to_string(),
                    args: h.acp_args.iter().map(|s| s.to_string()).collect(),
                    env: Default::default(),
                },
            },
        );
    }

    if let Some(agent) = config.agents.get_mut(agent_id) {
        crate::local_cli::apply(&mut agent.transport).await;
    }

    // `--model` only takes effect when the daemon route is applied; warn
    // rather than silently drop it on any path that launches direct.
    if opts.direct {
        warn_model_dropped(opts, "running --direct", diagnostics);
        return Ok(Routed::default());
    }

    // Match the (now-present-if-known) invocation back to a catalog harness.
    let harness = match config.agents.get(agent_id) {
        Some(entry) => {
            let AcpTransport::Stdio { command, args, .. } = &entry.transport;
            crate::harness::match_invocation(command, args)
        }
        // Unknown agent — let the caller's `Session::launch` surface the
        // configured-agents not-found error.
        None => return Ok(Routed::default()),
    };
    let Some(harness) = harness else {
        diagnostics(format!(
            "note: routing unavailable for '{agent_id}' (not catalog-matched); \
             launching direct — set its `env` to route manually"
        ));
        warn_model_dropped(opts, "the agent is not catalog-matched", diagnostics);
        return Ok(Routed::default());
    };
    if !harness.env_args_routable() {
        diagnostics(format!(
            "note: '{}' routes via synthesized config, which headless spawn doesn't do yet \
             in this ACP adapter; launching direct",
            harness.id
        ));
        warn_model_dropped(
            opts,
            "the harness routes only in the interactive facet",
            diagnostics,
        );
        return Ok(Routed::default());
    }
    let uses_maintained_adapter =
        config
            .agents
            .get(agent_id)
            .is_some_and(|entry| match &entry.transport {
                AcpTransport::Stdio { command, args, .. } => {
                    harness.uses_maintained_adapter(command, args)
                }
            });
    if harness.id == "codex-acp" && !uses_maintained_adapter {
        diagnostics(format!(
            "note: routing unavailable for '{agent_id}': Codex ACP endpoint configuration \
             requires @agentclientprotocol/codex-acp@1.10.0; launching direct"
        ));
        warn_model_dropped(
            opts,
            "the configured Codex ACP adapter is not the maintained pin",
            diagnostics,
        );
        return Ok(Routed::default());
    }

    // Base URL, auth mode, and whether the target is a remote we can't vouch for.
    let base_url = opts
        .base_url
        .clone()
        .unwrap_or_else(|| crate::spawn::derive_base_url(&config.server.listen));
    let target_authority = opts
        .base_url
        .as_deref()
        .and_then(crate::spawn::listen_from_base_url);
    let target_is_local = match &target_authority {
        Some(a) => crate::spawn::listen_is_local(a),
        None => crate::spawn::listen_is_local(&config.server.listen),
    };
    // A remote daemon's `skip_auth` is unknowable here, so require a key.
    let require_key = !target_is_local || !config.server.skip_auth;

    // A harness whose credential isn't Bearer (gemini's `x-goog-api-key`) is
    // rejected by the daemon's auth hook under `skip_auth: false` — warn
    // rather than let the session 401 mid-turn (SPAWN_SPEC §6.3).
    if require_key && !harness.auth_is_bearer() {
        diagnostics(format!(
            "warning: '{}' sends its API key as a non-Bearer header the daemon rejects under \
             auth mode (`skip_auth: false`) — this session will likely 401. Use `skip_auth: \
             true`, a `--direct` session, or a different harness.",
            harness.id
        ));
    }

    let explicit_key = crate::spawn::nonempty_env(crate::harness::BITROUTER_API_KEY_ENV);
    let stored_cloud_key = if explicit_key.is_none() && require_key && !target_is_local {
        cloud_credentials.routing_fallback(&base_url).await
    } else {
        None
    };
    // Mirrors `spawn.rs`: a user-supplied credential wins and is never
    // tagged, but a session that would otherwise send the placeholder gets a
    // freshly minted per-launch token instead. That is what makes this
    // session's spend separable from every other caller's — without it the
    // metering store has nothing to group by and cost can only be reported
    // daemon-wide.
    let supplied = explicit_key.or(stored_cloud_key);
    if supplied.is_none() && require_key {
        return Err(RoutingError::AuthRequired { via: base_url });
    }
    let auth = crate::spawn::resolve_launch_token(supplied, None);
    let launch_id = crate::spawn::is_launch_token(&auth).then(|| auth.clone());
    let api_principal = if target_is_local && config.server.skip_auth {
        "local".to_string()
    } else {
        crate::auth::keys::hash_key(&auth)
    };

    // Daemon liveness: auto-start a local daemon, then probe. Fail fast if the
    // daemon is still unreachable (a routed sub-agent without one is
    // guaranteed-dead) — before any session side effect.
    if opts.base_url.is_none() && target_is_local {
        crate::spawn::ensure_local_daemon(source, config, opts.no_start).await;
    }
    if !crate::spawn::base_url_reachable(&base_url).await {
        return Err(RoutingError::DaemonUnreachable { via: base_url });
    }

    // Compute one endpoint plan for maintained ACP adapters and render the
    // launch fallback from that same plan. The controller later applies the
    // returned plan through providers/set; no second route description is
    // constructed. Legacy harnesses retain their existing env/args overlay.
    let controller_instance_id = format!("brc_{}", uuid::Uuid::new_v4().simple());
    let endpoint_plan = if uses_maintained_adapter {
        harness.endpoint_plan(
            &base_url,
            &auth,
            opts.model.as_deref(),
            &controller_instance_id,
        )
    } else {
        None
    };
    let overlay = match &endpoint_plan {
        Some(plan) => {
            plan.fallback_overlay()
                .map_err(|error| RoutingError::EndpointConfiguration {
                    via: base_url.clone(),
                    message: format!("could not configure pinned '{}': {error}", harness.id),
                })?
        }
        None => harness.routing_overlay(&base_url, &auth, opts.model.as_deref()),
    };
    if let Some(entry) = config.agents.get_mut(agent_id) {
        let AcpTransport::Stdio { args, env, .. } = &mut entry.transport;
        for (k, v) in overlay.env {
            if let Some(existing) = env.get(&k)
                && existing != &v
            {
                diagnostics(format!(
                    "note: routing overrides your `env.{k}` for '{agent_id}' \
                     (pass --direct to keep your value)"
                ));
            }
            env.insert(k, v);
        }
        args.extend(overlay.args);
    }
    Ok(Routed {
        via: Some(base_url),
        launch_id,
        controller_instance_id: endpoint_plan.as_ref().map(|_| controller_instance_id),
        api_principal: endpoint_plan.as_ref().map(|_| api_principal),
        endpoint_plan,
    })
}

fn warn_model_dropped(opts: &RoutingOptions, why: &str, diagnostics: &mut dyn FnMut(String)) {
    if let Some(model) = &opts.model {
        diagnostics(format!("note: --model '{model}' ignored — {why}"));
    }
}

// ── NDJSON helpers ────────────────────────────────────────────────────────────

/// Terminal result line emitted after the prompt resolves.
///
/// Generic over the stop-reason type so the ACP `StopReason` (which derives
/// `serde::Serialize` with snake_case rename) renders its wire form directly —
/// `"end_turn"`, not the Rust `Debug` spelling `"EndTurn"`. Keeping it generic
/// also avoids naming `agent_client_protocol_schema` here (it isn't a direct
/// dependency of this crate).
#[derive(Serialize)]
struct ResultLine<S: Serialize> {
    #[serde(rename = "type")]
    kind: &'static str,
    stop_reason: S,
    /// Under `--result-schema`: the extracted, schema-valid result object —
    /// or JSON `null` when extraction/validation failed after the one repair
    /// re-prompt. Omitted entirely without the flag (byte-compatible).
    #[serde(skip_serializing_if = "Option::is_none")]
    result: Option<serde_json::Value>,
    /// Under `--result-schema`: whether `result` satisfied the schema.
    #[serde(skip_serializing_if = "Option::is_none")]
    schema_ok: Option<bool>,
    /// Under `--result-schema`, on failure only: the last reply's raw text so
    /// the orchestrator is never blocked.
    #[serde(skip_serializing_if = "Option::is_none")]
    raw: Option<String>,
}

/// Write one NDJSON line (JSON + `\n`) to `out`.
async fn write_ndjson_line<W, T>(out: &mut W, value: &T) -> Result<()>
where
    W: AsyncWrite + Unpin,
    T: Serialize,
{
    let mut line = serde_json::to_string(value).context("serialising NDJSON line")?;
    line.push('\n');
    out.write_all(line.as_bytes())
        .await
        .context("writing NDJSON line")
}

/// `bro agents check <agent>` — preflight the harness resolution, the
/// routing decision, and (when routing) daemon reachability, without launching
/// anything or auto-starting a daemon. Read-only.
pub async fn spawn_check(
    config: Config,
    agent_id: &str,
    routing: &RoutingOptions,
) -> Result<crate::spawn::SpawnCheckReport> {
    use crate::spawn::{SpawnCheckReport, SpawnCheckRow, SpawnCheckStatus};

    let row = |name: &str, status: SpawnCheckStatus, message: String| SpawnCheckRow {
        name: name.to_string(),
        status,
        message,
    };
    let mut checks = Vec::new();

    // 1. The agent resolves — either configured or a bundled-catalog id.
    let configured = config.agents.get(agent_id);
    let catalog = crate::harness::by_id(agent_id);
    let (command, args) = match (configured, catalog) {
        (Some(entry), _) => {
            let AcpTransport::Stdio { command, args, .. } = &entry.transport;
            checks.push(row(
                "agent",
                SpawnCheckStatus::Pass,
                format!("configured in agents: ({command})"),
            ));
            (command.clone(), args.clone())
        }
        (None, Some(h)) => match h.acp_command {
            Some(command) => {
                checks.push(row(
                    "agent",
                    SpawnCheckStatus::Pass,
                    format!("bundled catalog ({} {})", command, h.acp_args.join(" ")),
                ));
                (
                    command.to_string(),
                    h.acp_args.iter().map(|s| s.to_string()).collect(),
                )
            }
            None => {
                checks.push(row(
                    "agent",
                    SpawnCheckStatus::Fail,
                    format!("'{agent_id}' has no ACP adapter and cannot be used as an agent"),
                ));
                (String::new(), Vec::new())
            }
        },
        (None, None) => {
            checks.push(row(
                "agent",
                SpawnCheckStatus::Fail,
                format!("'{agent_id}' is neither a configured agent nor a bundled-catalog id"),
            ));
            (String::new(), Vec::new())
        }
    };

    // 2. Routing decision.
    let base_url = routing
        .base_url
        .clone()
        .unwrap_or_else(|| crate::spawn::derive_base_url(&config.server.listen));
    let harness = crate::harness::match_invocation(&command, &args);
    let mut routable_harness: Option<&'static crate::harness::Harness> = None;
    if routing.direct {
        checks.push(row(
            "routing",
            SpawnCheckStatus::Warn,
            "--direct: the sub-agent uses its own provider auth".to_string(),
        ));
    } else {
        match harness {
            Some(h) if h.env_args_routable() => {
                routable_harness = Some(h);
                checks.push(row(
                    "routing",
                    SpawnCheckStatus::Pass,
                    format!("via daemon {base_url} [{}]", h.id),
                ));
            }
            Some(h) => {
                checks.push(row(
                    "routing",
                    SpawnCheckStatus::Warn,
                    format!(
                        "'{}' routes via synthesized config (interactive facet only); \
                         will run direct",
                        h.id
                    ),
                ));
            }
            None => checks.push(row(
                "routing",
                SpawnCheckStatus::Warn,
                "not catalog-matched; will run direct (set its `env` to route manually)"
                    .to_string(),
            )),
        }
    }

    // 3. Auth preflight — mirror `apply_routing`'s require_key so `--check`
    //    surfaces the same `auth_required` gate the launch would fail fast on.
    if let Some(h) = routable_harness {
        let target_is_local = routing
            .base_url
            .as_deref()
            .and_then(crate::spawn::listen_from_base_url)
            .map(|a| crate::spawn::listen_is_local(&a))
            .unwrap_or_else(|| crate::spawn::listen_is_local(&config.server.listen));
        let require_key = !target_is_local || !config.server.skip_auth;
        let has_key = crate::spawn::nonempty_env(crate::harness::BITROUTER_API_KEY_ENV).is_some();
        checks.push(if require_key && !has_key {
            row(
                "auth",
                SpawnCheckStatus::Fail,
                "daemon requires auth but BITROUTER_API_KEY is not set — export it or pass --direct"
                    .to_string(),
            )
        } else if require_key && !h.auth_is_bearer() {
            row(
                "auth",
                SpawnCheckStatus::Warn,
                format!(
                    "'{}' sends a non-Bearer header the daemon rejects under skip_auth:false — \
                     the session will likely 401",
                    h.id
                ),
            )
        } else if has_key {
            row(
                "auth",
                SpawnCheckStatus::Pass,
                "BITROUTER_API_KEY present".to_string(),
            )
        } else {
            row(
                "auth",
                SpawnCheckStatus::Pass,
                "skip_auth: credential-less requests admitted".to_string(),
            )
        });
    }

    // 4. Daemon reachability — only meaningful when routing is active. Read-only
    //    (no auto-start): `--check` observes, it does not mutate.
    if routable_harness.is_some() {
        checks.push(if crate::spawn::base_url_reachable(&base_url).await {
            row(
                "daemon",
                SpawnCheckStatus::Pass,
                format!("{base_url} is reachable"),
            )
        } else {
            row(
                "daemon",
                SpawnCheckStatus::Fail,
                format!(
                    "{base_url} is unreachable — run `{} start` (or pass --direct)",
                    invocation::name()
                ),
            )
        });
    }

    Ok(SpawnCheckReport {
        agent: agent_id.to_string(),
        base_url,
        model: routing.model.clone(),
        checks,
    })
}

// ── serve ─────────────────────────────────────────────────────────────────────

fn controller_identity(
    agent_id: &str,
    command: &str,
    args: &[String],
    endpoint: Option<&crate::harness::HarnessEndpointPlan>,
) -> bitrouter_sdk::acp::controller::ControllerIdentity {
    if let Some(endpoint) = endpoint {
        return bitrouter_sdk::acp::controller::ControllerIdentity::new(
            endpoint.harness_id,
            endpoint.adapter_package,
            endpoint.adapter_version,
        );
    }
    if let Some(harness) = crate::harness::match_invocation(command, args)
        && let Some((package, version)) = harness.maintained_adapter_identity()
        && harness.uses_maintained_adapter(command, args)
    {
        return bitrouter_sdk::acp::controller::ControllerIdentity::new(
            harness.id, package, version,
        );
    }
    bitrouter_sdk::acp::controller::ControllerIdentity::new(
        agent_id,
        "configured-acp-adapter",
        "configured",
    )
}

fn controller_endpoint(
    endpoint: &crate::harness::HarnessEndpointPlan,
) -> bitrouter_sdk::acp::controller::ProviderEndpointPlan {
    let protocol = match endpoint.protocol {
        crate::harness::HarnessProtocol::Anthropic => LlmProtocol::Anthropic,
        crate::harness::HarnessProtocol::OpenAi => LlmProtocol::OpenAi,
    };
    bitrouter_sdk::acp::controller::ProviderEndpointPlan {
        provider_id: endpoint.provider_id.to_string(),
        protocol,
        base_url: endpoint.base_url.clone(),
        headers: endpoint.headers.clone(),
    }
}

struct DaemonRouteControl {
    socket_path: PathBuf,
    api_principal: String,
    controller_instance_id: String,
}

impl DaemonRouteControl {
    fn new(
        socket_path: PathBuf,
        api_principal: impl Into<String>,
        controller_instance_id: impl Into<String>,
    ) -> Self {
        Self {
            socket_path,
            api_principal: api_principal.into(),
            controller_instance_id: controller_instance_id.into(),
        }
    }

    async fn route_state(
        &self,
        command: crate::daemon::DaemonCommand,
    ) -> std::result::Result<RouteControlState, RouteControlError> {
        match crate::daemon::send_command(&self.socket_path, &command).await {
            Ok(crate::daemon::DaemonResponse::AcpRouteState {
                available, current, ..
            }) => Ok(RouteControlState { available, current }),
            Ok(crate::daemon::DaemonResponse::Error { message }) => {
                if message.starts_with("route is not available") {
                    Err(RouteControlError::invalid_route(message))
                } else {
                    Err(RouteControlError::unavailable(message))
                }
            }
            Ok(other) => Err(RouteControlError::unavailable(format!(
                "daemon returned an unexpected route-control response: {other:?}"
            ))),
            Err(error) => Err(RouteControlError::unavailable(format!(
                "daemon route control is unavailable: {error}"
            ))),
        }
    }
}

#[async_trait::async_trait]
impl AcpRouteControl for DaemonRouteControl {
    async fn list(
        &self,
        session_id: &str,
    ) -> std::result::Result<RouteControlState, RouteControlError> {
        self.route_state(crate::daemon::DaemonCommand::AcpRouteList {
            api_principal: self.api_principal.clone(),
            controller_instance_id: self.controller_instance_id.clone(),
            session_id: session_id.to_string(),
        })
        .await
    }

    async fn set(
        &self,
        session_id: &str,
        route: &str,
    ) -> std::result::Result<RouteControlState, RouteControlError> {
        self.route_state(crate::daemon::DaemonCommand::AcpRouteSet {
            api_principal: self.api_principal.clone(),
            controller_instance_id: self.controller_instance_id.clone(),
            session_id: session_id.to_string(),
            route: route.to_string(),
        })
        .await
    }

    async fn reset(
        &self,
        session_id: &str,
    ) -> std::result::Result<RouteControlState, RouteControlError> {
        self.route_state(crate::daemon::DaemonCommand::AcpRouteReset {
            api_principal: self.api_principal.clone(),
            controller_instance_id: self.controller_instance_id.clone(),
            session_id: session_id.to_string(),
        })
        .await
    }

    async fn session_closed(&self, session_id: &str) -> std::result::Result<(), RouteControlError> {
        self.reset(session_id).await.map(|_| ())
    }

    async fn disconnected(&self) -> std::result::Result<(), RouteControlError> {
        match crate::daemon::send_command(
            &self.socket_path,
            &crate::daemon::DaemonCommand::AcpControllerCleanup {
                api_principal: self.api_principal.clone(),
                controller_instance_id: self.controller_instance_id.clone(),
            },
        )
        .await
        {
            Ok(crate::daemon::DaemonResponse::Ok) => Ok(()),
            Ok(crate::daemon::DaemonResponse::Error { message }) => {
                Err(RouteControlError::unavailable(message))
            }
            Ok(other) => Err(RouteControlError::unavailable(format!(
                "daemon returned an unexpected cleanup response: {other:?}"
            ))),
            Err(error) => Err(RouteControlError::unavailable(format!(
                "daemon controller cleanup is unavailable: {error}"
            ))),
        }
    }
}

/// Session-attributed spend read back from the daemon's metering store for
/// this controller's own credential-bound traffic.
struct DaemonSessionCost {
    socket_path: PathBuf,
    api_principal: String,
    controller_instance_id: String,
}

/// How long one usage update may wait on the daemon before it is forwarded
/// without a figure. The lookup sits on the controller's forward path, so a
/// stalled daemon must cost the ACP client a cost line, never its update stream.
const SESSION_COST_LOOKUP_TIMEOUT: Duration = Duration::from_secs(2);

#[async_trait::async_trait]
impl AcpSessionCost for DaemonSessionCost {
    async fn attributed_cost(&self, session_id: &str) -> Option<Cost> {
        let command = crate::daemon::DaemonCommand::AcpSessionSpend {
            api_principal: self.api_principal.clone(),
            controller_instance_id: self.controller_instance_id.clone(),
            session_id: session_id.to_string(),
        };
        let response = tokio::time::timeout(
            SESSION_COST_LOOKUP_TIMEOUT,
            crate::daemon::send_command(&self.socket_path, &command),
        )
        .await;
        match response {
            Ok(Ok(crate::daemon::DaemonResponse::AcpSessionSpend {
                spend_micro_usd,
                requests,
                unpriced,
            })) => attributed_cost(spend_micro_usd, requests, unpriced),
            Ok(Ok(crate::daemon::DaemonResponse::Error { message })) => {
                tracing::debug!(session_id, %message, "session cost is unavailable");
                None
            }
            Ok(Ok(other)) => {
                tracing::debug!(
                    session_id,
                    ?other,
                    "daemon returned an unexpected session-spend response"
                );
                None
            }
            Ok(Err(error)) => {
                tracing::debug!(session_id, %error, "session cost lookup failed");
                None
            }
            Err(_) => {
                tracing::debug!(session_id, "session cost lookup timed out");
                None
            }
        }
    }
}

/// The figure a session's spend summary puts on the wire, or nothing.
///
/// Absent when no request was attributed to the session, and when none of
/// the attributed requests carries charge evidence: a session BitRouter routed
/// but could not price is not a free one, so it is never `$0.00`. When some
/// rows are priced the figure is their sum — the same floor
/// `bro status --requests` reports alongside its unpriced count.
fn attributed_cost(spend_micro_usd: u64, requests: u64, unpriced: u64) -> Option<Cost> {
    (requests > 0 && unpriced < requests)
        .then(|| Cost::new(spend_micro_usd as f64 / 1_000_000.0, "USD"))
}

/// Session-attributed spend, read **off** the controller's forward path.
///
/// The controller consults its cost bridge inside its dispatch handler, which
/// the SDK runs one message at a time — so a bridge that awaited the daemon
/// there would stop every `session/update` behind it for as long as the
/// lookup took. Under `acp prompt` nobody notices; under interactive `chat` a
/// person watches the transcript stop. This bridge therefore answers from a
/// cache at once and refreshes it in the background, single-flight per
/// session.
///
/// The honesty rule the cache keeps: a figure is stored only when the daemon
/// measured one, so a session that was never metered stays absent rather
/// than becoming `$0.00`; and a stale cumulative figure — the spend as of the
/// previous refresh — is still a figure BitRouter measured, which understates
/// rather than invents. A refresh that yields nothing never erases a figure
/// already held.
struct CachedSessionCost {
    inner: Arc<dyn AcpSessionCost>,
    /// Shared with each in-flight refresh task, which outlives the `&self`
    /// borrow it was spawned from.
    cache: Arc<CostCache>,
}

/// What the cache holds: the last figure the daemon confirmed per session,
/// and which sessions have a refresh in flight — so a burst of usage updates
/// costs one daemon round trip rather than one each.
#[derive(Default)]
struct CostCache {
    figures: std::sync::Mutex<std::collections::HashMap<String, Cost>>,
    refreshing: std::sync::Mutex<std::collections::HashSet<String>>,
}

impl CostCache {
    /// The figure held for `session_id`, if the daemon has ever confirmed one.
    fn held(&self, session_id: &str) -> Option<Cost> {
        match self.figures.lock() {
            Ok(figures) => figures.get(session_id).cloned(),
            Err(poisoned) => poisoned.into_inner().get(session_id).cloned(),
        }
    }

    /// Claim the refresh for `session_id`; `false` when one is already running.
    fn claim_refresh(&self, session_id: &str) -> bool {
        match self.refreshing.lock() {
            Ok(mut refreshing) => refreshing.insert(session_id.to_string()),
            Err(poisoned) => poisoned.into_inner().insert(session_id.to_string()),
        }
    }

    /// Record the daemon's answer and release the refresh claim. `None` keeps
    /// whatever was held: a lookup that yielded nothing is not evidence that
    /// an earlier figure was wrong.
    fn settle_refresh(&self, session_id: &str, figure: Option<Cost>) {
        if let Some(figure) = figure {
            match self.figures.lock() {
                Ok(mut figures) => {
                    figures.insert(session_id.to_string(), figure);
                }
                Err(poisoned) => {
                    poisoned.into_inner().insert(session_id.to_string(), figure);
                }
            }
        }
        match self.refreshing.lock() {
            Ok(mut refreshing) => {
                refreshing.remove(session_id);
            }
            Err(poisoned) => {
                poisoned.into_inner().remove(session_id);
            }
        }
    }
}

impl CachedSessionCost {
    fn new(inner: Arc<dyn AcpSessionCost>) -> Self {
        Self {
            inner,
            cache: Arc::new(CostCache::default()),
        }
    }
}

#[async_trait::async_trait]
impl AcpSessionCost for CachedSessionCost {
    async fn attributed_cost(&self, session_id: &str) -> Option<Cost> {
        // Runs on the forward path: nothing here awaits the daemon. The
        // refresh this update triggers lands in time for the *next* one.
        if self.cache.claim_refresh(session_id) {
            let inner = Arc::clone(&self.inner);
            let cache = Arc::clone(&self.cache);
            let session_id = session_id.to_string();
            tokio::spawn(async move {
                let figure = inner.attributed_cost(&session_id).await;
                cache.settle_refresh(&session_id, figure);
            });
        }
        self.cache.held(session_id)
    }
}

/// The route namespace both controller bridges address, and the socket to
/// reach it on.
///
/// **Nothing here is authenticated, and the naming is deliberate.** The
/// principal comes from the ordinary API credential (or `local` under
/// `skip_auth`), and the controller instance is a *claim* this process makes
/// about itself — it is never verified, and phase 3 removed the minted
/// credential that used to imply otherwise. What the pair buys is a namespace:
/// route leases are keyed by it, so one principal's leases cannot be reached
/// through another's. That is why a
/// `--direct` session, an unroutable harness, or an explicit `--base-url`
/// (no reviewed local control binding) yields no binding at all.
///
/// The credential is revoked on every exit through [`Self::revoke`]. It
/// also carries a daemon-side TTL, which is what bounds a panic exit — the
/// The route namespace both ACP bridges hang off.
///
/// Not a credential any more. `#853` replaced the minted `brac_` token with an
/// `api_principal` the daemon derives itself, so there is nothing to issue and
/// nothing to hand the harness — what remains is the principal/controller pair
/// that names a lease namespace, and the socket to reach it on.
///
/// `None` when the session has no controller instance to bind (direct, or a
/// harness without a maintained adapter), when the daemon supplied no
/// principal, or when `explicit_base_url` names a daemon this process cannot
/// vouch for. In every one of those the controller advertises no route control
/// and no attributed cost, which is what makes their absence honest rather
/// than a dead control.
/// Cloneable because a terminal login relaunches the harness (ACP_AUTH_SPEC
/// §7.3) and the second launch needs the same binding: re-deriving it would
/// repeat `open`'s `--base-url` note, and the binding is three owned strings.
#[derive(Clone)]
pub(crate) struct LocalControllerBinding {
    socket_path: PathBuf,
    api_principal: String,
    controller_instance_id: String,
}

/// Shared ACP launch/initialize owner used by `run` and the Code TUI.
///
/// It resolves the user-facing agent once, applies routing once, creates the
/// controller binding once, and only then lets a presentation driver open a
/// harness-native session. Nothing in this type depends on clap or a terminal.
pub(crate) struct SessionHost {
    config: Config,
    agent_id: String,
    options: LaunchOptions,
    routed: Routed,
    binding: Option<LocalControllerBinding>,
}

/// One live harness-native session returned by [`SessionHost`].
pub(crate) struct SessionHandle {
    pub(crate) client: AcpClient,
    pub(crate) session_id: String,
    pub(crate) agent_session_id: Option<String>,
    pub(crate) agent_id: String,
    pub(crate) via: Option<String>,
    pub(crate) launch_id: Option<String>,
    pub(crate) capabilities: CapabilitySnapshot,
    pub(crate) initial_settings: bitrouter_sdk::acp::client::SessionInitialSettings,
    cwd: PathBuf,
    mcp_servers: Vec<agent_client_protocol::schema::v1::McpServer>,
    pub(crate) updates: std::pin::Pin<Box<dyn futures::Stream<Item = SessionUpdate> + Send>>,
    pub(crate) permissions:
        std::pin::Pin<Box<dyn futures::Stream<Item = PendingPermission> + Send>>,
    session: ControlledSession,
}

impl SessionHandle {
    /// Open another native session unless `cancel` is signalled first.
    ///
    /// Cancellation deliberately leaves this handle unchanged: its current
    /// session, update stream, and controller remain owned by the caller.
    /// This lets an interactive driver restore the active session after a
    /// picker is dismissed while still shutting it down deterministically on
    /// process exit.
    pub(crate) async fn select_with_cancel(
        &mut self,
        selection: &SessionSelection,
        cancel: &CancellationToken,
    ) -> Result<()> {
        if cancel.is_cancelled() {
            return Err(lifecycle_cancelled());
        }
        let updates = self.client.subscribe_lifecycle_notifications();
        let ids = tokio::select! {
            biased;
            _ = cancel.cancelled() => return Err(lifecycle_cancelled()),
            result = open_session(
                &self.client,
                selection,
                self.cwd.clone(),
                self.mcp_servers.clone(),
            ) => result?,
        };
        let updates = updates.session_updates(&ids);
        self.session_id = ids.acp_session_id;
        self.agent_session_id = ids.agent_session_id;
        self.initial_settings = ids.initial_settings;
        self.updates = updates;
        Ok(())
    }

    pub(crate) fn take_updates(
        &mut self,
    ) -> std::pin::Pin<Box<dyn futures::Stream<Item = SessionUpdate> + Send>> {
        std::mem::replace(&mut self.updates, Box::pin(futures::stream::empty()))
    }

    pub(crate) fn take_permissions(
        &mut self,
    ) -> std::pin::Pin<Box<dyn futures::Stream<Item = PendingPermission> + Send>> {
        std::mem::replace(&mut self.permissions, Box::pin(futures::stream::empty()))
    }

    /// Disconnect the ACP client and deterministically reap the adapter child.
    pub(crate) async fn shutdown(&mut self) -> bool {
        self.session.shutdown().await
    }

    /// Resolve when the adapter process group has exited, whether it was a
    /// normal teardown or an abrupt harness death. This is independent of the
    /// controller duplex, which can otherwise remain open after the child has
    /// gone away.
    pub(crate) fn closed(&self) -> futures::future::BoxFuture<'static, ()> {
        self.session.closed()
    }
}

impl SessionHost {
    /// Resolve and prepare a controller without opening a native session.
    pub(crate) async fn prepare(ctx: SpawnContext<'_>, terminal_auth: bool) -> Result<Self> {
        let mut diagnostics = |message| eprintln!("{message}");
        Self::prepare_with_diagnostics(ctx, terminal_auth, &mut diagnostics).await
    }

    /// Prepare a controller while handing presentation-safe routing notes to
    /// the caller. Full-screen clients use this so ordinary fallback notices
    /// become part of their view instead of writing behind the renderer.
    pub(crate) async fn prepare_with_diagnostics(
        ctx: SpawnContext<'_>,
        terminal_auth: bool,
        diagnostics: &mut dyn FnMut(String),
    ) -> Result<Self> {
        let SpawnContext {
            source,
            mut config,
            agent_id,
            mut options,
            routing,
        } = ctx;
        let agent_id = resolve_agent_id(&config, agent_id)?;
        let cloud_credentials = crate::cloud::StandaloneCloudCredentials::new();
        let routed = apply_routing_with_diagnostics(
            source,
            &mut config,
            &agent_id,
            &routing,
            &cloud_credentials,
            diagnostics,
        )
        .await
        .map_err(anyhow::Error::new)?;
        config
            .agents
            .get(&agent_id)
            .with_context(|| format!("ACP agent '{agent_id}' is not configured"))?
            .validate()
            .with_context(|| format!("invalid ACP agent '{agent_id}'"))?;
        let binding = LocalControllerBinding::open_with_diagnostics(
            source,
            &config,
            &routed,
            routing.base_url.is_some(),
            diagnostics,
        );
        options.terminal_auth = terminal_auth;
        Ok(Self {
            config,
            agent_id,
            options,
            routed,
            binding,
        })
    }

    /// Launch/initialize the adapter and open the requested native session.
    pub(crate) async fn open(
        self,
        selection: &SessionSelection,
        cwd: PathBuf,
    ) -> Result<SessionHandle> {
        self.open_with_cancel(selection, cwd, &CancellationToken::new())
            .await
    }

    /// Launch/initialize the adapter and open a native session, stopping and
    /// reaping a child already allocated if `cancel` wins the lifecycle race.
    pub(crate) async fn open_with_cancel(
        self,
        selection: &SessionSelection,
        cwd: PathBuf,
        cancel: &CancellationToken,
    ) -> Result<SessionHandle> {
        if cancel.is_cancelled() {
            return Err(lifecycle_cancelled());
        }
        let mcp_servers = self.options.mcp_servers.clone();
        let terminal_auth = self.options.terminal_auth;
        let mut session = launch_controlled_with_cancel(
            &self.config,
            &self.agent_id,
            &self.routed,
            self.options,
            self.binding,
            cancel,
        )
        .await
        .with_context(|| format!("launching ACP session for agent '{}'", self.agent_id))?;
        // Subscribe before load: `session/load` replays history during the
        // request, and a later subscription would lose it.
        let updates = session.client.subscribe_lifecycle_notifications();
        let permissions = session.client.subscribe_permissions();
        let opened = tokio::select! {
            biased;
            _ = cancel.cancelled() => None,
            result = open_session(&session.client, selection, cwd.clone(), mcp_servers.clone()) => Some(result),
        };
        let ids = match opened {
            None => {
                return if session.shutdown().await {
                    Err(lifecycle_cancelled())
                } else {
                    Err(lifecycle_teardown_unconfirmed())
                };
            }
            Some(Ok(ids)) => ids,
            Some(Err(error)) => {
                return if session.shutdown().await {
                    Err(error.context("opening the harness session"))
                } else {
                    Err(error.context(
                        "opening the harness session; controller teardown did not confirm",
                    ))
                };
            }
        };
        let client = session.client.clone();
        let capabilities = CapabilitySnapshot::from_client(&client, terminal_auth);
        let updates = updates.session_updates(&ids);
        Ok(SessionHandle {
            client,
            session_id: ids.acp_session_id,
            agent_session_id: ids.agent_session_id,
            agent_id: self.agent_id,
            via: self.routed.via,
            launch_id: self.routed.launch_id,
            capabilities,
            initial_settings: ids.initial_settings,
            cwd,
            mcp_servers,
            updates,
            permissions,
            session,
        })
    }
}

/// The lifecycle owner was interrupted before a session could safely be
/// handed to its caller. Kept separate from an agent error so the UI can
/// restore its previous state without presenting an adapter failure.
#[derive(Debug)]
struct LifecycleCancelled;

impl std::fmt::Display for LifecycleCancelled {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str("ACP session lifecycle cancelled")
    }
}

impl std::error::Error for LifecycleCancelled {}

pub(crate) fn lifecycle_cancelled() -> anyhow::Error {
    anyhow::Error::new(LifecycleCancelled)
}

/// Whether a lifecycle job ended by the driver's explicit cancellation token.
/// Context layers preserve this cause, so a caller can distinguish a clean
/// user exit from an adapter failure without matching presentation text.
pub(crate) fn is_lifecycle_cancelled(error: &anyhow::Error) -> bool {
    error.chain().any(|cause| cause.is::<LifecycleCancelled>())
}

/// Cancellation only counts as clean after the controller child and its
/// route binding have both settled. Returning this separately keeps an
/// unconfirmed child from being silently converted into a normal UI exit.
fn lifecycle_teardown_unconfirmed() -> anyhow::Error {
    anyhow::anyhow!(
        "ACP session lifecycle cancelled, but controller teardown did not confirm; see the session log"
    )
}

impl LocalControllerBinding {
    fn open(
        source: &ConfigSource,
        config: &Config,
        routed: &Routed,
        explicit_base_url: bool,
    ) -> Option<Self> {
        let mut diagnostics = |message| eprintln!("{message}");
        Self::open_with_diagnostics(source, config, routed, explicit_base_url, &mut diagnostics)
    }

    fn open_with_diagnostics(
        source: &ConfigSource,
        config: &Config,
        routed: &Routed,
        explicit_base_url: bool,
        diagnostics: &mut dyn FnMut(String),
    ) -> Option<Self> {
        let (Some(_endpoint), Some(controller_instance_id), Some(api_principal)) = (
            routed.endpoint_plan.as_ref(),
            routed.controller_instance_id.clone(),
            routed.api_principal.clone(),
        ) else {
            return None;
        };
        if explicit_base_url {
            diagnostics(
                "note: _bitrouter/route/* is unavailable with an explicit --base-url; \
                 the remote model endpoint remains usable without session route control"
                    .to_string(),
            );
            return None;
        }
        Some(Self {
            socket_path: crate::daemon::socket_path_for(source, config),
            api_principal,
            controller_instance_id,
        })
    }

    /// The route bridge this binding makes trustworthy.
    fn route_control(&self) -> Arc<dyn AcpRouteControl> {
        Arc::new(DaemonRouteControl::new(
            self.socket_path.clone(),
            self.api_principal.clone(),
            self.controller_instance_id.clone(),
        ))
    }

    /// The cost bridge this binding makes attributable — cached, so the
    /// controller's forward path never waits on the daemon.
    fn session_cost(&self) -> Arc<dyn AcpSessionCost> {
        Arc::new(CachedSessionCost::new(Arc::new(DaemonSessionCost {
            socket_path: self.socket_path.clone(),
            api_principal: self.api_principal.clone(),
            controller_instance_id: self.controller_instance_id.clone(),
        })))
    }

    /// Drop every lease in this namespace. Idempotent on the daemon side, so
    /// calling it after the controller's own disconnect cleanup costs one
    /// round trip and nothing else; a failure is logged, because the caller is
    /// already on its way out.
    async fn revoke(&self) {
        let cleaned = crate::daemon::send_command(
            &self.socket_path,
            &crate::daemon::DaemonCommand::AcpControllerCleanup {
                api_principal: self.api_principal.clone(),
                controller_instance_id: self.controller_instance_id.clone(),
            },
        )
        .await;
        match cleaned {
            Ok(crate::daemon::DaemonResponse::Ok) => {}
            Ok(other) => tracing::debug!(?other, "controller lease cleanup was not acknowledged"),
            Err(error) => {
                tracing::debug!(%error, "controller lease cleanup did not reach the daemon")
            }
        }
    }
}

/// Launch one harness process and expose its connection-level ACP controller
/// over **stdio** until the ACP client disconnects.
///
/// Session creation, identifiers, lifecycle, history, and persistence remain
/// harness-native. `options` contributes only process-isolation settings here;
/// prompt deadlines belong to the ACP client on this transparent serve path.
pub async fn serve(ctx: SpawnContext<'_>) -> Result<()> {
    // `serve` never builds an `App` and never reaches `build_observability`,
    // so it gets the ignored-config warnings from neither. Emitted here, on
    // the config as read — before `apply_routing_with_cloud_credentials`
    // rewrites it — and safe to emit at once because `main` installs this
    // path's subscriber before dispatching (`init_session_log_tracing_
    // subscriber` for `bro acp serve`, `init_basic_tracing_subscriber`
    // for `bro spawn --serve`). Both write to stderr, so the JSON-RPC
    // stream on stdout stays pristine.
    for message in crate::assemble::ignored_config_warnings(&ctx.config) {
        tracing::warn!("{message}");
    }
    let host = SessionHost::prepare(ctx, false).await?;
    let agent_id = host.agent_id.as_str();
    let agent = host
        .config
        .agents
        .get(agent_id)
        .with_context(|| format!("ACP agent '{agent_id}' is not configured"))?;
    let AcpTransport::Stdio { command, args, env } = &agent.transport;
    let identity = controller_identity(agent_id, command, args, host.routed.endpoint_plan.as_ref());
    let mut controller_config = bitrouter_sdk::acp::controller::ControllerConfig::new(identity);
    if let Some(endpoint) = host.routed.endpoint_plan.as_ref() {
        controller_config = controller_config.endpoint(controller_endpoint(endpoint));
    }
    if host.options.turn_timeout.is_some() {
        eprintln!(
            "note: --turn-timeout is not enforced by transparent acp serve; \
             the ACP client controls prompt deadlines"
        );
    }
    let process =
        bitrouter_sdk::acp::up::AgentProcess::new(command.clone(), args.clone(), env.clone())
            .strip_inherited_env(host.options.strip_inherited_env);
    let mut controller =
        bitrouter_sdk::acp::controller::Controller::new(process, controller_config);
    if let Some(binding) = &host.binding {
        controller = controller
            .route_control(binding.route_control())
            .session_cost(binding.session_cost());
    }
    if let Some(capture) = open_capture(&host.config, agent_id, host.binding.as_ref()).await? {
        controller = controller.capture(capture);
    }
    controller
        .run(agent_client_protocol::Stdio::new())
        .await
        .map_err(|error| anyhow::anyhow!("acp serve: {error}"))
}

// ── chat ──────────────────────────────────────────────────────────────────────

/// Pass a `chat` startup failure through, having shown the session log first.
///
/// Piped `chat` uses it before its session exists. Every other ACP verb logs
/// to stderr as well as the file, so its child's account of a failed launch is
/// already on screen.
fn show_session_log(error: anyhow::Error) -> anyhow::Error {
    crate::chat::session::report_failed_launch();
    error
}

/// Launch a session for `agent_id` through the canonical Code conversation.
///
/// The legacy entry remains for compatibility. A terminal delegates before it
/// constructs a controller, so Code owns one interactive turn and permission
/// lifecycle. Redirected input or output retains the existing plain renderer.
pub async fn chat(ctx: SpawnContext<'_>) -> Result<()> {
    use std::io::IsTerminal as _;

    if std::io::stdin().is_terminal() && std::io::stdout().is_terminal() {
        let (services, request) = crate::actions::code::CodeServices::from_spawn_context(ctx)?;
        return crate::chat::code::run(services, Some(request)).await;
    }

    let SpawnContext {
        source,
        mut config,
        agent_id,
        options,
        routing,
    } = ctx;
    let cloud_credentials = crate::cloud::StandaloneCloudCredentials::new();
    let routed = apply_routing_with_cloud_credentials(
        source,
        &mut config,
        agent_id,
        &routing,
        &cloud_credentials,
    )
    .await
    .map_err(anyhow::Error::new)?;

    match &routed.via {
        Some(via) => eprintln!("code: '{agent_id}' routed via bitrouter ({via})"),
        None => eprintln!("code: '{agent_id}' running direct (not routed, not metered)"),
    }
    let binding =
        LocalControllerBinding::open(source, &config, &routed, routing.base_url.is_some());
    let ports = crate::actions::session::SessionPorts::open(
        source.clone(),
        crate::daemon::socket_path_for(source, &config),
    );
    let prompt_commands = crate::actions::session::prompt_commands(&config.chat)?;

    chat_piped(
        &config,
        agent_id,
        &routed,
        options,
        &cloud_credentials,
        binding,
        crate::actions::session::SessionSurface {
            commands: Vec::new(),
            prompt_commands,
            ports: &ports,
        },
    )
    .await
}

/// What to say when a harness answered `auth_required` and no login ran.
///
/// Names the harness and what it advertised, because those are the two things
/// the reader needs and the two things only the protocol knows.
///
/// The advice depends on **why** no login ran, and getting that wrong is worse
/// than saying less: telling someone who just cancelled to "sign in with the
/// harness's own tooling" hides the offer they are one keystroke from taking,
/// and telling a piped caller to choose from a prompt it will never see is a
/// control that cannot act.
///
/// Deliberately says nothing about providers, routes, or `providers login`:
/// this is the harness's own authentication, not the daemon's upstream
/// credential, and conflating the two sends the reader to the wrong fix.
fn unauthenticated_message(
    agent_id: &str,
    methods: &[agent_client_protocol::schema::v1::AuthMethod],
) -> String {
    use std::io::IsTerminal as _;

    if methods.is_empty() {
        return format!(
            "'{agent_id}' is not authenticated, and advertises no authentication method. \
             Sign in with the harness's own tooling, then run this again"
        );
    }
    let offered = methods
        .iter()
        .map(|method| match method.description() {
            Some(description) => format!("{} ({description})", method.name()),
            None => method.name().to_string(),
        })
        .collect::<Vec<_>>()
        .join("; ");
    // A method was offered and declined — either by a person choosing cancel,
    // or by there being no person to ask.
    if std::io::stdin().is_terminal() {
        format!(
            "'{agent_id}' is not authenticated and the sign-in was cancelled. \
             It offers: {offered}"
        )
    } else {
        format!(
            "'{agent_id}' is not authenticated. It offers: {offered}. \
             Run `{cli} code {agent_id}` in a terminal to sign in, or use the \
             harness's own tooling",
            cli = invocation::name()
        )
    }
}

async fn chat_piped(
    config: &Config,
    agent_id: &str,
    routed: &Routed,
    options: LaunchOptions,
    cloud_credentials: &crate::cloud::StandaloneCloudCredentials,
    binding: Option<LocalControllerBinding>,
    mut surface: crate::actions::session::SessionSurface<'_>,
) -> Result<()> {
    let cwd = std::env::current_dir().context("resolving current directory")?;
    let mcp_servers = options.mcp_servers.clone();
    let mut session = launch_controlled(config, agent_id, routed, options, binding)
        .await
        .with_context(|| format!("launching acp session for agent '{agent_id}'"))
        .map_err(show_session_log)?;
    let ids = match session.client.new_session(cwd, mcp_servers).await {
        Ok(ids) => ids,
        Err(error) => {
            // A harness that is merely unauthenticated is not a broken one, and
            // relaying its JSON-RPC error would leave the reader to guess which
            // of the two it is. The protocol already said which; say it.
            let unauthenticated = bitrouter_sdk::acp::client::is_auth_required(&error);
            let context = if unauthenticated {
                unauthenticated_message(agent_id, session.client.auth_methods())
            } else {
                "opening the harness session".to_string()
            };
            session.shutdown().await;
            let failure = error.context(context);
            // The log tail is for a launch that died with its reason on the
            // child's stderr. An unauthenticated harness already carries its
            // own explanation, and a tail would bury the sentence that gives
            // it.
            return Err(if unauthenticated {
                failure
            } else {
                show_session_log(failure)
            });
        }
    };
    let observability =
        build_observability(config, agent_id, &ids.acp_session_id, cloud_credentials).await;
    if let Some(recorder) = observability.recorder.clone() {
        spawn_tool_spans(recorder, session.client.subscribe_updates());
    }

    // Only now can the controller be asked what it advertised.
    surface.commands = crate::actions::session::offered_commands(&session.client);
    let ended = crate::chat::session::chat_plain(
        &session.client,
        &ids.acp_session_id,
        agent_id,
        observability.recorder,
        surface,
    )
    .await;

    // Teardown is here rather than in the loop: the loop borrows the client,
    // and the controller that owns the harness child is this function's.
    session.shutdown().await;
    if let Some(exporter) = observability.exporter {
        exporter.shutdown();
    }
    ended
}

// ── prompt ────────────────────────────────────────────────────────────────────

/// Launch a session for `agent_id`, send one prompt, and stream each
/// [`SessionUpdateKind`] as a self-describing NDJSON line to `out`.
///
/// When `no_wait` is false (the default): subscribe to updates, send the
/// prompt, stream updates while the prompt is in flight, emit a terminal
/// `{"type":"result","stop_reason":"…"}` line, then shut down the session.
///
/// When `no_wait` is true: shut down the session immediately after emitting
/// `{"type":"submitted"}`. The agent child is terminated; callers needing a
/// persistent session should use `bro acp serve` instead.
///
/// `options.contract` is the optional `--result-schema` contract: its
/// instruction rides the prompt, and the terminal `result` line gains
/// `result`/`schema_ok` (+ `raw` on failure) fields. `options.policy` answers
/// the harness's permission requests; the tally returned is what the caller
/// exits with.
pub async fn prompt<W>(
    ctx: SpawnContext<'_>,
    text: &str,
    options: PromptOptions,
    out: &mut W,
) -> Result<PermissionTally>
where
    W: AsyncWrite + Unpin + Send,
{
    let PromptOptions {
        no_wait,
        contract,
        policy,
        format,
        session: session_selection,
        cwd,
    } = options;
    let mut presenter = Presenter::new(format, contract.is_some(), out);
    let cloud_credentials = crate::cloud::StandaloneCloudCredentials::new();
    let host = match SessionHost::prepare(ctx, false).await {
        Ok(host) => host,
        Err(error) => {
            if let Some(routing) = error.downcast_ref::<RoutingError>() {
                presenter.routing_error(routing).await?;
            } else {
                presenter
                    .error("session_prepare_failed", &format!("{error:#}"))
                    .await?;
            }
            return Err(error);
        }
    };
    let config = host.config.clone();
    let agent_id = host.agent_id.clone();

    // A config command expands identically here and in the terminal, through
    // the one resolver. The BitRouter half is deliberately empty: `acp prompt
    // "/status"` is not a scheduled surface, so a name this process answers
    // goes to the agent exactly as it does today.
    let expanded = match bitrouter_tui::machine::resolve(
        &[],
        &crate::actions::session::prompt_commands(&config.chat)?,
        text,
    ) {
        bitrouter_tui::machine::Resolution::Expand(prompt) => prompt,
        _ => text.to_string(),
    };
    let text = expanded.as_str();

    let cwd = match cwd {
        Some(cwd) => cwd
            .canonicalize()
            .with_context(|| format!("resolving working directory {}", cwd.display()))?,
        None => std::env::current_dir().context("resolving current directory")?,
    };
    let mut session = match host.open(&session_selection, cwd).await {
        Ok(session) => session,
        Err(error) => {
            presenter
                .error("agent_not_ready", &format!("{error:#}"))
                .await?;
            return Err(error);
        }
    };
    let mut updates = session.take_updates();
    let permissions = session.take_permissions();
    let observability =
        build_observability(&config, &agent_id, &session.session_id, &cloud_credentials).await;
    if let Some(recorder) = observability.recorder.clone() {
        spawn_tool_spans(recorder, session.client.subscribe_updates());
    }

    // First line: correlate this session with the cost/metering the
    // orchestrator later queries.
    //
    // `session_id` is the harness-native ACP id — the one every subsequent
    // line and every ACP method uses. It is **not** the spend key on this
    // path: metering attributes ACP traffic by the controller instance a
    // request *claims*, and `prompt` routes with a launch token and declares no
    // controller, so `requests.controller_instance_id` is null for every row it
    // produces. `launch_id` is what joins here, as it always has.
    //
    // The controller id is deliberately absent rather than reported: the value
    // this process mints reaches the daemon only as a *claimed* header, which
    // §5.5 of the controller spec treats as correlation evidence and not as
    // authorization. Emitting it would name a column that is null. When this
    // path declares a controller namespace — as `acp serve` and `chat` do — it
    // can come back and be true.
    //
    // `via` is null when running direct.
    presenter
        .session(&serde_json::json!({
            "type": "session",
            "session_id": session.session_id,
            "agent_session_id": session.agent_session_id,
            "agent": agent_id,
            "via": session.via,
            // The token this session's requests carry, so an orchestrator can
            // group its spend. Null when the caller supplied their own
            // credential and the traffic is therefore not separable.
            "launch_id": session.launch_id,
        }))
        .await?;

    if no_wait {
        // v1 no-wait: emit ack, then shut down immediately. The agent child is
        // killed on shutdown. Callers needing a persistent background session
        // should use `bro acp serve` instead. Whatever the harness asks
        // between ack and teardown is denied by the client's own ledger.
        presenter.submitted().await?;
        let clean = session.shutdown().await;
        if let Some(exporter) = observability.exporter {
            exporter.shutdown();
        }
        return teardown_result(clean).map(|()| PermissionTally::default());
    }

    // Headless: there is no manager to broker permissions and none will ever
    // attach, so the policy answers each request as it arrives, through the
    // same wire a keystroke would. The client also denies whatever is still
    // outstanding when a turn is abandoned or the connection tears down; this
    // loop is the one that answers promptly, so a turn that only needs consent
    // it will never get ends now.
    let mut turn = Turn {
        client: &session.client,
        session_id: &session.session_id,
        agent_id: &agent_id,
        recorder: observability.recorder.clone(),
        wire: Wire::new(&session.client, &session.session_id),
        permissions,
        policy,
        tally: PermissionTally::default(),
    };
    let outcome = prompt_wait(&mut turn, &mut updates, text, contract, &mut presenter).await;
    let tally = turn.tally;
    let clean = session.shutdown().await;
    if let Some(exporter) = observability.exporter {
        // Flush the span batch before exit; spans are lost otherwise.
        exporter.shutdown();
    }
    // The turn's own failure is the more specific thing that went wrong, so it
    // outranks a teardown that did not confirm.
    if let Err(error) = outcome {
        presenter
            .error("turn_failed", &format!("{error:#}"))
            .await?;
        return Err(error);
    }
    teardown_result(clean).map(|()| tally)
}

async fn open_session(
    client: &AcpClient,
    selection: &SessionSelection,
    cwd: PathBuf,
    mcp_servers: Vec<agent_client_protocol::schema::v1::McpServer>,
) -> Result<bitrouter_sdk::acp::client::SessionIds> {
    match selection {
        SessionSelection::New => client.new_session(cwd, mcp_servers).await,
        SessionSelection::Load(session_id) => {
            client.load_session(session_id, cwd, mcp_servers).await
        }
        SessionSelection::Resume(session_id) => {
            client.resume_session(session_id, cwd, mcp_servers).await
        }
    }
}

/// What a `prompt` exits with when the turn itself was fine.
///
/// A teardown that did not confirm **fails the command**. Before `prompt`
/// moved onto the shared client it did this by `?`-ing on `shutdown()`; the
/// controlled session logs and returns a flag instead, and the flag went
/// unread — so an orchestrator scripting `bro acp prompt` saw exit 0
/// for a run whose harness child may still be alive. That was an undeclared
/// change, and this is it declared: the NDJSON on stdout is already complete
/// and correct, and the status code is what says the process did not end
/// cleanly.
fn teardown_result(clean: bool) -> Result<()> {
    if clean {
        Ok(())
    } else {
        Err(anyhow::anyhow!(
            "shutting down acp session: teardown did not confirm; see the session log"
        ))
    }
}

/// One harness behind an in-process connection-level controller, with the
/// shared ACP client connected to that controller as its manager.
///
/// This is exactly the stack `acp serve` exposes over stdio; the only
/// difference is that the ACP client on the other end of the duplex is this
/// process. Session identity, lifecycle, and history therefore stay
/// harness-native, and there is no second scheduler: the controller, the
/// harness child's I/O, and the client all run on this runtime.
pub(crate) struct ControlledSession {
    pub(crate) client: AcpClient,
    cleanup: ControlledCleanup,
}

/// Resources created before an [`AcpClient`] has completed initialize.
///
/// A cancelled initialize has no client through which to close the duplex, so
/// it must abort the controller first and still wait for the reaper/binding
/// cleanup. Keeping these resources together prevents a dropped lifecycle
/// future from orphaning a harness process.
struct ControlledCleanup {
    /// The route namespace this session's leases live in, when it has one.
    /// Cleaned up by [`ControlledSession::shutdown`] on every exit.
    binding: Option<LocalControllerBinding>,
    /// Resolves once the harness child and its process group are reaped.
    /// Awaited in [`ControlledSession::shutdown`], because tearing the
    /// connection down is not the same as the child being gone — the SDK drops
    /// the transport task rather than letting it confirm the kill.
    reaped: futures::future::Shared<futures::channel::oneshot::Receiver<()>>,
    /// The controller's own `run`. It ends when the client side of the duplex
    /// closes, and awaiting it is what proves the harness child was reaped.
    controller: tokio::task::JoinHandle<std::result::Result<(), anyhow::Error>>,
}

/// How long teardown waits for the controller to finish reaping its harness
/// before giving up and reporting it.
const CONTROLLER_EXIT_TIMEOUT: Duration = Duration::from_secs(10);

impl ControlledSession {
    /// Tear the client down (which closes the duplex and so ends the
    /// controller), wait for the controller to confirm, then drop every route
    /// lease in the namespace. Returns whether every step confirmed; failures
    /// are logged rather than propagated, because the caller has already
    /// asked to stop and there is nothing left to retry — but an interactive
    /// caller wants to know the session ended abnormally, so it can say where
    /// the reason is.
    pub(crate) async fn shutdown(&mut self) -> bool {
        let mut clean = true;
        if let Err(error) = self.client.shutdown().await {
            tracing::warn!(%error, "acp teardown unconfirmed; the harness may not have terminated");
            clean = false;
        }
        self.cleanup.finish(false).await && clean
    }

    /// Return an owned watcher so an interactive client can observe an
    /// adapter death without relying on the controller's connection close.
    fn closed(&self) -> futures::future::BoxFuture<'static, ()> {
        let reaped = self.cleanup.reaped.clone();
        Box::pin(async move {
            let _ = reaped.await;
        })
    }
}

impl ControlledCleanup {
    /// Stop a controller whose client initialization never completed, then
    /// wait through the same bounded cleanup sequence as a normal shutdown.
    async fn abort(&mut self) -> bool {
        self.controller.abort();
        self.finish(true).await
    }

    /// Confirm the child group, controller task, and route binding have all
    /// settled. `controller_aborted` accepts the expected cancelled join error
    /// from the initialize-cancellation path while preserving every other
    /// controller failure as diagnostic evidence.
    async fn finish(&mut self, controller_aborted: bool) -> bool {
        let mut clean = true;
        // The connection is down; the child may not be. `kill_on_drop` reaches
        // the wrapper (`npx`) and not the `node` it spawned, so the group kill
        // has to be confirmed rather than assumed.
        if tokio::time::timeout(bitrouter_sdk::acp::up::REAP_CONFIRM, self.reaped.clone())
            .await
            .is_err()
        {
            tracing::warn!("harness child not confirmed reaped; a grandchild may have survived");
            clean = false;
        }
        match tokio::time::timeout(CONTROLLER_EXIT_TIMEOUT, &mut self.controller).await {
            Ok(Ok(Ok(()))) => {}
            Ok(Ok(Err(error))) => tracing::debug!(%error, "acp controller ended with an error"),
            Ok(Err(error)) => {
                if controller_aborted && error.is_cancelled() {
                    tracing::debug!("acp controller aborted during lifecycle cancellation");
                } else {
                    tracing::warn!(%error, "acp controller task failed");
                    clean = false;
                }
            }
            Err(_) => {
                tracing::warn!("acp controller did not exit within {CONTROLLER_EXIT_TIMEOUT:?}");
                self.controller.abort();
                clean = false;
            }
        }
        // The controller revokes on its own disconnect; this is the backstop
        // for the paths where it never got that far.
        if let Some(binding) = &self.binding {
            binding.revoke().await;
        }
        clean
    }
}

async fn open_capture(
    config: &Config,
    agent_id: &str,
    binding: Option<&LocalControllerBinding>,
) -> Result<Option<Arc<dyn bitrouter_sdk::acp::capture::CapturePort>>> {
    if !config.acp_recording.enabled {
        return Ok(None);
    }
    let db = crate::db::connect(&config.database.url)
        .await
        .context("opening the opt-in ACP content store")?;
    crate::db::run_migrations(&db).await?;
    let store = crate::acp_trajectory::CanonicalStore::new(db);
    let recorder = store
        .recorder(crate::acp_trajectory::RecordingScope {
            owner: "local".into(),
            source: agent_id.to_owned(),
            controller_instance_id: binding.map(|value| value.controller_instance_id.clone()),
            route_scope_id: binding.map(|value| value.api_principal.clone()),
        })
        .await?;
    if let Some(binding) = binding {
        let command = crate::daemon::DaemonCommand::AcpRecordingRegister {
            connection_id: recorder.connection_id().to_owned(),
            api_principal: binding.api_principal.clone(),
            controller_instance_id: binding.controller_instance_id.clone(),
        };
        let acknowledged = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            crate::daemon::send_command(&binding.socket_path, &command),
        )
        .await;
        match acknowledged {
            Ok(Ok(crate::daemon::DaemonResponse::AcpRecordingRegistered { registration }))
                if registration.connection_id == recorder.connection_id()
                    && registration.version == crate::evolution::inventory::INVENTORY_VERSION
                    && registration.invalid_reason.is_none() => {}
            response => tracing::warn!(
                ?response,
                "ACP capture has no acknowledged gateway cost coverage"
            ),
        }
    }
    Ok(Some(recorder))
}

/// Launch `agent_id` behind an in-process controller and connect the shared
/// client to it. The controller carries the same identity and endpoint plan
/// `acp serve` gives it, so a routed session configures the harness's
/// provider through `providers/set` exactly as a served one does — and, with
/// a `binding`, the same route and cost bridges.
async fn launch_controlled(
    config: &Config,
    agent_id: &str,
    routed: &Routed,
    options: LaunchOptions,
    binding: Option<LocalControllerBinding>,
) -> Result<ControlledSession> {
    launch_controlled_with_cancel(
        config,
        agent_id,
        routed,
        options,
        binding,
        &CancellationToken::new(),
    )
    .await
}

/// Launch the controller and run its initialize handshake while retaining
/// cleanup ownership if the caller cancels mid-handshake.
async fn launch_controlled_with_cancel(
    config: &Config,
    agent_id: &str,
    routed: &Routed,
    options: LaunchOptions,
    binding: Option<LocalControllerBinding>,
    cancel: &CancellationToken,
) -> Result<ControlledSession> {
    if cancel.is_cancelled() {
        return Err(lifecycle_cancelled());
    }
    let agent = config
        .agents
        .get(agent_id)
        .with_context(|| format!("no acp agent configured for '{agent_id}'"))?;
    agent
        .validate()
        .with_context(|| format!("invalid ACP agent '{agent_id}'"))?;
    let AcpTransport::Stdio { command, args, env } = &agent.transport;

    let identity = controller_identity(agent_id, command, args, routed.endpoint_plan.as_ref());
    let mut controller_config = bitrouter_sdk::acp::controller::ControllerConfig::new(identity);
    if let Some(endpoint) = routed.endpoint_plan.as_ref() {
        controller_config = controller_config.endpoint(controller_endpoint(endpoint));
    }
    let mut process =
        bitrouter_sdk::acp::up::AgentProcess::new(command.clone(), args.clone(), env.clone())
            .strip_inherited_env(options.strip_inherited_env);
    let reaped = process.reaped().shared();
    let mut controller =
        bitrouter_sdk::acp::controller::Controller::new(process, controller_config);
    if let Some(binding) = &binding {
        controller = controller
            .route_control(binding.route_control())
            .session_cost(binding.session_cost());
    }

    if let Some(capture) = open_capture(config, agent_id, binding.as_ref()).await? {
        controller = controller.capture(capture);
    }

    // In-process duplex: no bytes, no pipe, no second process between the
    // manager and the controller it drives.
    let (manager_side, controller_side) = agent_client_protocol::Channel::duplex();
    let controller = tokio::spawn(async move {
        controller
            .run(controller_side)
            .await
            .map_err(|error| anyhow::anyhow!("acp controller: {error}"))
    });
    let mut cleanup = ControlledCleanup {
        binding,
        reaped,
        controller,
    };
    let connected = tokio::select! {
        biased;
        _ = cancel.cancelled() => None,
        result = AcpClient::connect(
            manager_side,
            ClientOptions {
                turn_timeout: options.turn_timeout,
                // §6.2's first condition always holds here — the harness is a local
                // child of this process, `AcpTransport::Stdio` being the only
                // transport, so its invocation is always reproducible. The second
                // is the caller's to answer, because only an interactive caller can
                // hand a person the terminal the login needs.
                terminal_auth: options.terminal_auth,
            },
        ) => Some(result),
    };
    match connected {
        Some(Ok(client)) => Ok(ControlledSession { client, cleanup }),
        Some(Err(error)) => {
            let clean = cleanup.abort().await;
            if !clean {
                tracing::warn!("acp lifecycle failure left teardown unconfirmed");
            }
            Err(error)
        }
        None => {
            let clean = cleanup.abort().await;
            if clean {
                Err(lifecycle_cancelled())
            } else {
                tracing::warn!("acp lifecycle cancellation left teardown unconfirmed");
                Err(lifecycle_teardown_unconfirmed())
            }
        }
    }
}

/// Everything one prompt turn needs beyond its text: the client to drive, the
/// harness-native session it belongs to, where its telemetry goes, and how
/// its permission requests are answered.
struct Turn<'a> {
    client: &'a AcpClient,
    session_id: &'a str,
    agent_id: &'a str,
    recorder: Option<Arc<bitrouter_telemetry::otel::acp::AcpSpanRecorder>>,
    /// The wire half of the session's effects — the same interpreter `chat`
    /// runs, so an answer here reaches the agent as an answer there does.
    wire: Wire<'a>,
    /// The harness's permission requests, polled beside the updates.
    permissions: std::pin::Pin<Box<dyn futures::Stream<Item = PendingPermission> + Send>>,
    /// The rule that answers them.
    policy: Policy,
    /// What it decided, for the exit status.
    tally: PermissionTally,
}

/// Inner implementation for the wait (non-`--no-wait`) path.
async fn prompt_wait<W>(
    turn: &mut Turn<'_>,
    updates: &mut (impl futures::Stream<Item = SessionUpdate> + Unpin),
    text: &str,
    contract: Option<crate::result_contract::ResultContract>,
    presenter: &mut Presenter<'_, W>,
) -> Result<()>
where
    W: AsyncWrite + Unpin,
{
    // Subscribe to updates BEFORE prompting so no streamed update is missed.
    // The **raw** stream: the text presentation is a journal, which is a
    // protocol client, and the NDJSON one translates for itself.
    let task = match &contract {
        // The contract clause rides the subagent's task prompt.
        Some(c) => format!("{text}{}", c.instruction()),
        None => text.to_string(),
    };
    let (response, reply) = run_turn(turn, updates, &task, contract.is_some(), presenter).await?;

    // Extract + validate the machine-consumable result. On failure: ONE
    // repair re-prompt, then `schema_ok:false` + raw text — the orchestrator
    // is never blocked on a malformed reply.
    let (response, result, schema_ok, raw) = match &contract {
        None => (response, None, None, None),
        Some(c) => match c.check(&reply) {
            Ok(value) => (response, Some(value), Some(true), None),
            Err(problem) => {
                let (response, reply) =
                    run_turn(turn, updates, &c.repair_prompt(&problem), true, presenter).await?;
                match c.check(&reply) {
                    Ok(value) => (response, Some(value), Some(true), None),
                    Err(_) => (
                        response,
                        Some(serde_json::Value::Null),
                        Some(false),
                        Some(reply),
                    ),
                }
            }
        },
    };

    // Emit the terminal result line. `response.stop_reason` is an ACP
    // `StopReason` that serializes to its snake_case wire form (e.g.
    // `"end_turn"`).
    presenter
        .result(&ResultLine {
            kind: "result",
            stop_reason: response.stop_reason,
            result,
            schema_ok,
            raw,
        })
        .await
}

/// Drive one prompt turn: present its updates (accumulating message text when
/// `capture`), answer its permission requests under the turn's policy, report
/// the turn's telemetry, and return the typed response plus the reply text.
async fn run_turn<W>(
    turn: &mut Turn<'_>,
    updates: &mut (impl futures::Stream<Item = SessionUpdate> + Unpin),
    text: &str,
    capture: bool,
    presenter: &mut Presenter<'_, W>,
) -> Result<(agent_client_protocol::schema::v1::PromptResponse, String)>
where
    W: AsyncWrite + Unpin,
{
    let mut reply = String::new();
    let started = std::time::Instant::now();
    // Drive updates, permissions, and the prompt concurrently. The loop
    // returns the resolved `PromptResponse` directly, so there is no `Option`
    // to unwrap afterward.
    let response = {
        let prompt_future = turn.client.prompt(turn.session_id, text);
        tokio::pin!(prompt_future);

        loop {
            tokio::select! {
                biased;

                result = &mut prompt_future => {
                    let response = result.context("acp prompt failed")?;
                    // Non-blocking drain of any already-buffered updates.
                    loop {
                        let maybe = tokio::select! {
                            biased;
                            v = updates.next() => v,
                            _ = std::future::ready(()) => None,
                        };
                        match maybe {
                            Some(update) => presenter.update(update, capture, &mut reply).await?,
                            None => break,
                        }
                    }
                    break response;
                }

                maybe_update = updates.next() => {
                    if let Some(update) = maybe_update {
                        presenter.update(update, capture, &mut reply).await?;
                    }
                }

                // `if let`, not a `Some(..) =` pattern: a stream that ended
                // must not spin the select.
                request = turn.permissions.next() => {
                    if let Some(request) = request {
                        let (decision, prompt) = turn.wire.answer(request, &turn.policy).await;
                        turn.tally.record(decision);
                        presenter.permission(&prompt, decision).await?;
                    }
                }
            }
        }
    };
    presenter.turn_settled().await?;
    report_turn(
        turn.client,
        turn.agent_id,
        turn.recorder.as_ref(),
        &response,
        started.elapsed(),
    );
    Ok((response, reply))
}

/// Re-derive this turn's completion record from the prompt round-trip and hand
/// it to both telemetry consumers.
///
/// The engine's pipeline hook used to be the only thing that produced these.
/// Latency and stop reason are visible right here on the round-trip, and
/// context occupancy on the client's usage slot, so the record is built from
/// what the turn itself observed.
///
/// Takes its inputs loose rather than a `Turn`, because the piped chat path
/// ([`crate::chat::session::chat_plain`]) reports the same record and has no
/// prompt contract, no NDJSON sink, and therefore no `Turn`.
pub(crate) fn report_turn(
    client: &AcpClient,
    agent_id: &str,
    recorder: Option<&Arc<bitrouter_telemetry::otel::acp::AcpSpanRecorder>>,
    response: &agent_client_protocol::schema::v1::PromptResponse,
    latency: Duration,
) {
    let record = RequestCompleted {
        agent: agent_id.to_string(),
        stop_reason: format!("{:?}", response.stop_reason),
        latency_ms: u64::try_from(latency.as_millis()).unwrap_or(u64::MAX),
        context: client.context_usage().lock().ok().and_then(|slot| *slot),
    };
    if let Some(recorder) = recorder {
        recorder.turn_completed(&bitrouter_telemetry::otel::acp::TurnRecord {
            stop_reason: record.stop_reason.clone(),
            latency,
            context_used: record.context.map(|c| c.used),
            context_size: record.context.map(|c| c.size),
        });
    }
    drain_telemetry_record(record);
}

/// How one headless run is printed.
///
/// One sink, three presentations, chosen once by `--format`. Everything the
/// turn has to say goes through here, so the three cannot drift in *what* is
/// said — only in how.
enum Presenter<'a, W> {
    /// NDJSON: every update as the [`SessionUpdateKind`] it translates to,
    /// plus `session`, `permission`, `result`, `submitted`, and `error` lines.
    Ndjson { out: &'a mut W, seq: u64 },
    /// The assistant's text as it streams, and a trailing newline. Under a
    /// result contract, only the extracted result instead — the text was the
    /// working, not the answer.
    Quiet { out: &'a mut W, contracted: bool },
    /// The transcript as `chat` prints it to a pipe, written once per turn.
    /// Boxed: a journal and a render cache are a lot larger than a writer.
    Text {
        out: &'a mut W,
        transcript: Box<bitrouter_tui::plain::Transcript>,
    },
}

impl<'a, W> Presenter<'a, W>
where
    W: AsyncWrite + Unpin,
{
    fn new(format: PromptFormat, contracted: bool, out: &'a mut W) -> Self {
        match format {
            PromptFormat::Ndjson => Presenter::Ndjson { out, seq: 0 },
            PromptFormat::Quiet => Presenter::Quiet { out, contracted },
            PromptFormat::Text => Presenter::Text {
                out,
                transcript: Box::default(),
            },
        }
    }

    /// The fail-fast routing failure, before any session exists. NDJSON only:
    /// `main` already prints the reason on stderr, and the other two formats
    /// promise stdout carries the session and nothing else.
    async fn routing_error(&mut self, error: &RoutingError) -> Result<()> {
        if matches!(self, Presenter::Ndjson { .. }) {
            self.ndjson(error.ndjson()).await?;
            let Presenter::Ndjson { out, .. } = self else {
                return Ok(());
            };
            out.flush().await.ok();
        }
        Ok(())
    }

    /// A terminal error in the versioned stream. Text/quiet reserve stdout for
    /// user content, so their diagnostics continue to use stderr via `main`.
    async fn error(&mut self, code: &str, message: &str) -> Result<()> {
        if matches!(self, Presenter::Ndjson { .. }) {
            self.ndjson(serde_json::json!({
                "type": "error",
                "code": code,
                "message": message,
            }))
            .await?;
        }
        Ok(())
    }

    /// Add the BitRouter event version and monotonically increasing sequence
    /// number to one translated event before writing it.
    async fn ndjson(&mut self, value: serde_json::Value) -> Result<()> {
        let Presenter::Ndjson { out, seq } = self else {
            return Ok(());
        };
        let mut object = value
            .as_object()
            .cloned()
            .ok_or_else(|| anyhow::anyhow!("NDJSON event must serialize as an object"))?;
        object.insert("version".to_string(), serde_json::json!(1));
        object.insert("seq".to_string(), serde_json::json!(*seq));
        *seq = seq.saturating_add(1);
        write_ndjson_line(out, &object).await
    }

    /// The `session` correlation line.
    async fn session(&mut self, line: &serde_json::Value) -> Result<()> {
        match self {
            Presenter::Ndjson { .. } => self.ndjson(line.clone()).await,
            Presenter::Quiet { .. } => Ok(()),
            Presenter::Text { out, .. } => {
                let field = |name: &str| {
                    line.get(name)
                        .and_then(|v| v.as_str())
                        .unwrap_or("-")
                        .to_string()
                };
                let via = line
                    .get("via")
                    .and_then(|v| v.as_str())
                    .map_or_else(|| "direct".to_string(), |via| format!("via {via}"));
                write_line(
                    out,
                    &format!(
                        "session {} · agent {} · {via}",
                        field("session_id"),
                        field("agent")
                    ),
                )
                .await
            }
        }
    }

    /// One update, accumulating message text into `reply` when the result
    /// contract needs it.
    async fn update(
        &mut self,
        update: SessionUpdate,
        capture: bool,
        reply: &mut String,
    ) -> Result<()> {
        match self {
            Presenter::Text { transcript, .. } => {
                if capture {
                    // The journal keeps the raw update; the reply text is what
                    // the contract reads, and only the translation names it.
                    if let Some(SessionUpdateKind::MessageChunk { text, .. }) =
                        translate(update.clone())
                    {
                        reply.push_str(&text);
                    }
                }
                transcript.apply(update);
                Ok(())
            }
            Presenter::Ndjson { .. } => {
                let Some(kind) = translate(update) else {
                    return Ok(());
                };
                if capture && let SessionUpdateKind::MessageChunk { text, .. } = &kind {
                    reply.push_str(text);
                }
                // The `SessionUpdateKind`'s own `type` tag (e.g.
                // `message_chunk`) makes the line self-describing.
                self.ndjson(serde_json::to_value(kind).context("serialising NDJSON event")?)
                    .await
            }
            Presenter::Quiet { out, contracted } => {
                let Some(SessionUpdateKind::MessageChunk { text, .. }) = translate(update) else {
                    return Ok(());
                };
                if capture {
                    reply.push_str(&text);
                }
                if *contracted {
                    return Ok(());
                }
                out.write_all(text.as_bytes())
                    .await
                    .context("writing assistant text")
            }
        }
    }

    /// A permission request the policy answered.
    async fn permission(&mut self, prompt: &PermissionPrompt, decision: Decision) -> Result<()> {
        match self {
            Presenter::Ndjson { .. } => {
                self.ndjson(serde_json::json!({
                    "type": "permission",
                    "decision": decision.to_string(),
                    "title": prompt.title(),
                    "kind": prompt.kind(),
                }))
                .await
            }
            Presenter::Quiet { .. } => Ok(()),
            Presenter::Text { out, .. } => {
                write_line(out, &format!("  permission {decision}: {}", prompt.title())).await
            }
        }
    }

    /// The turn's updates have all arrived. The text transcript is written
    /// here, once, because a pipe cannot take a row back.
    async fn turn_settled(&mut self) -> Result<()> {
        if let Presenter::Text { out, transcript } = self {
            let mut rendered = Vec::new();
            bitrouter_tui::plain::write(&mut rendered, &transcript.unwritten())
                .context("rendering the transcript")?;
            out.write_all(&rendered)
                .await
                .context("writing the transcript")?;
        }
        Ok(())
    }

    /// The terminal result.
    async fn result<S: Serialize>(&mut self, line: &ResultLine<S>) -> Result<()> {
        match self {
            Presenter::Ndjson { .. } => {
                self.ndjson(serde_json::to_value(line).context("serialising NDJSON result")?)
                    .await
            }
            Presenter::Quiet { out, contracted } => {
                let text = match (&line.result, &line.raw) {
                    (Some(result), None) => {
                        serde_json::to_string(result).context("serialising the result")?
                    }
                    // A failed contract: the raw reply is the only answer there
                    // is, and quiet promised to print the answer.
                    (_, Some(raw)) => raw.clone(),
                    // No contract: the text already streamed; end the line.
                    (None, None) => String::new(),
                };
                if *contracted || text.is_empty() {
                    write_line(out, &text).await
                } else {
                    Ok(())
                }
            }
            Presenter::Text { out, .. } => {
                let stop = serde_json::to_value(&line.stop_reason)
                    .ok()
                    .and_then(|v| v.as_str().map(str::to_string))
                    .unwrap_or_default();
                write_line(out, &format!("[{stop}]")).await?;
                if let Some(result) = &line.result {
                    let pretty =
                        serde_json::to_string_pretty(result).context("serialising the result")?;
                    write_line(out, &format!("result: {pretty}")).await?;
                }
                if let Some(raw) = &line.raw {
                    write_line(out, &format!("raw: {raw}")).await?;
                }
                Ok(())
            }
        }
    }

    /// The `--no-wait` acknowledgement.
    async fn submitted(&mut self) -> Result<()> {
        match self {
            Presenter::Ndjson { .. } => {
                self.ndjson(serde_json::json!({ "type": "submitted" }))
                    .await
            }
            Presenter::Quiet { .. } => Ok(()),
            Presenter::Text { out, .. } => write_line(out, "[submitted]").await,
        }
    }
}

#[cfg(test)]
mod ndjson_presenter_tests {
    use super::{Presenter, PromptFormat};

    #[tokio::test]
    async fn every_event_is_versioned_and_sequenced() -> anyhow::Result<()> {
        let mut output = Vec::new();
        let mut presenter = Presenter::new(PromptFormat::Ndjson, false, &mut output);
        presenter.submitted().await?;
        presenter.error("failed", "broken").await?;
        drop(presenter);

        let lines: Vec<serde_json::Value> = String::from_utf8(output)?
            .lines()
            .map(serde_json::from_str)
            .collect::<Result<_, _>>()?;
        assert_eq!(lines.len(), 2);
        assert_eq!(lines[0]["version"], 1);
        assert_eq!(lines[0]["seq"], 0);
        assert_eq!(lines[1]["version"], 1);
        assert_eq!(lines[1]["seq"], 1);
        assert_eq!(lines[1]["type"], "error");
        Ok(())
    }
}

/// Write one line of text to `out`.
async fn write_line<W>(out: &mut W, line: &str) -> Result<()>
where
    W: AsyncWrite + Unpin,
{
    out.write_all(line.as_bytes())
        .await
        .context("writing to stdout")?;
    out.write_all(b"\n").await.context("writing to stdout")
}

/// List what a session with this agent offers, and tear it down.
///
/// A fresh session, deliberately: nothing can name a `bro chat` running
/// in another terminal, so this reports what a session with this agent *would*
/// offer rather than what one already open is offering. That is the same
/// limitation the row's `SessionBound` reach records.
///
/// No prompt is sent. The only thing waited on is the agent's own
/// `available_commands_update`, and `wait_ms` bounds that wait — an agent that
/// says nothing in time is reported as not having answered, which is a
/// different answer from an empty list.
pub async fn commands(
    ctx: SpawnContext<'_>,
    wait_ms: u64,
) -> Result<crate::actions::commands::CommandsReport> {
    let SpawnContext {
        source,
        mut config,
        agent_id,
        options,
        routing,
    } = ctx;
    let cloud_credentials = crate::cloud::StandaloneCloudCredentials::new();
    let routed = apply_routing_with_cloud_credentials(
        source,
        &mut config,
        agent_id,
        &routing,
        &cloud_credentials,
    )
    .await
    .map_err(anyhow::Error::new)?;

    // Before anything is launched. `ControlledSession` has no `Drop`, so a `?`
    // between the launch and `shutdown()` would leave the harness child
    // unreaped — and a `chat.commands` collision is exactly the error this
    // call exists to raise.
    let configured = crate::actions::session::prompt_commands(&config.chat)?;
    // The same binding `chat` opens. Without it the controller advertises no
    // route control, and this leaf would report `/route` as unavailable for
    // every agent while `chat` offers it — the drift the shared report exists
    // to prevent. `shutdown()` revokes it.
    let binding =
        LocalControllerBinding::open(source, &config, &routed, routing.base_url.is_some());

    let cwd = std::env::current_dir().context("resolving current directory")?;
    let mcp_servers = options.mcp_servers.clone();
    let mut session = launch_controlled(&config, agent_id, &routed, options, binding)
        .await
        .with_context(|| format!("launching acp session for agent '{agent_id}'"))?;
    // Subscribed before the session opens, so an agent that advertises its
    // commands immediately is not raced.
    let mut updates = session.client.subscribe_raw_updates();
    if let Err(error) = session.client.new_session(cwd, mcp_servers).await {
        let context = if bitrouter_sdk::acp::client::is_auth_required(&error) {
            unauthenticated_message(agent_id, session.client.auth_methods())
        } else {
            "opening the harness session".to_string()
        };
        session.shutdown().await;
        return Err(error.context(context));
    }

    let mut advertised = Vec::new();
    let mut received = false;
    let deadline = tokio::time::sleep(std::time::Duration::from_millis(wait_ms));
    tokio::pin!(deadline);
    loop {
        tokio::select! {
            // The last update wins: the list is a replacement, not a delta.
            update = updates.next() => match update {
                Some(agent_client_protocol::schema::v1::SessionUpdate::AvailableCommandsUpdate(
                    update,
                )) => {
                    advertised = update.available_commands;
                    received = true;
                }
                Some(_) => continue,
                // The stream ended; nothing more will arrive, so stop waiting.
                None => break,
            },
            () = &mut deadline => break,
        }
    }

    let report = crate::actions::commands::commands_report(
        &crate::actions::session::offered_commands(&session.client),
        &configured,
        &advertised,
        received,
    );
    session.shutdown().await;
    Ok(report)
}

// ── helpers ───────────────────────────────────────────────────────────────────

/// The OTel exporter for this process and, when one exists, the span recorder
/// bound to one session.
struct Observability {
    exporter: Option<Arc<bitrouter_telemetry::otel::OtelExporter>>,
    recorder: Option<Arc<bitrouter_telemetry::otel::acp::AcpSpanRecorder>>,
}

/// Build the exporter and the session's span recorder. `conversation_id` is
/// the span attribute every turn and tool span correlates on — the
/// harness-native ACP session id on the controller path.
async fn build_observability(
    config: &Config,
    agent_id: &str,
    conversation_id: &str,
    cloud_credentials: &crate::cloud::StandaloneCloudCredentials,
) -> Observability {
    // These surfaces never build an `App`, so they do not get the daemon's
    // ignored-config warnings for free — and they read the same telemetry
    // config, which is exactly what the skill documents. Without this, a stale
    // `plugins.bitrouter-observe` block is silent on `chat` and `acp prompt`
    // while being loud on `bro serve`. `acp_cli::serve` emits the same
    // set itself, since it never reaches this function. The subscriber is
    // already installed on all of these paths, so unlike the daemon these can
    // be emitted in place.
    for message in crate::assemble::ignored_config_warnings(config) {
        tracing::warn!("{message}");
    }
    let exporter =
        crate::assemble::build_otel_exporter_standalone_with_credentials(config, cloud_credentials)
            .await;
    let recorder = exporter.as_ref().map(|exporter| {
        Arc::new(bitrouter_telemetry::otel::acp::AcpSpanRecorder::new(
            exporter,
            agent_id,
            conversation_id,
        ))
    });
    Observability { exporter, recorder }
}

/// Emit an `execute_tool` span per completed tool call, read off the
/// translated update stream. Exporter-gated by its `recorder` argument:
/// without one there is nothing to emit to.
fn spawn_tool_spans(
    recorder: Arc<bitrouter_telemetry::otel::acp::AcpSpanRecorder>,
    mut updates: std::pin::Pin<Box<dyn futures::Stream<Item = SessionUpdateKind> + Send>>,
) {
    tokio::spawn(async move {
        use bitrouter_sdk::acp::translate::ToolStatus;
        while let Some(update) = updates.next().await {
            match update {
                SessionUpdateKind::ToolCall {
                    id, title, status, ..
                } => match status {
                    ToolStatus::Pending | ToolStatus::Running => {
                        recorder.tool_started(id, title);
                    }
                    ToolStatus::Ok => recorder.tool_finished(&id, true, Some(&title)),
                    ToolStatus::Failed => recorder.tool_finished(&id, false, Some(&title)),
                },
                SessionUpdateKind::ToolCallUpdate {
                    id, status, title, ..
                } => match status {
                    Some(ToolStatus::Ok) => {
                        recorder.tool_finished(&id, true, title.as_deref());
                    }
                    Some(ToolStatus::Failed) => {
                        recorder.tool_finished(&id, false, title.as_deref());
                    }
                    _ => {}
                },
                _ => {}
            }
        }
    });
}

#[cfg(test)]
mod standalone_cloud_credentials_tests {
    use std::sync::Arc;

    use bitrouter_providers::hosted::account::credentials::{Credentials, StoredCredential};
    use bitrouter_providers::hosted::account::manager::CredentialManager;
    use chrono::{Duration, Utc};
    use serde_json::json;
    use wiremock::matchers::{body_string_contains, method, path};
    use wiremock::{Mock, MockServer, ResponseTemplate};

    use crate::cloud::StandaloneCloudCredentials;

    #[tokio::test]
    async fn standalone_wiring_single_flights_routing_and_account_telemetry() -> anyhow::Result<()>
    {
        let server = MockServer::start().await;
        let origin = server.uri();
        Mock::given(method("GET"))
            .and(path("/.well-known/oauth-authorization-server"))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "issuer": origin,
                "device_authorization_endpoint": format!("{origin}/oauth/device_authorization"),
                "token_endpoint": format!("{origin}/oauth/token"),
            })))
            .mount(&server)
            .await;
        Mock::given(method("POST"))
            .and(path("/oauth/token"))
            .and(body_string_contains("grant_type=refresh_token"))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "access_token": "rotated-access",
                "token_type": "Bearer",
                "expires_in": 3600,
                "refresh_token": "rotated-refresh",
                "scope": "inference:invoke",
            })))
            .mount(&server)
            .await;

        let directory = tempfile::tempdir()?;
        let manager = Arc::new(CredentialManager::with_client(
            directory.path().join("account-credentials.json"),
            reqwest::Client::new(),
        ));
        manager
            .save(StoredCredential::from(Credentials {
                access_token: "stale-access".to_owned(),
                refresh_token: Some("original-refresh".to_owned()),
                expires_at: Utc::now() + Duration::seconds(10),
                refresh_token_expires_at: None,
                token_type: "Bearer".to_owned(),
                scope: "inference:invoke".to_owned(),
                client_id: "bitrouter-cli".to_owned(),
                authorization_server: origin.clone(),
                namespace_id: Some("ns-test".to_owned()),
                subject: None,
            }))
            .await?;

        let credentials = StandaloneCloudCredentials::with_manager(manager);
        let telemetry = credentials
            .telemetry_bearer(&origin)
            .await
            .ok_or_else(|| anyhow::anyhow!("account telemetry source was not built"))?;
        let (routing, bearer) =
            tokio::join!(credentials.routing_fallback(&origin), telemetry.bearer(),);
        assert_eq!(routing.as_deref(), Some("rotated-access"));
        assert_eq!(bearer.as_deref(), Some("rotated-access"));

        let requests = server
            .received_requests()
            .await
            .ok_or_else(|| anyhow::anyhow!("wiremock did not record requests"))?;
        let metadata_requests = requests
            .iter()
            .filter(|request| request.url.path() == "/.well-known/oauth-authorization-server")
            .count();
        let refresh_requests = requests
            .iter()
            .filter(|request| request.url.path() == "/oauth/token")
            .count();
        assert_eq!(metadata_requests, 1);
        assert_eq!(refresh_requests, 1);
        Ok(())
    }
}

/// Emit one telemetry record to stderr via tracing. Stdout must stay clean
/// (ACP JSON-RPC for `serve`, NDJSON for `prompt`), so telemetry goes to
/// `tracing::info!` which the acp CLI routes to stderr.
fn drain_telemetry_record(r: RequestCompleted) {
    tracing::info!(
        agent = %r.agent,
        stop_reason = %r.stop_reason,
        latency_ms = r.latency_ms,
        context_used = r.context.map(|c| c.used),
        context_size = r.context.map(|c| c.size),
        "acp turn completed"
    );
}

/// How to launch one harness process and drive it.
///
/// Lived in the SDK while `engine::Session` did the launching. It is the
/// app's now: every field is consumed by a different part of the shared
/// stack — `strip_inherited_env` by `AgentProcess`, `turn_timeout` by
/// `ClientOptions`, `mcp_servers` by `AcpClient::new_session` — so there is no
/// one SDK type it belongs to.
#[derive(Clone, Debug, Default)]
pub struct LaunchOptions {
    /// Inherited environment names to remove before applying the explicit
    /// transport and launch overlays. This lets isolated callers prevent
    /// ambient credentials from crossing into an agent process while still
    /// permitting a deliberately configured credential to win.
    pub strip_inherited_env: Vec<String>,
    /// Per-turn deadline. On elapse the agent is asked to cancel cooperatively
    /// (`session/cancel`); if it does not comply within the client's grace the
    /// turn errors.
    pub turn_timeout: Option<Duration>,
    /// MCP servers passed to the agent in `session/new` (`mcpServers`) — the
    /// caller's tool surface for the session.
    pub mcp_servers: Vec<agent_client_protocol::schema::v1::McpServer>,
    /// Whether this caller can run a `terminal` authentication method's login
    /// (ACP_AUTH_SPEC §6.2, §7.3).
    ///
    /// **False by default.** The canonical Code surface does not hand its
    /// terminal to an adapter, and headless or piped callers have no person to
    /// complete the login. Only a caller that implements a supported terminal
    /// handoff may set this true.
    pub terminal_auth: bool,
}

/// A completed ACP turn, as this process records it.
///
/// Also came from the SDK, where an `ExecutionHook` on the ACP pipeline was
/// the only thing that produced one. That pipeline routed to a single pinned
/// target its executor ignored, and nothing outside the SDK ever registered a
/// hook on it, so it went with the engine. The record survives because it is
/// what the span recorder and the per-turn log consume — rebuilt now from the
/// prompt round-trip itself, which is where the latency and stop reason were
/// visible all along.
#[derive(Debug, Clone)]
pub struct RequestCompleted {
    /// The agent name that handled the turn.
    pub agent: String,
    /// The stop reason rendered as a string (e.g. `"EndTurn"`, `"MaxTokens"`).
    pub stop_reason: String,
    /// Wall-clock latency for the turn in milliseconds.
    pub latency_ms: u64,
    /// Context-window occupancy as of the latest `UsageUpdate`, when the agent
    /// has reported one.
    pub context: Option<bitrouter_sdk::acp::telemetry::ContextUsage>,
}

/// Build [`LaunchOptions`] from the CLI flags shared by `serve` and `prompt`:
/// `--turn-timeout <secs>`.
pub fn launch_options(turn_timeout_secs: Option<u64>) -> LaunchOptions {
    LaunchOptions {
        turn_timeout: turn_timeout_secs.map(std::time::Duration::from_secs),
        ..Default::default()
    }
}

#[cfg(all(test, unix))]
mod evolution_acceptance;

#[cfg(test)]
mod auth_report_tests {
    use agent_client_protocol::schema::v1::{AuthMethod, AuthMethodAgent};

    use super::{LaunchOptions, unauthenticated_message};

    /// **A control that cannot act is absent** (ACP_AUTH_SPEC §6.2). Standard
    /// callers leave terminal authentication disabled.
    #[test]
    fn terminal_auth_is_off_for_standard_launches() {
        assert!(
            !LaunchOptions::default().terminal_auth,
            "the default must not claim a terminal this caller may not have"
        );
        assert!(
            !super::launch_options(None).terminal_auth,
            "the shared serve/prompt constructor must not claim it either"
        );
    }

    /// The reader is told which harness, and what it actually advertised.
    #[test]
    fn the_message_names_the_harness_and_what_it_offers() {
        let methods = vec![AuthMethod::Agent(
            AuthMethodAgent::new("oauth", "Sign in with Anthropic")
                .description("opens a browser".to_string()),
        )];
        let message = unauthenticated_message("claude-acp", &methods);
        assert!(message.contains("'claude-acp'"), "{message}");
        assert!(message.contains("Sign in with Anthropic"), "{message}");
        assert!(message.contains("opens a browser"), "{message}");
    }

    /// A harness advertising nothing says so, rather than naming an empty list
    /// or implying a control that is not there.
    #[test]
    fn advertising_nothing_is_reported_as_nothing() {
        let message = unauthenticated_message("pi-acp", &[]);
        assert!(
            message.contains("advertises no authentication method"),
            "{message}"
        );
    }

    /// **The two axes are never conflated** (ACP_AUTH_SPEC §6.4). This message
    /// is about the harness's own authentication; sending the reader to
    /// `providers login` or a route would be the wrong fix for it.
    #[test]
    fn the_message_never_mentions_the_routing_axis() {
        let methods = vec![AuthMethod::Agent(AuthMethodAgent::new("t", "Terminal"))];
        for message in [
            unauthenticated_message("claude-acp", &methods),
            unauthenticated_message("claude-acp", &[]),
        ] {
            // The product's own name contains "route"; strip it before looking
            // for the routing *concept*, or the check fails on "bitrouter".
            let prose = message.replace("bitrouter", "");
            for forbidden in ["providers login", "provider", "route", "--direct"] {
                assert!(
                    !prose.contains(forbidden),
                    "an agent-auth message must not name the routing axis \
                     ({forbidden:?}): {message}"
                );
            }
        }
    }
}

#[cfg(test)]
mod controller_tests {
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::time::Duration;

    use agent_client_protocol::schema::v1::Cost;
    use bitrouter_sdk::acp::controller::SessionCost;

    use super::{CachedSessionCost, attributed_cost, controller_identity};

    /// A daemon bridge that does not answer until released, and counts how
    /// often it was asked.
    struct StallingCost {
        release: tokio::sync::Notify,
        asked: AtomicUsize,
        figure: Option<Cost>,
    }

    #[async_trait::async_trait]
    impl SessionCost for StallingCost {
        async fn attributed_cost(&self, _session_id: &str) -> Option<Cost> {
            self.asked.fetch_add(1, Ordering::SeqCst);
            self.release.notified().await;
            self.figure.clone()
        }
    }

    async fn settled(cache: &CachedSessionCost, session_id: &str) -> Option<Cost> {
        let deadline = std::time::Instant::now() + Duration::from_secs(5);
        loop {
            let held = cache.cache.held(session_id);
            if held.is_some() || std::time::Instant::now() >= deadline {
                return held;
            }
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
    }

    /// The forward path never waits on the daemon: the first call returns at
    /// once with no figure, the refresh it started lands in the background,
    /// and the next call reads it. A burst of updates costs one lookup.
    #[tokio::test]
    async fn the_cached_cost_answers_without_awaiting_the_daemon() {
        let daemon = Arc::new(StallingCost {
            release: tokio::sync::Notify::new(),
            asked: AtomicUsize::new(0),
            figure: Some(Cost::new(0.42, "USD")),
        });
        let cache = CachedSessionCost::new(daemon.clone());

        let first = tokio::time::timeout(Duration::from_millis(200), cache.attributed_cost("s1"))
            .await
            .expect("the forward path must not block on the daemon");
        assert_eq!(first, None, "nothing measured yet is nothing, not zero");
        let second = tokio::time::timeout(Duration::from_millis(200), cache.attributed_cost("s1"))
            .await
            .expect("still not blocked while the refresh is in flight");
        assert_eq!(second, None);
        tokio::task::yield_now().await;
        assert_eq!(
            daemon.asked.load(Ordering::SeqCst),
            1,
            "one refresh in flight per session, however many updates arrive"
        );

        daemon.release.notify_one();
        assert_eq!(settled(&cache, "s1").await, Some(Cost::new(0.42, "USD")));
        assert_eq!(
            cache.attributed_cost("s1").await,
            Some(Cost::new(0.42, "USD")),
            "the figure the daemon confirmed is what the next update carries"
        );
    }

    /// A session the daemon never measured stays absent, and a refresh that
    /// yields nothing never erases a figure already held.
    #[tokio::test]
    async fn an_unmeasured_session_stays_absent_and_a_held_figure_survives() {
        let daemon = Arc::new(StallingCost {
            release: tokio::sync::Notify::new(),
            asked: AtomicUsize::new(0),
            figure: None,
        });
        let cache = CachedSessionCost::new(daemon.clone());
        assert_eq!(cache.attributed_cost("never").await, None);
        daemon.release.notify_one();
        tokio::time::sleep(Duration::from_millis(20)).await;
        assert_eq!(
            cache.attributed_cost("never").await,
            None,
            "no priced evidence is no figure — not $0.00"
        );

        cache
            .cache
            .settle_refresh("held", Some(Cost::new(0.25, "USD")));
        cache.cache.settle_refresh("held", None);
        assert_eq!(
            cache.cache.held("held"),
            Some(Cost::new(0.25, "USD")),
            "a stale measured figure understates; it is never dropped for nothing"
        );
    }

    /// The provenance marker is one spelling in two crates: what the
    /// controller writes is what the renderer reads.
    #[test]
    fn the_cost_marker_is_spelled_the_same_on_both_sides() {
        assert_eq!(
            bitrouter_tui::cost::COST_PROVENANCE_META_KEY,
            bitrouter_sdk::acp::controller::COST_PROVENANCE_META_KEY
        );
        assert_eq!(
            bitrouter_tui::cost::COST_PROVENANCE_ROUTER,
            bitrouter_sdk::acp::controller::COST_PROVENANCE_ROUTER
        );
    }

    /// C5: a session with no attributed traffic, or none that could be
    /// priced, renders nothing — never `$0.00`.
    #[test]
    fn unattributed_or_unpriced_sessions_carry_no_figure() {
        assert_eq!(attributed_cost(0, 0, 0), None);
        assert_eq!(attributed_cost(0, 3, 3), None);
    }

    /// Priced evidence becomes a USD figure; a partially priced session
    /// reports its priced floor rather than hiding it.
    #[test]
    fn priced_evidence_becomes_a_usd_figure() {
        assert_eq!(attributed_cost(420_000, 2, 0), Some(Cost::new(0.42, "USD")));
        assert_eq!(attributed_cost(250_000, 3, 2), Some(Cost::new(0.25, "USD")));
    }

    #[test]
    fn direct_maintained_adapter_keeps_exact_identity() {
        let identity = controller_identity(
            "codex-acp",
            "npx",
            &[
                "-y".to_string(),
                "@agentclientprotocol/codex-acp@1.10.0".to_string(),
            ],
            None,
        );
        assert_eq!(identity.harness_id, "codex-acp");
        assert_eq!(identity.adapter_package, "@agentclientprotocol/codex-acp");
        assert_eq!(identity.adapter_version, "1.10.0");
    }

    #[test]
    fn custom_adapter_identity_exposes_no_launch_command() {
        let identity = controller_identity(
            "private-agent",
            "/private/bin/agent-with-token-in-path",
            &["--secret-argument".to_string()],
            None,
        );
        assert_eq!(identity.harness_id, "private-agent");
        assert_eq!(identity.adapter_package, "configured-acp-adapter");
        assert_eq!(identity.adapter_version, "configured");
        let rendered = format!("{identity:?}");
        assert!(!rendered.contains("token"));
        assert!(!rendered.contains("secret"));
    }

    #[test]
    fn unpinned_catalog_invocation_is_not_reported_as_the_pin() {
        let identity = controller_identity(
            "codex-acp",
            "npx",
            &["@agentclientprotocol/codex-acp@1.6.0".to_string()],
            None,
        );
        assert_eq!(identity.harness_id, "codex-acp");
        assert_eq!(identity.adapter_package, "configured-acp-adapter");
        assert_eq!(identity.adapter_version, "configured");
    }
}