ai-crew-sync 0.7.10

//! `ai-crew-sync mcp proxy`: a local stdio MCP server that forwards every
//! tool call to the remote bus as **one** agent in **one** session.
//!
//! Why a proxy at all: the bus is stateless Streamable HTTP and any MCP client
//! can talk to it directly. What a direct connection cannot do is give each
//! *conversation* its own session when several windows open the same
//! repository with the same token — a header set once in a client config is
//! the same header in every window. A stdio server is started by the host
//! once per conversation (that is the MCP norm, and what Claude Code and
//! Codex do), so the process itself is the unit of isolation: it mints the
//! session label, sends it on every forwarded request, and keeps the
//! project/role metadata for exactly that window.
//!
//! The proxy is host-agnostic. What a host provides is used as an
//! enrichment, never required:
//!
//! - **Conversation identity**, in order: `--host-session` / `BUS_HOST_SESSION`
//!   (any host that can set per-window environment), `CLAUDE_CODE_SESSION_ID`
//!   (Claude Code exports it to MCP server processes; trusted only once the
//!   client names itself `claude-code`, because hosts started from a Claude
//!   Code terminal inherit it),
//!   `ANTIGRAVITY_CONVERSATION_ID` (where an Antigravity surface exports it
//!   to MCP servers), the conversation id a request carries in `_meta`
//!   (`antigravity.google/conversation_id` from Antigravity, `threadId` from
//!   Codex), and otherwise the process itself
//!   — a random id that lives as long as this instance. A known conversation
//!   id derives a **stable** session label, so a resumed conversation
//!   reconnects to the same session and a forked one gets a new one.
//! - **Start-of-session context** goes into the `initialize` result's
//!   `instructions`, which every MCP client hands to the model. No hook
//!   needed.
//! - **Presence** is kept by the proxy itself: a heartbeat on connect (with
//!   repo and branch read from the project directory), a periodic
//!   keep-alive, `idle` on exit.
//!
//! Two local tools, `configure_session` and `session_status`, never reach the
//! remote server's catalogue. They change metadata (project, role, channel)
//! and, within the same team, the profile; a team change needs a new
//! conversation, because switching credentials cannot erase what this
//! conversation has already seen. Every profile change is verified with
//! `whoami` before it is committed, in-flight calls to the old context are
//! cancelled rather than replayed, and claims or locks held by the old
//! identity are reported, never transferred.
//!
//! Stdout carries MCP framing only; everything else goes to stderr.

use std::{
    path::{Path, PathBuf},
    sync::{
        Arc,
        atomic::{AtomicUsize, Ordering},
    },
    time::Duration,
};

use anyhow::Context as _;
use rmcp::{
    ErrorData, ServerHandler, ServiceExt,
    model::{
        CallToolRequestParams, CallToolResponse, CallToolResult, ErrorCode, ListToolsResult,
        PaginatedRequestParams, ServerCapabilities, ServerConfig, Tool,
    },
    service::{ClientInitializeError, RequestContext, RoleServer, RunningService, ServiceError},
    transport::{
        StreamableHttpClientTransport,
        streamable_http_client::{StreamableHttpClientTransportConfig, StreamableHttpError},
    },
};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use tokio::sync::{Mutex, RwLock};
use tokio_util::sync::CancellationToken;

use crate::context::{self, Inputs, Resolved};

/// Prefix of a session label the proxy mints. Opaque on purpose: the label
/// is an address, `project`/`role` are the human-facing part.
pub const SESSION_PREFIX: &str = "s-";
/// Hex characters after the prefix. 128 bits: the label is an address that
/// partitions cursors, claims and locks, so two conversations colliding
/// would merge them silently. The session-label limit is 64 bytes, which
/// leaves room to spare.
const SESSION_HEX: usize = 32;

/// Presence lease the proxy keeps alive, and how often it renews it.
const PRESENCE_TTL_SECS: i64 = 900;
const KEEPALIVE_EVERY: Duration = Duration::from_secs(300);
/// Credential lifetime the proxy asks the bus for, in seconds, when
/// `BUS_SESSION_TTL_SECS` is set; unset takes the bus default. The bus
/// bounds it (60 s to 24 h).
const SESSION_TTL_ENV: &str = "BUS_SESSION_TTL_SECS";
/// How long before its expiry the credential is renewed, in seconds, when
/// `BUS_SESSION_RENEW_LEAD_SECS` is set; unset renews half-way through the
/// lifetime, so a transient failure has the other half to retry in.
const RENEW_LEAD_ENV: &str = "BUS_SESSION_RENEW_LEAD_SECS";
/// How long a context switch waits for in-flight calls to the old context
/// after cancelling them, before swapping anyway.
const DRAIN_TIMEOUT: Duration = Duration::from_secs(5);
/// How long the exit heartbeat may take; the host is waiting.
const EXIT_TIMEOUT: Duration = Duration::from_secs(3);
/// Attempts at opening a connection with a credential `resume_session` just
/// rotated, before the window gives up and goes off the bus. The old secret
/// is already refused, so one transient failure must not strand it.
const ADOPT_ATTEMPTS: u32 = 3;
/// How long a request waits for the project an MCP root may name, and how
/// long the client gets to answer `roots/list`.
const ROOTS_WAIT: Duration = Duration::from_secs(5);

pub const CONFIGURE_TOOL: &str = "configure_session";
pub const STATUS_TOOL: &str = "session_status";
/// What the bus serves in one `wait_for_updates` request.
const SERVER_WAIT_LIMIT_SECS: i64 = 55;
/// Each slice of a long wait: under the bus's limit with room to spare.
const SERVER_WAIT_SLICE_SECS: i64 = 50;
/// The longest wait the proxy covers in one call (30 minutes). A host must
/// allow a tool call that long (Codex: `tool_timeout_sec`).
const PROXY_MAX_WAIT_SECS: i64 = 1800;

type Remote = RunningService<rmcp::RoleClient, rmcp::model::ClientConfig>;

/// The `_meta` key under which Antigravity names the conversation on every
/// `tools/call`.
pub const ANTIGRAVITY_META_KEY: &str = "antigravity.google/conversation_id";

/// Where the conversation id came from, reported by `session_status`.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, JsonSchema)]
#[serde(rename_all = "kebab-case")]
pub enum Binding {
    /// `--host-session` or `BUS_HOST_SESSION`.
    Explicit,
    /// `CLAUDE_CODE_SESSION_ID` in the environment, with a client that named
    /// itself Claude Code.
    ClaudeCode,
    /// Google Antigravity's conversation id: `_meta["antigravity.google/
    /// conversation_id"]` on the first forwarded call, or
    /// `ANTIGRAVITY_CONVERSATION_ID` in the environment where a surface
    /// exports it to MCP servers.
    Antigravity,
    /// `_meta.threadId` on the first forwarded call.
    RequestMeta,
    /// No conversation id: this process is the conversation.
    Instance,
}

/// Command-line shape of the proxy.
#[derive(Clone, Debug, Default)]
pub struct ProxyOptions {
    pub inputs: Inputs,
    pub project: Option<String>,
    pub role: Option<String>,
    pub channel: Option<String>,
    /// Explicit conversation id.
    pub host_session: Option<String>,
    /// Where the proxy keeps its binding record; the config directory.
    pub state_dir: PathBuf,
}

/// Session label derived from a conversation id: stable for the same id,
/// unrelated to repository, role or pid. Shared with the resolver so a hook
/// of the same conversation lands on the same session without coordinating.
pub use crate::context::session_for_host as session_for;

/// Validate a discovery label the way the bus does, so `configure_session`
/// refuses what `heartbeat` would reject rather than storing it locally and
/// reporting a success the server never saw. Empty clears.
fn check_label(field: &str, raw: &str) -> Result<Option<String>, ErrorData> {
    crate::store::presence::normalize_label(field, raw)
        .map(|v| (!v.is_empty()).then_some(v))
        .map_err(|e| ErrorData::invalid_params(e.to_string(), None))
}

/// A non-blank environment variable, trimmed.
fn env_id(name: &str) -> Option<String> {
    std::env::var(name)
        .ok()
        .map(|s| s.trim().to_owned())
        .filter(|s| !s.is_empty())
}

/// The local path a `file://` URI names, percent-decoded. Anything else
/// (another scheme, a remote host) is not a directory this process can read.
fn path_of_file_uri(uri: &str) -> Option<PathBuf> {
    let rest = uri.strip_prefix("file://")?;
    let rest = rest.strip_prefix("localhost").unwrap_or(rest);
    if !rest.starts_with('/') {
        return None;
    }
    let bytes = rest.as_bytes();
    let mut out = Vec::with_capacity(bytes.len());
    let mut i = 0;
    while i < bytes.len() {
        if bytes[i] == b'%'
            && i + 2 < bytes.len()
            && let Ok(b) = u8::from_str_radix(rest.get(i + 1..i + 3)?, 16)
        {
            out.push(b);
            i += 3;
        } else {
            out.push(bytes[i]);
            i += 1;
        }
    }
    let path = PathBuf::from(String::from_utf8(out).ok()?);
    path.is_dir().then_some(path)
}

/// Whether an MCP client name is Claude Code's own (`claude-code`).
fn is_claude_code(name: &str) -> bool {
    name.trim().eq_ignore_ascii_case("claude-code")
}

fn random_session() -> String {
    let raw = crate::auth::generate_token();
    format!(
        "{SESSION_PREFIX}{}",
        &raw[crate::auth::TOKEN_PREFIX.len()..crate::auth::TOKEN_PREFIX.len() + SESSION_HEX]
    )
}

fn env_secs(name: &str) -> Option<i64> {
    std::env::var(name)
        .ok()
        .and_then(|v| v.trim().parse::<i64>().ok())
        .filter(|v| *v > 0)
}

/// The lifetime the proxy asks for on register, resume and renew; `None`
/// leaves it to the bus. Clamped to the bounds the bus applies, so the
/// renewal schedule is computed from the lifetime actually issued rather
/// than from a number the bus would have cut down.
fn requested_session_ttl() -> Option<i64> {
    env_secs(SESSION_TTL_ENV).map(|v| v.clamp(60, crate::auth::MAX_SESSION_TTL_SECS))
}

/// Seconds before expiry at which the credential is renewed, for a
/// credential of `lifetime` seconds: the override when set, else half the
/// lifetime, and always inside the lifetime so the deadline is never
/// already past on issue.
fn renewal_lead_secs(lifetime: i64, override_secs: Option<i64>) -> i64 {
    let lifetime = lifetime.max(2);
    override_secs.unwrap_or(lifetime / 2).clamp(1, lifetime - 1)
}

/// The credential this window proved itself with, when the bus issues them.
#[derive(Clone, Debug)]
pub struct SessionProof {
    /// The secret. Written only to the 0600 binding file, never to a tool
    /// result, a log or argv.
    pub token: String,
    pub session_id: String,
    pub epoch: i64,
    pub expires_at: String,
}

/// What one renewal attempt did.
enum Renewal {
    Renewed,
    /// The bus rejected the credential: revoked, expired or replaced.
    Refused,
    /// Transient: try again before the credential lapses.
    Retry,
    /// Nothing to renew, or the answer no longer applies to this context.
    Nothing,
}

/// The verified, connected context of this instance.
struct Connected {
    resolved: Resolved,
    agent: String,
    team: String,
    remote: Arc<Remote>,
    tools: Vec<Tool>,
    remote_instructions: Option<String>,
    proof: Option<SessionProof>,
    /// Cancelled when this context is replaced; forwarded calls race it.
    ct: CancellationToken,
}

struct State {
    /// Present once a profile resolved and verified. Absent means the proxy
    /// serves only its local tools and says why in `instructions`.
    connected: Option<Connected>,
    /// Why there is no connection, for the model.
    disconnected_reason: Option<String>,
    session: String,
    binding: Binding,
    host_id: Option<String>,
    /// True when `host_id` came from a request's `_meta` rather than from
    /// the environment or a flag: only then is a second id of the same kind
    /// proof that the host multiplexes conversations over this process.
    bound_by_request: bool,
    project: Option<String>,
    role: Option<String>,
    channel: Option<String>,
    /// Bumped on every context switch.
    generation: u64,
    /// A credential `resume_session` rotated whose new secret could not open
    /// a connection. The context it belonged to is retired, and this keeps
    /// the binding holding the new secret so a restart resumes with it.
    /// It is still this window's identity: read it through
    /// [`State::identity`], never through `connected` alone.
    stranded: Option<Identity>,
    /// The workspace an MCP root named, when the proxy's own directory had
    /// no project (#236). Used for `.acs.toml` and for repo/branch in place
    /// of the working directory, and only ever filled when nothing more
    /// explicit chose the project.
    roots_dir: Option<PathBuf>,
}

/// Who this window is on the bus: the agent, team and credential it speaks
/// as, whether or not a connection is open right now.
#[derive(Clone)]
struct Identity {
    profile: Option<String>,
    agent: String,
    team: String,
    mcp_url: String,
    proof: Option<SessionProof>,
}

impl State {
    /// This window's identity: the connected context, else the one a
    /// failed adoption retired. Every check of team, ownership or
    /// credential reads it here: a window that lost its connection is
    /// still bound to its team and still owns its binding, and a check
    /// that looked only at `connected` would treat it as nobody.
    fn identity(&self) -> Option<Identity> {
        match (&self.connected, &self.stranded) {
            (Some(c), _) => Some(Identity {
                profile: c.resolved.profile.clone(),
                agent: c.agent.clone(),
                team: c.team.clone(),
                mcp_url: c.resolved.mcp_url.clone(),
                proof: c.proof.clone(),
            }),
            (None, Some(s)) => Some(s.clone()),
            (None, None) => None,
        }
    }
}

/// Counts calls currently forwarded; decremented on drop so a request the
/// host abandons mid-flight still lets a context switch drain.
struct InFlight(Arc<AtomicUsize>);

impl InFlight {
    fn enter(counter: &Arc<AtomicUsize>) -> Self {
        counter.fetch_add(1, Ordering::SeqCst);
        Self(counter.clone())
    }
}

impl Drop for InFlight {
    fn drop(&mut self) {
        self.0.fetch_sub(1, Ordering::SeqCst);
    }
}

/// Tools whose call blocks on the bus waiting for teammates: while one is
/// in flight the model's turn is waiting, not working.
const WAITING_TOOLS: &[&str] = &["wait_for_updates", "ask_agent"];
/// A tool call is reported at most this often while the reported state is
/// unchanged. Short, because the bus may hold a newer observation this
/// proxy cannot see (a Stop hook's `idle`): a call more than this long after
/// the last report always goes out.
const ATTENTION_REPORT_EVERY: Duration = Duration::from_secs(15);
/// Least time between two attempts to flush a deferred call.
const FLUSH_RETRY: Duration = Duration::from_secs(2);

/// What this proxy has seen of its model's turn (#221). The host's tool
/// calls are the only evidence: a call proves a turn is running, a call to a
/// waiting tool proves it is blocked on the bus. Nothing here can see a turn
/// end; hooks report that where the host has them, so an observation older
/// than [`ATTENTION_REPORT_EVERY`] is never sent late as if it were current.
#[derive(Default)]
struct Attention {
    /// Host tool calls in flight.
    calls: usize,
    /// Of those, the ones blocked in a bus wait.
    waits: usize,
    /// The latest tool call not yet reported.
    unreported_call: Option<std::time::Instant>,
    /// What the bus last accepted from this proxy, and when.
    reported: Option<(&'static str, std::time::Instant)>,
}

/// One attention report to send. It is committed with [`Attention::sent`]
/// only once the bus took it, so a failed send is retried.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct AttentionReport {
    state: &'static str,
    /// The call this report covers, which also stamps `last_tool_at`.
    call: Option<std::time::Instant>,
}

impl AttentionReport {
    fn tool_activity(&self) -> bool {
        self.call.is_some()
    }

    /// The heartbeat fields for this report. A call reported late says how
    /// long ago it happened, so the bus can keep a newer observation (a Stop
    /// hook's idle) rather than let a late report overwrite it.
    fn fields(&self) -> Value {
        let mut v = json!({
            "refresh_only": true,
            "attention": self.state,
            "tool_activity": self.tool_activity(),
        });
        if self.state == "attending"
            && let Some(at) = self.call
        {
            v["attention_observed_ms_ago"] = json!(at.elapsed().as_millis() as i64);
        }
        v
    }
}

impl Attention {
    /// What to report now, if anything. `refresh` is the keepalive: it
    /// re-reports a wait still in flight so that observation stays fresh,
    /// and it never sends a call that has gone stale: once the model has
    /// been quiet that long, its turn may have ended, and saying
    /// `attending` would overwrite what a Stop hook observed.
    fn due(&mut self, refresh: bool) -> Option<AttentionReport> {
        let now = std::time::Instant::now();
        let recent = |at: std::time::Instant| now.duration_since(at) < ATTENTION_REPORT_EVERY * 2;
        if self.unreported_call.is_some_and(|at| !recent(at)) && self.calls == 0 {
            self.unreported_call = None;
        }
        let reported_age = self.reported.map(|(_, at)| now.duration_since(at));
        let (state, call) = if self.waits > 0 {
            let changed = self.reported.is_none_or(|(s, _)| s != "waiting");
            if !(changed || refresh) {
                return None;
            }
            ("waiting", self.unreported_call)
        } else if let Some(call) = self.unreported_call {
            let changed = self.reported.is_none_or(|(s, _)| s != "attending");
            let old = reported_age.is_none_or(|age| age >= ATTENTION_REPORT_EVERY);
            if !(changed || old || (refresh && self.calls > 0)) {
                return None;
            }
            ("attending", Some(call))
        } else if refresh && self.calls > 0 {
            // A long call still running: the turn is too.
            ("attending", None)
        } else {
            return None;
        };
        Some(AttentionReport { state, call })
    }

    /// When a call deferred by the throttle must be flushed, if one is
    /// waiting: as soon as the throttle allows, so the window does not read
    /// stale until the next keepalive.
    fn flush_at(&self) -> Option<std::time::Instant> {
        // A wait in flight is reported on its own terms (and re-reported by
        // the keepalive); nothing to flush meanwhile.
        if self.waits > 0 {
            return None;
        }
        self.unreported_call?;
        self.reported.map(|(_, at)| at + ATTENTION_REPORT_EVERY)
    }

    /// The bus accepted `report`.
    fn sent(&mut self, report: AttentionReport) {
        self.reported = Some((report.state, std::time::Instant::now()));
        // Only the call this report covered; a newer one is still pending.
        if self.unreported_call.is_some() && self.unreported_call == report.call {
            self.unreported_call = None;
        }
    }
}

/// One host tool call, counted for attention while it runs.
struct ToolTurn {
    attention: Arc<std::sync::Mutex<Attention>>,
    changed: Arc<tokio::sync::Notify>,
    waits: bool,
}

impl ToolTurn {
    fn enter(proxy: &Proxy, tool: &str) -> Self {
        let waits = WAITING_TOOLS.contains(&tool);
        if let Ok(mut a) = proxy.attention.lock() {
            a.calls += 1;
            a.waits += usize::from(waits);
            a.unreported_call = Some(std::time::Instant::now());
        }
        proxy.attention_changed.notify_one();
        Self {
            attention: proxy.attention.clone(),
            changed: proxy.attention_changed.clone(),
            waits,
        }
    }
}

impl Drop for ToolTurn {
    fn drop(&mut self) {
        if let Ok(mut a) = self.attention.lock() {
            a.calls = a.calls.saturating_sub(1);
            if self.waits {
                a.waits = a.waits.saturating_sub(1);
                // Back from a wait: the turn is working again.
                a.unreported_call = Some(std::time::Instant::now());
            }
        }
        // Also the trailing edge of a throttled burst: a call that ended
        // after the throttle window reports what it was not allowed to.
        self.changed.notify_one();
    }
}

#[derive(Clone)]
pub struct Proxy {
    state: Arc<RwLock<State>>,
    in_flight: Arc<AtomicUsize>,
    /// The model's turn as this proxy has seen it, and a ping for the
    /// keepalive to report a change.
    attention: Arc<std::sync::Mutex<Attention>>,
    attention_changed: Arc<tokio::sync::Notify>,
    /// Serialises context switches.
    switch: Arc<Mutex<()>>,
    /// Serialises heartbeats, from reading the labels to the bus's answer.
    /// The bus stores whatever labels arrive last, so a keepalive or
    /// attention report that read them before a `configure_session` and
    /// landed after it would put the old role back (#246).
    presence: Arc<Mutex<()>>,
    /// Pinged whenever the connected context changes, so the keepalive
    /// recomputes its renewal deadline at once instead of on its next
    /// presence tick: a credential established after startup, or after a
    /// refusal, must not wait five minutes for its first schedule.
    wake: Arc<tokio::sync::Notify>,
    opts: Arc<ProxyOptions>,
    project_dir: PathBuf,
    /// A `CLAUDE_CODE_SESSION_ID` from the environment that is not yet
    /// trusted. A host started from a Claude Code terminal inherits the
    /// parent's id, so the variable alone cannot tell a Claude Code child
    /// from Codex or Antigravity launched there (#234). Nothing is resumed
    /// until the client names itself; held across the connect, so a request
    /// racing the first one waits for the decision instead of bypassing it.
    pending_claude: Arc<Mutex<Option<String>>>,
    /// Held while the proxy asks the client for its roots after
    /// `initialized`, so the first requests wait for the project it finds
    /// instead of racing it. Bounded by [`ROOTS_WAIT`].
    roots_gate: Arc<Mutex<()>>,
    /// The gate, taken while answering `initialize` and handed to the
    /// discovery `initialized` starts. Taken there because the `initialize`
    /// answer precedes every later request of the client, whereas the
    /// `initialized` notification and the first `tools/list` are handled
    /// concurrently and the request could slip past a gate taken later.
    roots_hold: Arc<std::sync::Mutex<Option<tokio::sync::OwnedMutexGuard<()>>>>,
}

// -------------------------------------------------------------- local tools --

#[derive(Debug, Default, Deserialize, JsonSchema)]
pub struct ConfigureArgs {
    /// What this window does on the project: implementation, design,
    /// review, … One lower-case word. Changing it keeps the session and its
    /// cursors; it only changes how teammates find you.
    #[serde(default)]
    pub role: Option<String>,
    /// Logical project, usually the repository name. Also the channel this
    /// session posts to by default when one of that name exists.
    #[serde(default)]
    pub project: Option<String>,
    /// Channel to post to by default; overrides the project's.
    #[serde(default)]
    pub channel: Option<String>,
    /// Switch to another locally approved profile. Verified with whoami
    /// before anything changes; must stay within the same team.
    #[serde(default)]
    pub profile: Option<String>,
}

#[derive(Debug, Serialize, JsonSchema)]
pub struct Status {
    /// True when calls are being forwarded to the bus.
    pub connected: bool,
    /// Why not, when `connected` is false.
    pub error: Option<String>,
    /// Verified with whoami; never asserted.
    pub agent: Option<String>,
    pub team: Option<String>,
    /// Session label every forwarded call carries.
    pub session: String,
    /// `agent/session`: what a teammate puts in `to` to reach this window.
    pub address: Option<String>,
    pub project: Option<String>,
    pub role: Option<String>,
    pub channel: Option<String>,
    pub profile: Option<String>,
    /// Where this window's credential came from: the environment, a
    /// profile flag, the project's .acs.toml, or the user default. Never
    /// the credential itself.
    pub credential_from: Option<String>,
    pub binding: Binding,
    pub project_root: Option<String>,
    /// How `project_root` was found: `project-dir` (`--project-dir` /
    /// `BUS_PROJECT_DIR`), `working-directory` (searched upward from where
    /// the host started the proxy) or `mcp-roots` (a workspace the client
    /// listed in `roots/list`). Absent when there is no project root.
    pub project_root_from: Option<String>,
    pub bus: Option<String>,
    /// What this proxy last reported about the model's turn: `attending`
    /// (it saw a tool call), `waiting` (a bus wait is in flight), or absent
    /// when it has seen no call yet. The proxy cannot see a turn end; the
    /// host's hooks report `idle` where they exist, and list_sessions shows
    /// the combined observation.
    pub attention: Option<String>,
    /// Seconds since that report.
    pub attention_reported_seconds_ago: Option<u64>,
}

#[derive(Debug, Serialize, JsonSchema)]
pub struct ConfigureResult {
    pub status: Status,
    /// What the previous identity still holds, when the profile changed.
    /// Nothing is transferred: these expire by their own leases, or the
    /// previous identity releases them from its own window.
    pub previous: Option<PreviousIdentity>,
}

#[derive(Debug, Serialize, JsonSchema)]
pub struct PreviousIdentity {
    pub agent: String,
    pub team: String,
    pub session: String,
    pub open_claims: Vec<String>,
    pub held_locks: Vec<String>,
}

fn schema_of<T: JsonSchema>() -> Arc<rmcp::model::JsonObject> {
    let schema = schemars::schema_for!(T);
    match serde_json::to_value(schema) {
        Ok(Value::Object(map)) => Arc::new(map),
        _ => Arc::new(rmcp::model::JsonObject::new()),
    }
}

fn local_tools() -> Vec<Tool> {
    vec![
        Tool::new(
            CONFIGURE_TOOL,
            "Set how THIS window presents itself on the bus: role (implementation, design, \
             review, …), project and default channel. Metadata only — it never changes who \
             you are or your session id, so cursors, claims and locks stay yours. `profile` \
             switches to another locally approved credential of the same team after \
             verifying it; a different team needs a new conversation. Affects this window \
             only.",
            schema_of::<ConfigureArgs>(),
        )
        .with_title("Configure this session")
        .with_output_schema::<ConfigureResult>(),
        Tool::new(
            STATUS_TOOL,
            "Who this window is on the bus (verified agent and team), its session id and \
             address (`agent/session`, what teammates use to reach exactly this window), \
             project, role, default channel, and where its credential came from \
             (`credential_from`). Never returns credentials.",
            schema_of::<EmptyArgs>(),
        )
        .with_title("Session status")
        .with_output_schema::<Status>(),
    ]
}

#[derive(Debug, Default, Deserialize, JsonSchema)]
pub struct EmptyArgs {}

// --------------------------------------------------------------- connecting --

async fn connect_remote(
    url: &str,
    credential: &str,
    session: &str,
    epoch: Option<i64>,
) -> anyhow::Result<Remote> {
    let mut config = StreamableHttpClientTransportConfig::with_uri(url.to_owned());
    config.auth_header = Some(credential.to_owned());
    config.allow_stateless = true;
    config.custom_headers.insert(
        crate::auth::SESSION_HEADER.parse()?,
        session
            .parse()
            .context("session label is not a valid header value")?,
    );
    // Fencing is opt-in per connection: with it, a request from a process
    // that has been resumed away is refused instead of writing as the window
    // that replaced it.
    if let Some(epoch) = epoch {
        config.custom_headers.insert(
            crate::auth::EPOCH_HEADER.parse()?,
            epoch
                .to_string()
                .parse()
                .context("epoch is not a valid header value")?,
        );
    }
    let transport = StreamableHttpClientTransport::from_config(config);
    let remote = rmcp::model::ClientConfig::default()
        .serve(transport)
        .await
        .map_err(|e| {
            // `session_status` and every refusal while disconnected show
            // this text to the model: the bus's words when it refused, one
            // sentence when it could not be reached, the URL and the OS
            // error only in the log.
            tracing::warn!(error = %e, url, "could not open a connection to the bus");
            let lost = || "the connection failed before an answer came back".to_owned();
            let (why, rejected) = match &e {
                ClientInitializeError::JsonRpcError(data) => (data.message.to_string(), false),
                ClientInitializeError::TransportError { error, .. } => {
                    let rejected = matches!(
                        http_error_in(&*error.error),
                        Some(StreamableHttpError::AuthRequired(_))
                    );
                    let why = match refusal_in(&*error.error) {
                        Some(r) => r.text(),
                        None if rejected => "the bus rejected the credential".to_owned(),
                        None => lost(),
                    };
                    (why, rejected)
                }
                _ => (lost(), false),
            };
            let text = format!("could not connect to the bus: {why}");
            if rejected {
                anyhow::Error::new(Verdict::Unauthorized).context(text)
            } else {
                anyhow::anyhow!(text)
            }
        })?;
    // The handshake now carries the bus's real version (a 0.7.0-or-older
    // server identifies as `rmcp`, which says nothing). Skew is reported as
    // a hint, never as a refusal: nothing here decides that two versions
    // are incompatible — it only ends the search when something else fails.
    let peer_info = remote.peer_info();
    if let Some(si) = peer_info.as_ref().and_then(|i| i.server_info.as_ref()) {
        let ours = env!("CARGO_PKG_VERSION");
        if si.name == "ai-crew-sync" && si.version != ours {
            tracing::warn!(
                binary = ours,
                bus = %si.version,
                "this binary and the bus run different ai-crew-sync versions; \
                 if tools fail to load or calls are refused, align the two \
                 before debugging anything else"
            );
        } else if si.name != "ai-crew-sync" {
            tracing::debug!(
                server = %si.name,
                version = %si.version,
                "the bus did not identify an ai-crew-sync version (0.7.0 or older)"
            );
        }
    }
    Ok(remote)
}

/// The bus refused this window's bearer. Only the transport can say so: a
/// 401 never reaches the JSON-RPC layer, and a tool's own error is the bus
/// talking *to* the model, free to quote a session label, a lease or an id
/// that happens to spell "401". Matching that text once turned an ordinary
/// "held by joaquin" refusal into "your credential was revoked".
fn unauthorized(e: &ServiceError) -> bool {
    let ServiceError::TransportSend(sent) = e else {
        return false;
    };
    match http_error_in(&*sent.error) {
        Some(http) => matches!(http, StreamableHttpError::AuthRequired(_)),
        // A transport error that is not the reqwest one: fall back to the
        // wording rmcp gives a rejected bearer, still never a tool's text.
        None => sent.error.to_string().contains("Auth required"),
    }
}

/// The HTTP transport's own error inside a transport failure, found by
/// type along the source chain. Every classification of a failed request
/// (rejected bearer, HTTP refusal) reads it, never the rendered text.
fn http_error_in<'a>(
    root: &'a (dyn std::error::Error + 'static),
) -> Option<&'a StreamableHttpError<reqwest::Error>> {
    let mut cause = Some(root);
    while let Some(err) = cause {
        if let Some(http) = err.downcast_ref::<StreamableHttpError<reqwest::Error>>() {
            return Some(http);
        }
        cause = err.source();
    }
    None
}

/// The bus has no such tool. rmcp answers an unknown tool with
/// `invalid_params("tool not found")`, and a JSON-RPC layer without the
/// method with -32601. A refusal from a tool that exists is neither,
/// whatever it quotes: the bus's own "not found: …" errors share the code,
/// and a live-label conflict quotes the label, which a conversation id can
/// hash to `s-32601…`. Matching that text once kept a window label-only,
/// forwarding with the parent token, exactly when it had to fail closed.
fn no_such_tool(e: &ServiceError) -> bool {
    match e {
        ServiceError::McpError(err) => {
            err.code == ErrorCode::METHOD_NOT_FOUND
                || (err.code == ErrorCode::INVALID_PARAMS && err.message.trim() == "tool not found")
        }
        _ => false,
    }
}

/// What a failed call means, decided from the error's shape, never from its
/// wording. Carried inside the `anyhow` chain by [`call_remote`] so a caller
/// that only sees `anyhow::Error` reads the same fact.
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
enum Verdict {
    #[error("the bus rejected the credential")]
    Unauthorized,
    #[error("the bus has no such tool")]
    NoSuchTool,
}

fn verdict(e: &ServiceError) -> Option<Verdict> {
    if unauthorized(e) {
        Some(Verdict::Unauthorized)
    } else if no_such_tool(e) {
        Some(Verdict::NoSuchTool)
    } else {
        None
    }
}

/// Whether the bus refused a stored session credential because it expired,
/// as opposed to being revoked (401 "invalid or revoked") or replaced by a
/// newer process (409, stale epoch). The transport carries the bus's own
/// refusal text, which is the only place the distinction is visible.
fn credential_expired(e: &anyhow::Error) -> bool {
    format!("{e:#}").contains("session credential has expired")
}

fn verdict_of(e: &anyhow::Error) -> Option<Verdict> {
    e.chain().find_map(|c| c.downcast_ref::<Verdict>().copied())
}

/// The delivery ids a confirmation settled: the ones the bus committed
/// now plus the ones it says were already confirmed for this caller. A bus
/// too old to list ids answers with a count alone; then the count settles
/// everything only when it matches what was sent, as before.
fn settled_ids(reply: Option<&Value>, sent: &[String]) -> std::collections::HashSet<String> {
    let mut settled = std::collections::HashSet::new();
    let Some(reply) = reply else {
        return settled;
    };
    let listed = reply.get("confirmed_ids").is_some() || reply.get("already_confirmed").is_some();
    if listed {
        for key in ["confirmed_ids", "already_confirmed"] {
            if let Some(ids) = reply.get(key).and_then(|v| v.as_array()) {
                settled.extend(ids.iter().filter_map(|v| v.as_str()).map(str::to_owned));
            }
        }
    } else if reply.get("confirmed").and_then(|v| v.as_i64()) == Some(sent.len() as i64) {
        settled.extend(sent.iter().cloned());
    }
    settled
}

/// The bus answered over HTTP and refused the request before running it.
///
/// The bus's middleware (the body limit, the rate limit, a stale epoch, an
/// expired session credential) answers with a status and `{"error": "…"}`
/// written for the model. That body is not a JSON-RPC error, so rmcp does
/// not hand it back as one: it arrives as
/// `UnexpectedServerResponse("HTTP {status}: {body}")` inside the transport
/// error, where it looks like a lost connection unless read for its shape.
#[derive(Debug, PartialEq, Eq)]
struct Refusal {
    status: u16,
    /// The bus's own words, when the body was the bus's `{"error": …}`.
    said: Option<String>,
}

impl Refusal {
    fn text(&self) -> String {
        match &self.said {
            Some(said) => format!("the bus refused it before running it: {said}"),
            None => format!(
                "the bus refused it with HTTP {} before running it",
                self.status
            ),
        }
    }
}

/// The refusal inside a transport error, if the bus answered with one.
/// Only an answer that certainly ran nothing counts: the bus's own body at
/// any status, or a bare 4xx. A bare 5xx may come from a gateway that timed
/// out waiting for a call that did run, so it stays a lost connection.
fn refusal_in(root: &(dyn std::error::Error + 'static)) -> Option<Refusal> {
    let StreamableHttpError::UnexpectedServerResponse(msg) = http_error_in(root)? else {
        return None;
    };
    let rest = msg.strip_prefix("HTTP ")?;
    let (head, body) = rest.split_once(": ").unwrap_or((rest, ""));
    let status: u16 = head.split_whitespace().next()?.parse().ok()?;
    let said = serde_json::from_str::<Value>(body)
        .ok()
        .and_then(|v| v.get("error")?.as_str().map(str::to_owned));
    (said.is_some() || (400..500).contains(&status)).then_some(Refusal { status, said })
}

fn refusal(e: &ServiceError) -> Option<Refusal> {
    match e {
        ServiceError::TransportSend(sent) => refusal_in(&*sent.error),
        _ => None,
    }
}

/// What a failed call to the bus means, in words the model can use.
///
/// The bus writes its errors for the model, so an MCP error from it is its
/// own message — never `ServiceError`'s rendering of it (`"Mcp error:
/// -32602: …"`), which names a JSON-RPC code nobody downstream can act on.
/// An HTTP refusal is the bus's too, in its middleware's words. Anything
/// else is this proxy losing the bus: said once, without the transport's
/// internals (URLs, OS errors), which go to the log.
fn remote_error_text(e: &ServiceError) -> String {
    if let ServiceError::McpError(data) = e {
        return data.message.to_string();
    }
    tracing::warn!(error = %e, "the call to the bus failed in transport");
    match refusal(e) {
        Some(r) => r.text(),
        None => "the connection to the bus failed before an answer came back".to_owned(),
    }
}

async fn call_remote(remote: &Remote, name: &str, args: Value) -> anyhow::Result<Value> {
    let arguments: rmcp::model::JsonObject =
        serde_json::from_value(args).context("arguments must be an object")?;
    let result = remote
        .call_tool(CallToolRequestParams::new(name.to_owned()).with_arguments(arguments))
        .await
        .map_err(|e| {
            // The verdict is read from the error's shape and carried in the
            // chain; the text is only what a reader of the chain will see.
            let text = format!("{name}: {}", remote_error_text(&e));
            match verdict(&e) {
                Some(v) => anyhow::Error::new(v).context(text),
                None => anyhow::anyhow!(text),
            }
        })?;
    if result.is_error == Some(true) {
        // The tool's own words, not the Rust `Debug` of its content blocks.
        let said: Vec<String> = result
            .content
            .iter()
            .filter_map(|c| c.as_text().map(|t| t.text.clone()))
            .collect();
        anyhow::bail!("{name}: {}", said.join(" "));
    }
    Ok(result.structured_content.unwrap_or(Value::Null))
}

/// Resolve, connect with the session header, and verify who the token is.
/// Register this window for the first time. A server too old to know the
/// tool keeps the label-only connection it has always served; any other
/// failure is fatal, because continuing would forward with the parent token
/// under an asserted label while reporting a proven identity.
async fn register_new(remote: &Remote, session: &str) -> anyhow::Result<Option<SessionProof>> {
    let mut args = json!({ "session": session });
    if let Some(ttl) = requested_session_ttl() {
        args["ttl_seconds"] = json!(ttl);
    }
    match call_remote(remote, "register_session", args).await {
        Ok(v) => match v["session_token"].as_str() {
            Some(token) => Ok(Some(SessionProof {
                token: token.to_owned(),
                session_id: v["session_id"].as_str().unwrap_or_default().to_owned(),
                epoch: v["epoch"].as_i64().unwrap_or(1),
                expires_at: v["expires_at"].as_str().unwrap_or_default().to_owned(),
            })),
            None => anyhow::bail!(
                "the bus accepted register_session but returned no credential; refusing to \
                 continue with an asserted label while reporting a proven identity"
            ),
        },
        Err(e) => {
            let text = e.to_string();
            // Only "this server has no such tool" is a legacy bus. A refusal,
            // a database error or a dropped connection is not, and failing
            // closed is the point.
            if verdict_of(&e) == Some(Verdict::NoSuchTool) {
                tracing::warn!(
                    "this bus does not issue session credentials; continuing with the label only"
                );
                Ok(None)
            } else {
                Err(anyhow::anyhow!(
                    "could not register this window's session: {text}"
                ))
            }
        }
    }
}

/// Reconnect a window we already hold a credential for. Its own credential
/// is the proof, so no agent token is involved; a rejected resume means the
/// window was revoked, rotated or expired and the caller must register
/// afresh.
async fn resume_with(
    url: &str,
    prior: &SessionProof,
    session: &str,
) -> anyhow::Result<Option<SessionProof>> {
    let remote = connect_remote(url, &prior.token, session, Some(prior.epoch))
        .await
        .context("the stored session credential could not open a connection")?;
    let mut args = json!({});
    if let Some(ttl) = requested_session_ttl() {
        args["ttl_seconds"] = json!(ttl);
    }
    let outcome = call_remote(&remote, "resume_session", args).await;
    let _ = remote.cancel().await;
    let v =
        outcome.context("this window's session could not be resumed; it may have been revoked")?;
    let token = v["session_token"]
        .as_str()
        .context("resume_session returned no credential")?;
    Ok(Some(SessionProof {
        token: token.to_owned(),
        session_id: v["session_id"].as_str().unwrap_or_default().to_owned(),
        epoch: v["epoch"].as_i64().unwrap_or(1),
        expires_at: v["expires_at"].as_str().unwrap_or_default().to_owned(),
    }))
}

/// The rotation `resume_session` asked for through the proxy, with the
/// credential the window holds. The error is the one sentence the model
/// reads, classified once like a forwarded call's: whether the resume
/// happened, and what to do. The transport's detail goes to the log only.
async fn rotate(url: &str, prior: &SessionProof, session: &str) -> Result<SessionProof, String> {
    const KEPT: &str = "nothing was resumed and this window keeps the credential it holds";
    let rejected = || {
        "resume_session: the bus rejected this window's credential (revoked or rotated), so \
         nothing was resumed; restart this MCP server to reconnect the window"
            .to_owned()
    };
    let remote = connect_remote(url, &prior.token, session, Some(prior.epoch))
        .await
        .map_err(|e| {
            tracing::warn!(error = format!("{e:#}"), "resume_session could not connect");
            if verdict_of(&e) == Some(Verdict::Unauthorized) {
                rejected()
            } else {
                // `connect_remote` words its own error for the model.
                format!("resume_session: {e}; {KEPT}. Try again once the bus answers")
            }
        })?;
    let mut args = rmcp::model::JsonObject::new();
    if let Some(ttl) = requested_session_ttl() {
        args.insert("ttl_seconds".into(), json!(ttl));
    }
    let outcome = remote
        .call_tool(CallToolRequestParams::new("resume_session").with_arguments(args))
        .await;
    let _ = remote.cancel().await;
    let lost = |what: &str| {
        format!(
            "resume_session: {what}, so the resume may or may not have happened. If this \
             window's next call is refused, it did and the new credential is lost: close the \
             window with revoke_session (its label, from another window of this agent or \
             `ai-crew-sync client call revoke_session`) and restart this MCP server"
        )
    };
    let result = outcome.map_err(|e| {
        tracing::warn!(error = %e, "resume_session through the proxy failed");
        if unauthorized(&e) {
            rejected()
        } else if let ServiceError::McpError(data) = &e {
            format!("resume_session: {KEPT}: {}", data.message)
        } else if let Some(r) = refusal(&e) {
            format!("resume_session: {}; {KEPT}", r.text())
        } else {
            lost("the connection to the bus failed before an answer came back")
        }
    })?;
    if result.is_error == Some(true) {
        let said: Vec<String> = result
            .content
            .iter()
            .filter_map(|c| c.as_text().map(|t| t.text.clone()))
            .collect();
        return Err(format!("resume_session: {KEPT}: {}", said.join(" ")));
    }
    let v = result.structured_content.unwrap_or(Value::Null);
    let Some(token) = v["session_token"].as_str() else {
        tracing::warn!("resume_session answered without a credential");
        return Err(lost("the bus answered without a credential"));
    };
    Ok(SessionProof {
        token: token.to_owned(),
        session_id: v["session_id"].as_str().unwrap_or_default().to_owned(),
        epoch: v["epoch"].as_i64().unwrap_or(1),
        expires_at: v["expires_at"].as_str().unwrap_or_default().to_owned(),
    })
}

/// A connection on `proof` with the bus's tool list, or why not in words
/// the model can read.
async fn open_with(
    url: &str,
    proof: &SessionProof,
    session: &str,
) -> Result<(Remote, Vec<Tool>), String> {
    let remote = connect_remote(url, &proof.token, session, Some(proof.epoch))
        .await
        // `connect_remote` logged the transport's detail and words its own
        // error for the model.
        .map_err(|e| e.to_string())?;
    match remote.list_all_tools().await {
        Ok(tools) => Ok((remote, tools)),
        Err(e) => {
            let why = format!("could not list the bus's tools: {}", remote_error_text(&e));
            let _ = remote.cancel().await;
            Err(why)
        }
    }
}

async fn establish(
    inputs: &Inputs,
    session: &str,
    existing_proof: Option<SessionProof>,
) -> anyhow::Result<(
    Resolved,
    String,
    String,
    Remote,
    Vec<Tool>,
    Option<String>,
    Option<SessionProof>,
)> {
    let resolved = context::resolve(inputs)?;
    // Shadow warnings surface in the log (stderr): the host shows them with
    // the server's output, and MCP stdout stays protocol-clean.
    for w in &resolved.warnings {
        tracing::warn!("{w}");
    }
    // Nothing outranked anything, but the identity was not chosen for this
    // directory either: say so where the host shows the server's output.
    if resolved.source == context::Source::UserDefault {
        tracing::warn!(
            "no BUS_TOKEN, no BUS_PROFILE and no .acs.toml naming a profile here: this window \
             uses the user default profile '{}'. `ai-crew-sync context profile default --clear` removes it; a \
             project's .acs.toml selects its own",
            resolved.profile.as_deref().unwrap_or("?")
        );
    }
    // First connection: the agent token, with the label in a header, exactly
    // as any direct client would.
    // The same wording a forwarded 401 gets, so a window started with a
    // rotated token says what to do rather than "the bus did not accept the
    // credential". The bus refuses it at connect or at whoami. Provenance
    // names where the failing credential came from without revealing it.
    let rejected = || {
        anyhow::anyhow!(
            "the bus rejected this window's credential — it has been revoked or \
             rotated{}. The credential came from {}. Issue a new token \
             (`ai-crew-sync admin token issue --save`) or select another \
             approved profile",
            resolved
                .profile
                .as_deref()
                .map(|p| format!(" (profile '{p}')"))
                .unwrap_or_default(),
            resolved.credential_provenance()
        )
    };
    let remote = match connect_remote(&resolved.mcp_url, &resolved.token, session, None).await {
        Ok(remote) => remote,
        Err(e) if verdict_of(&e) == Some(Verdict::Unauthorized) => return Err(rejected()),
        Err(e) => return Err(e),
    };
    let me = match call_remote(&remote, "whoami", json!({})).await {
        Ok(me) => me,
        Err(e) => {
            let raw = e.to_string();
            let _ = remote.cancel().await;
            if verdict_of(&e) == Some(Verdict::Unauthorized) {
                return Err(rejected());
            }
            anyhow::bail!("the bus did not accept the credential: {raw}");
        }
    };
    let agent = me["agent"].as_str().unwrap_or_default().to_owned();
    let team = me["team"].as_str().unwrap_or_default().to_owned();
    if let Some((exp_team, exp_agent)) = &resolved.expected
        && (&agent != exp_agent || &team != exp_team)
    {
        let _ = remote.cancel().await;
        anyhow::bail!(
            "profile '{}' expects {exp_agent}@{exp_team} but the token authenticates as \
             {agent}@{team}; fix the profile or its token entry",
            resolved.profile.as_deref().unwrap_or("?")
        );
    }

    // Then upgrade: talk with a credential that *proves* which window this
    // is. Reconnecting an existing window resumes it with the credential
    // already on disk — the bus refuses to hand a live window to whoever
    // holds the agent token — and a window with no stored credential
    // registers a new one.
    let stored = existing_proof;
    let proof = match &stored {
        Some(prior) => match resume_with(&resolved.mcp_url, prior, session).await {
            Ok(proof) => proof,
            // Expired: the window was away longer than its credential lives
            // (closed, or the machine asleep). The bus says to register
            // again with the agent token, and the label — with everything
            // filed under it — is unchanged.
            Err(e) if credential_expired(&e) => {
                tracing::warn!(
                    "this window's stored session credential had expired; registering \
                     session '{session}' again with the agent token"
                );
                register_new(&remote, session).await?
            }
            // Revoked or rotated: the bus no longer knows this secret (a
            // `revoke_session`, a revoked parent token, a rotation this
            // file never saw, or a bus that lost its sessions). Retrying
            // the same secret can only fail again, so the agent token —
            // already verified above — asks for the label. The bus decides:
            // a label that is revoked or dead is registered again, as
            // `register_session` documents, and one that is still live (the
            // process that rotated it holds the new secret) is refused, so
            // nothing is taken back from a window that can still speak for
            // itself.
            Err(e) if verdict_of(&e) == Some(Verdict::Unauthorized) => {
                tracing::warn!(
                    "the bus rejected this window's stored session credential (revoked or \
                     rotated); registering session '{session}' again with the agent token"
                );
                register_new(&remote, session).await?
            }
            // A stale epoch (409): another process resumed this window
            // after the stored credential was written. Fail closed.
            Err(e) => return Err(e),
        },
        None => register_new(&remote, session).await?,
    };

    let (remote, tools, instructions) = match &proof {
        Some(proof) => {
            let _ = remote.cancel().await;
            let remote =
                connect_remote(&resolved.mcp_url, &proof.token, session, Some(proof.epoch))
                    .await
                    .context("the session credential could not open a connection")?;
            let tools = remote.list_all_tools().await.map_err(|e| {
                anyhow::anyhow!("could not list the bus's tools: {}", remote_error_text(&e))
            })?;
            let instructions = remote.peer_info().and_then(|i| i.instructions.clone());
            (remote, tools, instructions)
        }
        None => {
            let tools = remote.list_all_tools().await.map_err(|e| {
                anyhow::anyhow!("could not list the bus's tools: {}", remote_error_text(&e))
            })?;
            let instructions = remote.peer_info().and_then(|i| i.instructions.clone());
            (remote, tools, instructions)
        }
    };
    Ok((resolved, agent, team, remote, tools, instructions, proof))
}

/// Repository and branch of the project directory, for presence. Best
/// effort: a directory that is not a checkout simply reports neither.
fn git_place(dir: &Path) -> (Option<String>, Option<String>) {
    let run = |args: &[&str]| {
        std::process::Command::new("git")
            .args(args)
            .current_dir(dir)
            .output()
            .ok()
            .filter(|o| o.status.success())
            .map(|o| String::from_utf8_lossy(&o.stdout).trim().to_owned())
            .filter(|s| !s.is_empty())
    };
    let repo = run(&["config", "--get", "remote.origin.url"]).map(|url| {
        let trimmed = url.trim_end_matches(".git");
        let tail: Vec<&str> = trimmed.rsplit(['/', ':']).take(2).collect();
        if tail.len() == 2 {
            format!("{}/{}", tail[1], tail[0])
        } else {
            trimmed.to_owned()
        }
    });
    let branch = run(&["branch", "--show-current"]);
    (repo, branch)
}

impl Proxy {
    /// Resolve the conversation id, connect and verify. Never fails: a proxy
    /// that cannot connect still serves its local tools and explains itself.
    pub async fn start(opts: ProxyOptions) -> Self {
        let project_dir = opts
            .inputs
            .project_dir
            .clone()
            .or_else(|| std::env::var_os("CLAUDE_PROJECT_DIR").map(PathBuf::from))
            .or_else(|| std::env::current_dir().ok())
            .unwrap_or_else(|| PathBuf::from("."));

        let explicit = opts
            .host_session
            .as_deref()
            .map(str::trim)
            .filter(|s| !s.is_empty())
            .map(str::to_owned);
        let claude = env_id("CLAUDE_CODE_SESSION_ID");
        let (host_id, binding) = match (&explicit, &claude) {
            (Some(id), _) => (Some(id.clone()), Binding::Explicit),
            // Deferred: see `pending_claude`. Until then the window is not
            // bound and not connected.
            (None, Some(_)) => (None, Binding::Instance),
            (None, None) => match env_id("ANTIGRAVITY_CONVERSATION_ID") {
                Some(id) => (Some(id), Binding::Antigravity),
                None => (None, Binding::Instance),
            },
        };
        let pending = explicit.is_none().then_some(claude).flatten();
        let session = match &host_id {
            Some(id) => session_for(id),
            None => random_session(),
        };
        let deferred = pending.is_some();

        let proxy = Self {
            state: Arc::new(RwLock::new(State {
                connected: None,
                disconnected_reason: deferred.then(|| {
                    "waiting for the MCP client to identify itself before binding the \
                     conversation"
                        .to_owned()
                }),
                session,
                binding,
                host_id,
                bound_by_request: false,
                project: opts.project.clone(),
                role: opts.role.clone(),
                channel: opts.channel.clone(),
                generation: 0,
                stranded: None,
                roots_dir: None,
            })),
            in_flight: Arc::new(AtomicUsize::new(0)),
            attention: Arc::new(std::sync::Mutex::new(Attention::default())),
            attention_changed: Arc::new(tokio::sync::Notify::new()),
            switch: Arc::new(Mutex::new(())),
            presence: Arc::new(Mutex::new(())),
            wake: Arc::new(tokio::sync::Notify::new()),
            opts: Arc::new(opts),
            project_dir,
            pending_claude: Arc::new(Mutex::new(pending)),
            roots_gate: Arc::new(Mutex::new(())),
            roots_hold: Arc::new(std::sync::Mutex::new(None)),
        };
        if !deferred {
            proxy.connect_initial().await;
        }
        proxy
    }

    /// The first connection of this window. Never fails: a proxy that cannot
    /// connect still serves its local tools and explains itself.
    async fn connect_initial(&self) {
        let inputs = self.inputs(&*self.state.read().await);
        if let Err(e) = self.connect_with(&inputs, true).await {
            tracing::warn!(error = %e, "proxy started without a bus connection");
            self.state.write().await.disconnected_reason = Some(format!("{e:#}"));
        }
    }

    /// Decide a deferred Claude Code binding once the client has named
    /// itself, then connect. Only Claude Code's own client gets the
    /// conversation the variable names; any other host started under it is
    /// a conversation of its own, so the parent's session, epoch and binding
    /// record are never touched (#234). A no-op after the first call.
    async fn admit(&self, client: Option<&rmcp::model::Implementation>) {
        let mut pending = self.pending_claude.lock().await;
        let Some(id) = pending.take() else { return };
        let name = client.map(|c| c.name.as_str()).unwrap_or_default();
        {
            let mut st = self.state.write().await;
            if is_claude_code(name) {
                st.session = session_for(&id);
                st.host_id = Some(id);
                st.binding = Binding::ClaudeCode;
            } else {
                tracing::info!(
                    client = name,
                    "CLAUDE_CODE_SESSION_ID is set but the client is not Claude Code; it was \
                     inherited from a parent window and is ignored"
                );
                if let Some(ag) = env_id("ANTIGRAVITY_CONVERSATION_ID") {
                    st.session = session_for(&ag);
                    st.host_id = Some(ag);
                    st.binding = Binding::Antigravity;
                }
            }
            st.disconnected_reason = None;
        }
        self.connect_initial().await;
    }

    /// Wait, bounded, for a `roots/list` discovery in flight.
    async fn await_roots(&self) {
        let _ = tokio::time::timeout(ROOTS_WAIT * 2, self.roots_gate.lock()).await;
    }

    /// What the resolver is given: the command line's inputs, with the
    /// project directory an MCP root named when there is one.
    fn inputs(&self, st: &State) -> Inputs {
        let mut inputs = self.opts.inputs.clone();
        if let Some(dir) = &st.roots_dir {
            inputs.project_dir = Some(dir.clone());
        }
        inputs
    }

    /// Find the project through the client's MCP roots when the proxy's own
    /// directory has none (#236): a globally installed host plugin starts
    /// the proxy in the plugin's directory, far from the workspace. Never
    /// overrides `--project-dir`, a profile chosen by flag or environment,
    /// or a `.acs.toml` the working directory already found.
    // Roots are deprecated by SEP-2577, but Antigravity still advertises
    // and answers them, and it is the host this exists for.
    #[allow(deprecated)]
    async fn discover_roots(&self, peer: &rmcp::Peer<RoleServer>) {
        if self.opts.inputs.project_dir.is_some() || self.opts.inputs.profile.is_some() {
            return;
        }
        let start = std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."));
        if matches!(context::find_project(&start), Ok(Some(_))) {
            return;
        }
        let roots = match tokio::time::timeout(ROOTS_WAIT, peer.list_roots()).await {
            Ok(Ok(r)) => r.roots,
            Ok(Err(e)) => {
                tracing::debug!(error = %e, "the client did not answer roots/list");
                return;
            }
            Err(_) => {
                tracing::debug!("the client did not answer roots/list in time");
                return;
            }
        };
        let Some(dir) = roots
            .iter()
            .filter_map(|r| path_of_file_uri(&r.uri))
            .find(|p| matches!(context::find_project(p), Ok(Some(_))))
        else {
            return;
        };
        tracing::info!(root = %dir.display(), "project found through the client's MCP roots");
        if let Err(e) = self.adopt_project_dir(dir).await {
            tracing::warn!(error = %e, "the project an MCP root names could not be used");
            return;
        }
        // The catalogue may have gone from local tools only to the bus's.
        let _ = peer.notify_tool_list_changed().await;
    }

    /// Use `dir` as the project directory. A window already connected as the
    /// profile the project names only picks up its labels; otherwise it
    /// connects (or switches, within the team) through the usual path.
    async fn adopt_project_dir(&self, dir: PathBuf) -> anyhow::Result<()> {
        let (inputs, same_profile, connected) = {
            let mut st = self.state.write().await;
            let previous = st.roots_dir.replace(dir);
            let inputs = self.inputs(&st);
            let resolved = match context::resolve(&inputs) {
                Ok(r) => r,
                Err(e) => {
                    st.roots_dir = previous;
                    return Err(e);
                }
            };
            // The same credential, not just the same profile: the project's
            // `key` or name can pick another token entry of that profile.
            let same = st.connected.as_ref().is_some_and(|c| {
                c.resolved.profile == resolved.profile
                    && c.resolved.token == resolved.token
                    && c.resolved.mcp_url == resolved.mcp_url
            });
            if same {
                if st.project.is_none() {
                    st.project = resolved.project.clone();
                }
                if st.channel.is_none() {
                    st.channel = resolved.channel.clone();
                }
                if let Some(c) = st.connected.as_mut() {
                    c.resolved.project_root = resolved.project_root.clone();
                }
            }
            (inputs, same, st.connected.is_some())
        };
        if !same_profile {
            // Connected under another profile (the user default) means a
            // switch: that credential belongs to the identity left behind.
            if let Err(e) = self.connect_with(&inputs, !connected).await {
                self.state.write().await.roots_dir = None;
                return Err(e);
            }
            return Ok(());
        }
        self.heartbeat("active").await;
        self.write_binding().await;
        Ok(())
    }

    /// Connect a context and make it current. Verifies before it touches
    /// state; on failure the previous context, if any, stays.
    /// `reuse_proof` is true when this is the *same* window reconnecting, so
    /// the credential on disk is its own and a resume is right. A profile
    /// switch passes false: that credential belongs to the identity being
    /// left behind, and resuming it would keep speaking as them.
    async fn connect_with(
        &self,
        inputs: &Inputs,
        reuse_proof: bool,
    ) -> anyhow::Result<Option<PreviousIdentity>> {
        let _guard = self.switch.lock().await;
        let (session, binding_key) = {
            let st = self.state.read().await;
            (
                st.session.clone(),
                st.host_id.clone().unwrap_or_else(|| st.session.clone()),
            )
        };
        // A credential already on disk for this conversation means this is a
        // reconnect: resume that window with its own proof rather than
        // asking the bus to hand it over. Only when the identity is
        // unchanged — see `reuse_proof`.
        let existing = reuse_proof
            .then(|| context::read_binding(&self.opts.state_dir, &binding_key))
            .flatten()
            .and_then(|b| match (b.session_token, b.session_id, b.epoch) {
                (Some(token), Some(session_id), Some(epoch)) if !token.is_empty() => {
                    Some(SessionProof {
                        token,
                        session_id,
                        epoch,
                        expires_at: b.expires_at.unwrap_or_default(),
                    })
                }
                _ => None,
            });
        let (resolved, agent, team, remote, tools, instructions, proof) =
            establish(inputs, &session, existing).await?;

        // A team switch would let one conversation carry another team's
        // transcript into this one. The credential was verified and is
        // dropped unused. Checked against the window's identity, not its
        // connection: a window off the bus after a failed adoption has the
        // same transcript and is bound to the same team.
        {
            let st = self.state.read().await;
            if let Some(old) = st.identity()
                && old.team != team
            {
                let _ = remote.cancel().await;
                anyhow::bail!(
                    "this conversation is bound to team '{}'; the profile '{}' belongs to team \
                     '{team}'. Switching teams inside a conversation is not allowed — the \
                     transcript already holds '{}' material. Start a new conversation with \
                     that profile instead",
                    old.team,
                    resolved.profile.as_deref().unwrap_or("?"),
                    old.team
                );
            }
        }

        // Project and channel defaults from the project file, unless the
        // caller set them explicitly.
        {
            let mut st = self.state.write().await;
            if st.project.is_none() {
                st.project = resolved.project.clone();
            }
            if st.channel.is_none() {
                st.channel = resolved.channel.clone();
            }
        }

        // Barrier: cancel the old context's in-flight calls, wait for them to
        // leave, then swap. Nothing is replayed.
        let previous = {
            let old = {
                let mut st = self.state.write().await;
                st.connected.take()
            };
            match old {
                Some(old) => {
                    old.ct.cancel();
                    let started = std::time::Instant::now();
                    while self.in_flight.load(Ordering::SeqCst) > 0
                        && started.elapsed() < DRAIN_TIMEOUT
                    {
                        tokio::time::sleep(Duration::from_millis(20)).await;
                    }
                    let held = report_holdings(&old.remote, &old.agent, &old.team, &session).await;
                    // This window is no longer that identity, so its session
                    // is closed rather than left live for nobody: a session
                    // with no process behind it blocks its own label and
                    // keeps a credential valid for a day. Claims and locks
                    // are NOT transferred — they stay with the old identity
                    // and expire on their leases, which is what `held`
                    // reports back to the caller.
                    if old.proof.is_some() {
                        let _ = tokio::time::timeout(
                            EXIT_TIMEOUT,
                            call_remote(&old.remote, "revoke_session", json!({})),
                        )
                        .await;
                    }
                    // The old identity goes quiet in its own name.
                    let _ = tokio::time::timeout(
                        EXIT_TIMEOUT,
                        call_remote(
                            &old.remote,
                            "heartbeat",
                            json!({"status": "idle", "ttl_seconds": 30}),
                        ),
                    )
                    .await;
                    close_remote(old.remote).await;
                    Some(held)
                }
                None => None,
            }
        };

        let connected = Connected {
            resolved,
            agent,
            team,
            remote: Arc::new(remote),
            tools,
            remote_instructions: instructions,
            proof,
            ct: CancellationToken::new(),
        };
        {
            let mut st = self.state.write().await;
            st.connected = Some(connected);
            st.disconnected_reason = None;
            st.stranded = None;
            st.generation += 1;
        }
        // A stored permit, not a broadcast: a keepalive that is not waiting
        // at this instant still sees the change on its next select.
        self.wake.notify_one();
        self.heartbeat("active").await;
        self.write_binding().await;
        Ok(previous)
    }

    /// Presence for this window: status, repo/branch from the checkout, and
    /// the discovery labels. Best effort.
    async fn heartbeat(&self, status: &str) {
        self.publish_presence(json!({"status": status})).await;
    }

    /// Keep this window's presence alive without saying anything about its
    /// status: a live process is not news about the model (#221). Carries
    /// the model's attention when something new was seen, or re-reports a
    /// wait still in flight.
    async fn refresh_presence(&self) {
        let due = self.attention.lock().ok().and_then(|mut a| a.due(true));
        let args = match due {
            Some(report) => report.fields(),
            None => json!({"refresh_only": true}),
        };
        if self.publish_presence(args).await
            && let Some(report) = due
            && let Ok(mut a) = self.attention.lock()
        {
            a.sent(report);
        }
    }

    /// Report attention now if something new was seen; nothing otherwise.
    async fn report_attention(&self) {
        let due = self.attention.lock().ok().and_then(|mut a| a.due(false));
        let Some(report) = due else { return };
        if self.publish_presence(report.fields()).await
            && let Ok(mut a) = self.attention.lock()
        {
            a.sent(report);
        }
    }

    /// One heartbeat with `fields`, plus repo/branch and this window's
    /// labels. True when the bus accepted it.
    async fn publish_presence(&self, fields: Value) -> bool {
        let _ordered = self.presence.lock().await;
        let (remote, project, role, dir) = {
            let st = self.state.read().await;
            let Some(c) = &st.connected else { return false };
            let dir = st
                .roots_dir
                .clone()
                .unwrap_or_else(|| self.project_dir.clone());
            (c.remote.clone(), st.project.clone(), st.role.clone(), dir)
        };
        let (repo, branch) = git_place(&dir);
        let mut args = fields;
        args["ttl_seconds"] = json!(PRESENCE_TTL_SECS);
        if let Some(r) = repo {
            args["repo"] = Value::String(r);
        }
        if let Some(b) = branch {
            args["branch"] = Value::String(b);
        }
        // Omitted keeps, "" clears: send exactly what this window knows.
        args["project"] = Value::String(project.unwrap_or_default());
        args["role"] = Value::String(role.unwrap_or_default());
        match call_remote(&remote, "heartbeat", args).await {
            Ok(_) => true,
            Err(e) => {
                tracing::warn!(error = %e, "heartbeat failed");
                false
            }
        }
    }

    /// Record this instance's binding so lifecycle hooks of the same
    /// conversation can find the session and labels. Keyed by the
    /// conversation id when there is one (hooks know it), by the session
    /// otherwise (nothing else can look it up, but `session_status` can
    /// still say where it is).
    async fn write_binding(&self) {
        let st = self.state.read().await;
        // Keyed by the conversation id when the host gives one — that is what
        // a hook of the same conversation can look up — and by the session
        // label otherwise, where nothing else can find it anyway.
        let key = st.host_id.clone().unwrap_or_else(|| st.session.clone());
        // The connected context, else the one a failed adoption retired:
        // its secret is the only one the bus still accepts for this window.
        let identity = st.identity();
        let (profile, agent, team, mcp_url, proof) = match identity {
            Some(i) => (
                i.profile,
                Some(i.agent),
                Some(i.team),
                Some(i.mcp_url),
                i.proof,
            ),
            None => (None, None, None, None, None),
        };
        // The credential goes in here, which is why the file is 0600 inside a
        // 0700 directory and why `context hook` is the only thing that reads
        // it. It never reaches a tool result, a log or argv.
        let record = json!({
            "host_id_present": st.host_id.is_some(),
            "binding": st.binding,
            "session": st.session,
            "project": st.project,
            "role": st.role,
            "channel": st.channel,
            "profile": profile,
            "agent": agent,
            "team": team,
            "mcp_url": mcp_url,
            "session_token": proof.as_ref().map(|p| p.token.clone()),
            "session_id": proof.as_ref().map(|p| p.session_id.clone()),
            // Hooks send this epoch, so they are fenced with their proxy
            // rather than fencing it: a hook never bumps it.
            "epoch": proof.as_ref().map(|p| p.epoch),
            "expires_at": proof.as_ref().map(|p| p.expires_at.clone()),
            "proxy_pid": std::process::id(),
            "updated_at": chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true),
        });
        drop(st);
        // Under the same lock as every other writer of this record: a
        // renewal stamping the expiry checks ownership and writes inside
        // that lock, so this replacement cannot slip in between.
        let path = context::binding_path(&self.opts.state_dir, &key);
        let text = record.to_string();
        if let Err(e) = under_config_lock(self.opts.state_dir.clone(), move || {
            context::write_binding_file(&path, &text)
        })
        .await
        {
            tracing::warn!(error = %e, "could not write the session binding");
        }
    }

    async fn status(&self) -> Status {
        let reported = self.attention.lock().ok().and_then(|a| a.reported);
        let st = self.state.read().await;
        let c = st.connected.as_ref();
        Status {
            connected: c.is_some(),
            error: st.disconnected_reason.clone(),
            agent: c.map(|c| c.agent.clone()),
            team: c.map(|c| c.team.clone()),
            session: st.session.clone(),
            address: c.map(|c| format!("{}/{}", c.agent, st.session)),
            project: st.project.clone(),
            role: st.role.clone(),
            channel: st.channel.clone(),
            profile: c.and_then(|c| c.resolved.profile.clone()),
            credential_from: c.map(|c| c.resolved.credential_provenance()),
            binding: st.binding,
            project_root: c
                .and_then(|c| c.resolved.project_root.as_ref())
                .map(|p| p.display().to_string()),
            project_root_from: c.and_then(|c| c.resolved.project_root.as_ref()).map(|_| {
                if st.roots_dir.is_some() {
                    "mcp-roots"
                } else if self.opts.inputs.project_dir.is_some() {
                    "project-dir"
                } else {
                    "working-directory"
                }
                .to_owned()
            }),
            bus: c.map(|c| c.resolved.mcp_url.clone()),
            attention: reported.map(|(s, _)| s.to_owned()),
            attention_reported_seconds_ago: reported.map(|(_, at)| at.elapsed().as_secs()),
        }
    }

    async fn configure(&self, args: ConfigureArgs) -> anyhow::Result<ConfigureResult> {
        // Labels first, before anything is committed anywhere.
        let mut previous = None;
        // Validated like the server validates them, and staged rather than
        // committed: a call that also switches profile must leave the
        // previous context *entirely* intact when the switch fails, labels
        // included.
        let staged_role = match args.role {
            Some(v) => Some(check_label("role", &v).map_err(|e| anyhow::anyhow!("{}", e.message))?),
            None => None,
        };
        let staged_project = match args.project {
            Some(v) => {
                Some(check_label("project", &v).map_err(|e| anyhow::anyhow!("{}", e.message))?)
            }
            None => None,
        };
        let staged_channel = match args.channel {
            Some(v) => {
                Some(check_label("channel", &v).map_err(|e| anyhow::anyhow!("{}", e.message))?)
            }
            None => None,
        };

        if let Some(profile) = args
            .profile
            .map(|p| p.trim().to_owned())
            .filter(|p| !p.is_empty())
        {
            let mut inputs = self.inputs(&*self.state.read().await);
            inputs.profile = Some(profile);
            // A profile switch is a switch of credentials: explicit ones from
            // the environment would otherwise win and make the call a no-op.
            inputs.explicit_token = None;
            inputs.explicit_url = None;
            previous = self.connect_with(&inputs, false).await?;
        }
        // Only now, with the switch (if any) verified and committed.
        {
            let mut st = self.state.write().await;
            if let Some(role) = staged_role {
                st.role = role;
            }
            if let Some(project) = staged_project {
                st.project = project;
            }
            if let Some(channel) = staged_channel {
                st.channel = channel;
            }
        }
        self.heartbeat("active").await;
        self.write_binding().await;
        Ok(ConfigureResult {
            status: self.status().await,
            previous,
        })
    }

    /// Bind to the conversation a request says it belongs to. First id seen
    /// becomes the binding when there was none; a different id later means
    /// the host multiplexes conversations over one process, which this proxy
    /// does not support and says so rather than mixing them.
    async fn observe_meta(&self, meta: &rmcp::model::RequestMetaObject) -> Result<(), ErrorData> {
        let id_at = |key: &str| {
            meta.0
                .0
                .get(key)
                .and_then(Value::as_str)
                .map(str::trim)
                .filter(|s| !s.is_empty())
        };
        // Antigravity names the conversation under its own key on every
        // tools/call (verified with agy 1.4.3, which does not export it to
        // MCP servers' environment); Codex sends threadId.
        let (thread, source) = match id_at(ANTIGRAVITY_META_KEY) {
            Some(id) => (id, Binding::Antigravity),
            None => match id_at("threadId").or_else(|| id_at("sessionId")) {
                Some(id) => (id, Binding::RequestMeta),
                None => return Ok(()),
            },
        };
        let (bound, current, by_request) = {
            let st = self.state.read().await;
            (st.host_id.clone(), st.binding, st.bound_by_request)
        };
        match bound {
            Some(id) if id == thread => Ok(()),
            Some(_) if by_request && current == source => Err(ErrorData::invalid_request(
                "this proxy instance is bound to another conversation; a second one is using \
                 the same MCP process, which is not supported. Configure the host to start \
                 one `ai-crew-sync mcp proxy` per conversation",
                None,
            )),
            // Bound by environment or flag: the request's id is informative
            // only, the operator's binding wins.
            Some(_) => Ok(()),
            None => self.rebind(thread.to_owned(), source).await.map_err(|e| {
                ErrorData::internal_error(format!("could not bind the conversation: {e:#}"), None)
            }),
        }
    }

    /// Adopt a conversation id discovered on the wire: derive the stable
    /// session and reconnect so every forwarded call, this one included,
    /// carries it.
    async fn rebind(&self, host_id: String, source: Binding) -> anyhow::Result<()> {
        // Staged, not committed: if the new session cannot connect, the
        // previous one must keep serving. Advertising the new identity over
        // the old connection is how a window ends up reporting one session
        // and writing as another.
        let previous = {
            let st = self.state.read().await;
            (
                st.host_id.clone(),
                st.binding,
                st.session.clone(),
                st.bound_by_request,
            )
        };
        {
            let mut st = self.state.write().await;
            st.host_id = Some(host_id.clone());
            st.binding = source;
            st.bound_by_request = true;
            st.session = session_for(&host_id);
        }
        // The profile this window speaks as, connected or not: a window off
        // the bus reconnects as itself, not as the startup default.
        let inputs = {
            let st = self.state.read().await;
            match st.identity() {
                Some(id) => {
                    let mut i = self.inputs(&st);
                    i.profile = id.profile;
                    i
                }
                None => self.inputs(&st),
            }
        };
        match self.connect_with(&inputs, true).await {
            Ok(_) => Ok(()),
            Err(e) => {
                let mut st = self.state.write().await;
                (st.host_id, st.binding, st.session, st.bound_by_request) = previous;
                st.disconnected_reason = Some(format!("{e:#}"));
                Err(e)
            }
        }
    }

    /// Forward one call to the connected context, racing the host's own
    /// cancellation and the context's replacement.
    /// Spool the references a fetch returned, fsync, then confirm them.
    ///
    /// Everything here is best effort in one direction only: a reference
    /// that cannot be written to disk is **not** confirmed, so the bus keeps
    /// offering it. The model still sees it in this turn — it is in the
    /// result either way — but nothing claims durability that does not
    /// exist.
    async fn spool_and_confirm(
        &self,
        result: CallToolResult,
        host_ct: CancellationToken,
    ) -> CallToolResult {
        let Some(structured) = result.structured_content.clone() else {
            return result;
        };
        let session = self.state.read().await.session.clone();
        let path = crate::spool::spool_path(&self.opts.state_dir, &session);

        let mut entries: Vec<crate::spool::Entry> = structured
            .get("references")
            .and_then(|v| v.as_array())
            .map(|refs| {
                refs.iter()
                    .filter_map(|r| {
                        Some(crate::spool::Entry {
                            delivery_id: r.get("delivery_id")?.as_str()?.to_owned(),
                            message_id: r.get("message_id")?.as_str()?.to_owned(),
                            conversation_id: r
                                .get("conversation_id")
                                .and_then(|v| v.as_str())
                                .unwrap_or_default()
                                .to_owned(),
                            seq: r.get("seq").and_then(|v| v.as_i64()).unwrap_or(0),
                            from_address: r
                                .get("from_address")
                                .and_then(|v| v.as_str())
                                .unwrap_or_default()
                                .to_owned(),
                            created_at: r
                                .get("created_at")
                                .and_then(|v| v.as_str())
                                .unwrap_or_default()
                                .to_owned(),
                            confirmed: false,
                            spooled_at: chrono::Utc::now().to_rfc3339(),
                        })
                    })
                    .collect::<Vec<_>>()
            })
            .unwrap_or_default();

        // Anything a previous process spooled and never confirmed goes in
        // the same confirmation: that is what an interrupted delivery looks
        // like from here.
        let mut held = crate::spool::read(&path);
        let spooled = match crate::spool::append(&path, &entries) {
            Ok(written) => written,
            Err(e) => {
                tracing::warn!(error = %e, "could not spool inbox references; not confirming");
                return result;
            }
        };
        held.extend(spooled.iter().cloned());
        entries.clear();
        let to_confirm = crate::spool::unconfirmed(&held);
        if to_confirm.is_empty() {
            return result;
        }

        let mut params = rmcp::model::JsonObject::new();
        params.insert(
            "delivery_ids".into(),
            Value::Array(
                to_confirm
                    .iter()
                    .map(|id| Value::String(id.clone()))
                    .collect(),
            ),
        );
        let confirm =
            CallToolRequestParams::new("confirm_inbox_delivery".to_string()).with_arguments(params);
        match self.forward(confirm, host_ct).await {
            Ok(confirmation) => {
                // The bus says which ids it committed now and which of ours
                // it had already: both are settled. An id in neither (a
                // stale epoch, somebody else's reference) stays in the
                // spool as the only evidence that it is still owed. The
                // reply the caller gets is the fetch, never this one: the
                // confirmation is the proxy's business.
                let settled = settled_ids(confirmation.structured_content.as_ref(), &to_confirm);
                let sent: std::collections::HashSet<&String> = to_confirm.iter().collect();
                let mut left = 0usize;
                for entry in held.iter_mut() {
                    if sent.contains(&entry.delivery_id) {
                        if settled.contains(&entry.delivery_id) {
                            entry.confirmed = true;
                        } else {
                            left += 1;
                        }
                    }
                }
                if left > 0 {
                    tracing::warn!(
                        sent = to_confirm.len(),
                        left,
                        "the bus did not settle every reference sent; keeping the rest in \
                         the spool"
                    );
                }
                if let Err(e) = crate::spool::rewrite(&path, &held) {
                    // The bus has the truth; this only costs a repeated
                    // confirmation next time, which the bus answers with
                    // `already_confirmed`.
                    tracing::warn!(error = %e, "could not compact the inbox spool");
                }
            }
            Err(e) => tracing::warn!(error = %e, "could not confirm inbox delivery"),
        }
        result
    }

    async fn forward(
        &self,
        request: CallToolRequestParams,
        host_ct: CancellationToken,
    ) -> Result<CallToolResult, ErrorData> {
        self.forward_bound(request, host_ct, None).await
    }

    /// `forward`, refusing to run unless the connection is still the one of
    /// `bound` (a generation). A call that spans several requests (a long
    /// wait) passes the generation it started under, checked under the same
    /// lock that hands out the connection, so it can never carry on as an
    /// identity the window switched to midway.
    async fn forward_bound(
        &self,
        request: CallToolRequestParams,
        host_ct: CancellationToken,
        bound: Option<u64>,
    ) -> Result<CallToolResult, ErrorData> {
        let (remote, ct, generation, _guard) = {
            let st = self.state.read().await;
            if let Some(started) = bound
                && st.generation != started
            {
                return Err(ErrorData::invalid_request(
                    format!(
                        "{} was cancelled: this window switched credentials while the call was \
                         in flight (generation {started}). Nothing was replayed; call again if \
                         it is still wanted, as the new identity",
                        request.name
                    ),
                    None,
                ));
            }
            let Some(c) = &st.connected else {
                return Err(ErrorData::invalid_request(
                    format!(
                        "not connected to the bus: {}. Call {CONFIGURE_TOOL} with an approved \
                         profile, or fix the local configuration and start a new conversation",
                        st.disconnected_reason
                            .as_deref()
                            .unwrap_or("no profile resolved")
                    ),
                    None,
                ));
            };
            (
                c.remote.clone(),
                c.ct.clone(),
                st.generation,
                InFlight::enter(&self.in_flight),
            )
        };
        let name = request.name.to_string();
        let profile = {
            let st = self.state.read().await;
            st.connected
                .as_ref()
                .and_then(|c| c.resolved.profile.clone())
        };
        let outcome = tokio::select! {
            r = remote.call_tool(request) => r.map_err(|e| {
                // A rejected bearer says "Auth required" and nothing about
                // what to do. This is what a revoked or rotated token looks
                // like from here; see `unauthorized` for what it is not.
                if unauthorized(&e) {
                    self.mark_unauthorized(&profile, generation);
                    ErrorData::invalid_request(
                        format!(
                            "{name}: the bus rejected this window's credential — it has been \
                             revoked or rotated{}. Issue a new token (`ai-crew-sync admin \
                             token issue --save`) and call {CONFIGURE_TOOL} with an approved \
                             profile; nothing was sent",
                            profile
                                .as_deref()
                                .map(|p| format!(" (profile '{p}')"))
                                .unwrap_or_default()
                        ),
                        None,
                    )
                } else if let ServiceError::McpError(data) = e {
                    // The bus wrote this for the model: same code, same
                    // words, same data, exactly as a direct call gets it.
                    data
                } else if let Some(r) = refusal(&e) {
                    // The bus answered and ran nothing: its middleware's
                    // words, and the caller knows the call did not happen.
                    tracing::warn!(error = %e, tool = %name, "the bus refused a forwarded call");
                    ErrorData::invalid_request(format!("{name}: {}", r.text()), None)
                } else {
                    // This proxy lost the bus. One classification, and the
                    // fact the caller needs to act safely: whether the call
                    // ran is unknown. The transport's detail goes to the log.
                    tracing::warn!(error = %e, tool = %name, "a forwarded call failed in transport");
                    ErrorData::internal_error(
                        format!(
                            "{name} could not reach the bus: the connection failed before an \
                             answer came back, so the call may or may not have run. Check \
                             before repeating anything that is not safe to repeat."
                        ),
                        None,
                    )
                }
            }),
            _ = ct.cancelled() => Err(ErrorData::invalid_request(
                format!(
                    "{name} was cancelled: this window switched credentials while the call was \
                     in flight (generation {generation}). Nothing was replayed; call again if \
                     it is still wanted, as the new identity"
                ),
                None,
            )),
            _ = host_ct.cancelled() => Err(ErrorData::invalid_request(
                format!("{name} was cancelled by the client"),
                None,
            )),
        };
        // The bus answers this proxy's client on its own negotiated
        // revision, so a deserialized result may carry no `resultType`.
        // Protocol `2026-07-28` requires it on every result the host
        // reads, and absent-means-complete holds only for earlier
        // revisions: a call the bus finished is marked complete here.
        outcome.map(|mut result| {
            result
                .result_type
                .get_or_insert(rmcp::model::ResultType::COMPLETE);
            result
        })
    }

    /// `wait_for_updates` with a `timeout_seconds` above what the bus serves
    /// in one request: wait on the bus in slices until something happens or
    /// the whole time has passed, and answer once. Up to the bus's own limit
    /// the call is forwarded unchanged.
    async fn long_wait(
        &self,
        request: CallToolRequestParams,
        host_ct: CancellationToken,
    ) -> Result<CallToolResult, ErrorData> {
        let requested = request
            .arguments
            .as_ref()
            .and_then(|a| a.get("timeout_seconds"))
            .and_then(Value::as_i64);
        let Some(total) = requested.filter(|t| *t > SERVER_WAIT_LIMIT_SECS) else {
            return self.forward(request, host_ct).await;
        };
        // The whole wait belongs to the identity it started under.
        let started_under = self.state.read().await.generation;
        // Each slice resumes where the previous one stopped, so a task, lock
        // or note change between two slices is replayed, not lost (#211).
        let mut since = request
            .arguments
            .as_ref()
            .and_then(|a| a.get("since"))
            .and_then(Value::as_str)
            .map(str::to_owned);
        let deadline = std::time::Instant::now()
            + std::time::Duration::from_secs(total.min(PROXY_MAX_WAIT_SECS) as u64);
        loop {
            let left = deadline
                .saturating_duration_since(std::time::Instant::now())
                .as_secs() as i64;
            let mut slice = request.clone();
            let args = slice.arguments.get_or_insert_with(Default::default);
            args.insert(
                "timeout_seconds".into(),
                json!(left.clamp(5, SERVER_WAIT_SLICE_SECS)),
            );
            if let Some(position) = &since {
                args.insert("since".into(), json!(position));
            }
            let result = self
                .forward_bound(slice, host_ct.clone(), Some(started_under))
                .await?;
            let woke = result
                .structured_content
                .as_ref()
                .and_then(|v| v["woke"].as_bool())
                .unwrap_or(true);
            if woke || left <= SERVER_WAIT_SLICE_SECS || host_ct.is_cancelled() {
                return Ok(result);
            }
            since = result
                .structured_content
                .as_ref()
                .and_then(|v| v["resume_from"].as_str().map(str::to_owned))
                .or(since);
        }
    }

    /// Record that the bus refused this window's credential, so
    /// `session_status` and the next `initialize` say so instead of
    /// claiming a healthy connection. Non-blocking: a busy lock means the
    /// next call reports it. Only for the context the refused call ran
    /// in: a call that was still in flight on a connection this window has
    /// since replaced (a resume, a profile switch) says nothing about the
    /// credential it holds now.
    fn mark_unauthorized(&self, profile: &Option<String>, generation: u64) {
        if let Ok(mut st) = self.state.try_write()
            && st.generation == generation
        {
            st.disconnected_reason = Some(format!(
                "the bus rejected the credential{} (revoked or rotated)",
                profile
                    .as_deref()
                    .map(|p| format!(" of profile '{p}'"))
                    .unwrap_or_default()
            ));
        }
    }

    /// `resume_session` asked for through this window.
    ///
    /// Forwarded as is, the bus rotates the secret and bumps the epoch of
    /// the credential this proxy sends, hands the new secret to the model,
    /// and every later call — this proxy's, its hooks', its next restart's —
    /// presents the old one and is refused. So the proxy runs the resume
    /// with the credential it holds, adopts what comes back exactly as a
    /// reconnect does (new connection, binding rewritten), and answers
    /// with the credential's description, never its secret. `None` when
    /// this window holds no credential (a bus that issues none, or no
    /// connection): the call is forwarded and the bus answers it. The error
    /// is the sentence the model reads; the detail behind it is logged.
    async fn resume_in_place(&self) -> Option<Result<Value, String>> {
        let _guard = self.switch.lock().await;
        let (url, proof, session) = {
            let st = self.state.read().await;
            let c = st.connected.as_ref()?;
            let proof = c.proof.clone()?;
            (c.resolved.mcp_url.clone(), proof, st.session.clone())
        };
        Some(self.adopt_resumed(&url, &proof, &session).await)
    }

    async fn adopt_resumed(
        &self,
        url: &str,
        prior: &SessionProof,
        session: &str,
    ) -> Result<Value, String> {
        let proof = rotate(url, prior, session).await?;
        // From here the bus refuses the old secret. A connection on the new
        // one gets a few tries: one transient failure must not cost the
        // window the bus.
        let mut attempt = 1;
        let opened = loop {
            match open_with(url, &proof, session).await {
                Ok(opened) => break Ok(opened),
                Err(why) if attempt < ADOPT_ATTEMPTS => {
                    tracing::warn!(
                        attempt,
                        "the resumed credential could not open a connection ({why}); retrying"
                    );
                    tokio::time::sleep(Duration::from_millis(250 * 4u64.pow(attempt - 1))).await;
                    attempt += 1;
                }
                Err(why) => break Err(why),
            }
        };
        let old = {
            let mut st = self.state.write().await;
            let Some(c) = st.connected.take() else {
                return Err(
                    "resume_session: the session was resumed, but this window lost its \
                     connection meanwhile; restart this MCP server to reconnect it"
                        .to_owned(),
                );
            };
            match opened {
                Ok((remote, tools)) => {
                    let instructions = remote.peer_info().and_then(|i| i.instructions.clone());
                    st.connected = Some(Connected {
                        resolved: c.resolved.clone(),
                        agent: c.agent.clone(),
                        team: c.team.clone(),
                        remote: Arc::new(remote),
                        tools,
                        remote_instructions: instructions,
                        proof: Some(proof.clone()),
                        ct: CancellationToken::new(),
                    });
                    st.disconnected_reason = None;
                }
                Err(why) => {
                    // The old secret is refused and the new one opens no
                    // connection: serving on the old connection would only
                    // collect refusals while reporting a healthy window. It
                    // is retired, and the new credential stays in the
                    // binding, where a restart resumes with it.
                    st.stranded = Some(Identity {
                        profile: c.resolved.profile.clone(),
                        agent: c.agent.clone(),
                        team: c.team.clone(),
                        mcp_url: c.resolved.mcp_url.clone(),
                        proof: Some(proof.clone()),
                    });
                    let said = format!(
                        "the session was resumed and the bus rotated this window's credential, \
                         but the new one could not open a connection ({why}); it is saved, and \
                         this window is off the bus until this MCP server is restarted, which \
                         reconnects with it"
                    );
                    st.disconnected_reason = Some(said.clone());
                    st.generation += 1;
                    drop(st);
                    c.ct.cancel();
                    close_remote(c.remote).await;
                    self.wake.notify_one();
                    self.write_binding().await;
                    return Err(format!("resume_session: {said}"));
                }
            }
            st.generation += 1;
            c
        };
        // Calls still in flight on the old connection are refused by the
        // bus from now on; cancel them rather than let them report a
        // rejected credential the window no longer has.
        old.ct.cancel();
        close_remote(old.remote).await;
        self.wake.notify_one();
        self.write_binding().await;
        let expires_in_seconds = chrono::DateTime::parse_from_rfc3339(&proof.expires_at)
            .map(|t| (t.with_timezone(&chrono::Utc) - chrono::Utc::now()).num_seconds())
            .unwrap_or(0)
            .max(0);
        // The bus's own shape, without the secret: the proxy holds it.
        let described = crate::model::SessionCredential {
            session_token: None,
            session_id: proof.session_id,
            session: session.to_owned(),
            address: format!("{}/{session}", old.agent),
            epoch: proof.epoch,
            expires_at: proof.expires_at,
            expires_in_seconds,
        };
        serde_json::to_value(described).map_err(|e| {
            tracing::warn!(error = %e, "could not describe the resumed credential");
            "resume_session: the session was resumed and this window holds the new \
             credential, but its description could not be built; session_status shows it"
                .to_owned()
        })
    }

    /// The channel this window posts to when a message names none: its own
    /// `channel`, else the channel named after its project.
    async fn default_channel(&self) -> Option<String> {
        let st = self.state.read().await;
        st.channel.clone().or_else(|| st.project.clone())
    }

    fn instructions(&self, st: &State) -> String {
        let mut lines = Vec::new();
        match &st.connected {
            Some(c) => {
                lines.push(format!(
                    "[ai-crew-sync] You are agent '{}' on team '{}', in session '{}'. Teammates \
                     reach exactly this window at '{}/{}'.",
                    c.agent, c.team, st.session, c.agent, st.session
                ));
                lines.push(format!(
                    "- project: {}, role: {}, default channel: {}. Change them with \
                     {CONFIGURE_TOOL}; see them with {STATUS_TOOL}. Find teammates' windows \
                     with list_sessions.",
                    st.project.as_deref().unwrap_or("(none — set it)"),
                    st.role.as_deref().unwrap_or("(none — set it)"),
                    st.channel
                        .as_deref()
                        .or(st.project.as_deref())
                        .unwrap_or("(none)"),
                ));
                lines.push(
                    "- Nothing is pushed into an idle turn: call read_messages or wait_for_updates \
                     to receive what teammates sent."
                        .to_owned(),
                );
                if let Some(remote) = &c.remote_instructions {
                    lines.push(String::new());
                    lines.push(remote.clone());
                }
            }
            None => {
                lines.push(format!(
                    "[ai-crew-sync] Not connected to the team bus: {}. Only {CONFIGURE_TOOL} and \
                     {STATUS_TOOL} are available until a locally approved profile connects.",
                    st.disconnected_reason
                        .as_deref()
                        .unwrap_or("no profile resolved")
                ));
            }
        }
        lines.join("\n")
    }

    /// Periodic presence, and the credential renewed before it expires,
    /// until cancelled. Presence is on a fixed cadence; the renewal is
    /// scheduled from the expiry the bus last stated, so an idle window is
    /// renewed exactly as a busy one.
    pub async fn keepalive(self, ct: CancellationToken) {
        let mut next_heartbeat = tokio::time::Instant::now() + KEEPALIVE_EVERY;
        let mut last_flush = tokio::time::Instant::now() - FLUSH_RETRY;
        // Earliest next renewal attempt, whatever the deadline says: keeps a
        // deadline already in the past (a bus that will not renew, a reply
        // without an expiry) from becoming a tight loop. Tied to the context
        // generation it was set for: a profile switch brings a credential of
        // its own, whose first renewal must not wait out the old one's
        // backoff.
        let mut not_before: Option<(u64, tokio::time::Instant)> = None;
        loop {
            let deadline = self.renewal_deadline().await;
            let renew_at = deadline.map(|(generation, at)| match not_before {
                Some((for_generation, nb)) if for_generation == generation => at.max(nb),
                _ => at,
            });
            // A call the throttle deferred is flushed once it may be.
            // Never sooner than a moment after the last attempt: a bus that
            // refuses the report must not turn this into a busy loop.
            let flush_at = self
                .attention
                .lock()
                .ok()
                .and_then(|a| a.flush_at())
                .map(tokio::time::Instant::from_std)
                .map(|at| at.max(last_flush + FLUSH_RETRY));
            let flush_sleep = tokio::time::sleep_until(
                flush_at.unwrap_or_else(|| tokio::time::Instant::now() + KEEPALIVE_EVERY),
            );
            let renew_sleep = tokio::time::sleep_until(
                renew_at.unwrap_or_else(|| tokio::time::Instant::now() + KEEPALIVE_EVERY),
            );
            tokio::select! {
                _ = ct.cancelled() => return,
                // A new context: go round and schedule for its credential.
                _ = self.wake.notified() => {}
                _ = tokio::time::sleep_until(next_heartbeat) => {
                    // Renew, never re-assert "active": the window's hooks
                    // may have said its turn ended (#221).
                    self.refresh_presence().await;
                    next_heartbeat = tokio::time::Instant::now() + KEEPALIVE_EVERY;
                }
                _ = self.attention_changed.notified() => {
                    self.report_attention().await;
                }
                _ = flush_sleep, if flush_at.is_some() => {
                    last_flush = tokio::time::Instant::now();
                    self.report_attention().await;
                }
                _ = renew_sleep, if renew_at.is_some() => {
                    let pause = match self.renew_credential().await {
                        // A refused credential is refused again a moment
                        // later; look again on the presence cadence, in
                        // case a profile switch brought a live one.
                        Renewal::Refused => KEEPALIVE_EVERY,
                        Renewal::Renewed | Renewal::Retry | Renewal::Nothing => self.renewal_retry().await,
                    };
                    if let Some((generation, _)) = deadline {
                        not_before = Some((generation, tokio::time::Instant::now() + pause));
                    }
                }
            }
        }
    }

    /// When the credential this window holds should be renewed: its expiry
    /// less the lead, never earlier than now, with the context generation
    /// the credential belongs to. `None` without a credential.
    async fn renewal_deadline(&self) -> Option<(u64, tokio::time::Instant)> {
        let (generation, expires_at) = {
            let st = self.state.read().await;
            let expires_at = st
                .connected
                .as_ref()
                .and_then(|c| c.proof.as_ref())
                .map(|p| p.expires_at.clone())?;
            (st.generation, expires_at)
        };
        let expires_at = chrono::DateTime::parse_from_rfc3339(&expires_at).ok()?;
        let remaining = (expires_at.with_timezone(&chrono::Utc) - chrono::Utc::now())
            .num_seconds()
            .max(0);
        let due_in = (remaining - self.renewal_lead().await).max(0) as u64;
        Some((
            generation,
            tokio::time::Instant::now() + Duration::from_secs(due_in),
        ))
    }

    async fn renewal_lead(&self) -> i64 {
        let lifetime = requested_session_ttl().unwrap_or(crate::auth::SESSION_TTL_SECS);
        renewal_lead_secs(lifetime, env_secs(RENEW_LEAD_ENV))
    }

    /// Gap between renewal attempts: a quarter of the lead, so a transient
    /// failure gets several tries before the credential lapses.
    async fn renewal_retry(&self) -> Duration {
        Duration::from_secs((self.renewal_lead().await / 4).clamp(2, 60) as u64)
    }

    /// Extend the credential this window holds, without rotating its secret
    /// or epoch. Raced against the connected context's cancellation and
    /// fenced on the generation, so a profile switch under way discards the
    /// answer rather than applying it to the wrong identity.
    async fn renew_credential(&self) -> Renewal {
        let (remote, proof, generation, ct, profile) = {
            let st = self.state.read().await;
            let Some(c) = &st.connected else {
                return Renewal::Nothing;
            };
            let Some(p) = &c.proof else {
                return Renewal::Nothing;
            };
            (
                c.remote.clone(),
                p.clone(),
                st.generation,
                c.ct.clone(),
                c.resolved.profile.clone(),
            )
        };
        let mut args = json!({});
        if let Some(ttl) = requested_session_ttl() {
            args["ttl_seconds"] = json!(ttl);
        }
        let outcome = tokio::select! {
            _ = ct.cancelled() => return Renewal::Nothing,
            r = call_remote(&remote, "renew_session", args) => r,
        };
        match outcome {
            Ok(v) => {
                let Some(expires_at) = v["expires_at"].as_str().map(str::to_owned) else {
                    tracing::warn!("renew_session answered without an expiry; keeping the old one");
                    return Renewal::Retry;
                };
                if v["epoch"].as_i64().is_some_and(|e| e != proof.epoch) {
                    // A renewal never rotates; an answer that says otherwise
                    // is not applied to a credential it does not describe.
                    tracing::warn!(
                        "renew_session answered for another epoch; keeping the credential this \
                         window holds"
                    );
                    return Renewal::Retry;
                }
                {
                    let mut st = self.state.write().await;
                    if st.generation != generation {
                        return Renewal::Nothing;
                    }
                    let Some(current) = st.connected.as_mut().and_then(|c| c.proof.as_mut()) else {
                        return Renewal::Nothing;
                    };
                    if current.session_id != proof.session_id || current.epoch != proof.epoch {
                        return Renewal::Nothing;
                    }
                    current.expires_at = expires_at.clone();
                }
                self.stamp_binding_expiry(&proof, &expires_at).await;
                tracing::debug!(expires_at = %expires_at, "session credential renewed");
                Renewal::Renewed
            }
            Err(e) => match verdict_of(&e) {
                Some(Verdict::Unauthorized) => {
                    tracing::warn!(error = %e, "the bus refused to renew this window's credential");
                    self.mark_unauthorized(&profile, generation);
                    Renewal::Refused
                }
                Some(Verdict::NoSuchTool) => {
                    tracing::debug!("this bus does not renew credentials");
                    Renewal::Nothing
                }
                None => {
                    tracing::warn!(error = %e, "could not renew this window's credential; retrying");
                    Renewal::Retry
                }
            },
        }
    }

    /// Persist a renewed expiry into the binding, **only if the record still
    /// describes this credential**: a successor's record is left alone, as
    /// in `mark_closed`.
    async fn stamp_binding_expiry(&self, proof: &SessionProof, expires_at: &str) {
        let key = {
            let st = self.state.read().await;
            st.host_id.clone().unwrap_or_else(|| st.session.clone())
        };
        let path = context::binding_path(&self.opts.state_dir, &key);
        // Read, check and write under the configuration lock every writer
        // of this directory takes: a successor that replaces the record
        // between the check and the write would otherwise be overwritten
        // with this instance's older credential.
        let (session_id, epoch, expires_at) =
            (proof.session_id.clone(), proof.epoch, expires_at.to_owned());
        let stamped = under_config_lock(self.opts.state_dir.clone(), move || {
            let Ok(text) = std::fs::read_to_string(&path) else {
                return Ok(false);
            };
            let Ok(mut value) = serde_json::from_str::<Value>(&text) else {
                return Ok(false);
            };
            let same = value["session_id"].as_str() == Some(session_id.as_str())
                && value["epoch"].as_i64() == Some(epoch);
            if !same {
                return Ok(false);
            }
            if let Some(map) = value.as_object_mut() {
                map.insert("expires_at".into(), json!(expires_at));
                map.insert(
                    "updated_at".into(),
                    json!(chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true)),
                );
            }
            context::write_binding_file(&path, &value.to_string())?;
            Ok(true)
        })
        .await;
        match stamped {
            Ok(true) => {}
            Ok(false) => tracing::debug!(
                binding = %key,
                "another instance owns this binding now, or it is gone; not stamping"
            ),
            Err(e) => {
                tracing::warn!(error = %e, binding = %key, "could not record the renewed expiry")
            }
        }
    }

    /// Go quiet on the bus; the host is closing this window.
    pub async fn shutdown(&self) {
        let remote = {
            let st = self.state.read().await;
            st.connected.as_ref().map(|c| c.remote.clone())
        };
        if let Some(remote) = remote {
            let _ = tokio::time::timeout(
                EXIT_TIMEOUT,
                call_remote(
                    &remote,
                    "heartbeat",
                    // The host closed this window: nothing will start a turn
                    // in it.
                    // Nor will anything wake it: withdraw any announcement.
                    json!({"status": "idle", "ttl_seconds": 120, "attention": "idle", "wake": ""}),
                ),
            )
            .await;
            close_remote(remote).await;
        }
        // The credential stays on disk so a restart of this conversation can
        // resume the window; the record is only stamped as closed, and only
        // if it still describes this instance.
        self.mark_closed().await;
    }

    /// Mark this window closed, **only if the record still describes this
    /// instance**.
    ///
    /// Two things were wrong with clearing it unconditionally. A successor
    /// proxy for the same conversation may already have replaced the record,
    /// and wiping it would cut the live window's hooks off from their own
    /// credential. And the credential itself has to stay: a restart of this
    /// conversation resumes with it, which is the only way back in — the bus
    /// refuses to hand a live session to whoever holds the agent token, and
    /// rightly so. It is a 0600 file scoped to one window and it expires on
    /// its own.
    async fn mark_closed(&self) {
        let (key, mine) = {
            let st = self.state.read().await;
            let key = st.host_id.clone().unwrap_or_else(|| st.session.clone());
            // The window's identity, not its connection: one retired by a
            // failed adoption still owns the record it wrote, and only that
            // record. A window with no identity at all wrote none.
            let Some(identity) = st.identity() else {
                return;
            };
            let mine = identity.proof.map(|p| (p.session_id, p.epoch));
            (key, mine)
        };
        let path = context::binding_path(&self.opts.state_dir, &key);
        let shown = key.clone();
        let outcome = under_config_lock(self.opts.state_dir.clone(), move || {
            let Ok(text) = std::fs::read_to_string(&path) else {
                return Ok(false);
            };
            let Ok(mut value) = serde_json::from_str::<Value>(&text) else {
                return Ok(false);
            };
            // Ownership check: a record whose session or epoch has moved on
            // belongs to the instance that replaced us.
            if let Some((session_id, epoch)) = mine {
                let same = value["session_id"].as_str() == Some(session_id.as_str())
                    && value["epoch"].as_i64() == Some(epoch);
                if !same {
                    return Ok(false);
                }
            }
            if let Some(map) = value.as_object_mut() {
                map.insert("closed_at".into(), json!(chrono::Utc::now().to_rfc3339()));
            }
            context::write_binding_file(&path, &value.to_string())?;
            Ok(true)
        })
        .await;
        match outcome {
            Ok(true) => {}
            Ok(false) => tracing::debug!(
                binding = %shown,
                "another instance owns this binding now, or it is gone; leaving it alone"
            ),
            Err(e) => {
                tracing::warn!(error = %e, binding = %shown, "could not mark the binding closed")
            }
        }
    }
}

/// Run a binding read-modify-write under the configuration lock, on a
/// blocking thread: the lock is a file lock shared with every other writer
/// of the directory (other proxies of this conversation included), and a
/// wait for it must not stall the runtime that serves the host.
async fn under_config_lock<T: Send + 'static>(
    dir: PathBuf,
    f: impl FnOnce() -> anyhow::Result<T> + Send + 'static,
) -> anyhow::Result<T> {
    tokio::task::spawn_blocking(move || context::with_config_lock(&dir, f))
        .await
        .map_err(|e| anyhow::anyhow!("the binding writer task failed: {e}"))?
}

/// Close a remote connection we may share with an in-flight call. Sole
/// owner: cancel cleanly. Otherwise the last holder drops it, and dropping a
/// `RunningService` closes it as well.
async fn close_remote(remote: Arc<Remote>) {
    if let Ok(owned) = Arc::try_unwrap(remote) {
        let _ = owned.cancel().await;
    }
}

/// What an identity still holds on the bus, read before it is set aside.
async fn report_holdings(
    remote: &Remote,
    agent: &str,
    team: &str,
    session: &str,
) -> PreviousIdentity {
    let open_claims = call_remote(remote, "list_tasks", json!({"mine_only": true}))
        .await
        .ok()
        .and_then(|v| v["tasks"].as_array().cloned())
        .unwrap_or_default()
        .iter()
        .filter(|t| t["status"] == "claimed")
        .filter_map(|t| t["key"].as_str().map(str::to_owned))
        .collect();
    let held_locks = call_remote(remote, "list_locks", json!({}))
        .await
        .ok()
        .and_then(|v| v["locks"].as_array().cloned())
        .unwrap_or_default()
        .iter()
        // Agent AND session: locks are session-scoped, so a sibling window's
        // lock is not this one's to report as left behind.
        .filter(|l| {
            l["holder"] == agent && l["holder_session"].as_str().unwrap_or_default() == session
        })
        .filter_map(|l| l["name"].as_str().map(str::to_owned))
        .collect();
    PreviousIdentity {
        agent: agent.to_owned(),
        team: team.to_owned(),
        session: session.to_owned(),
        open_claims,
        held_locks,
    }
}

/// Why `resume_session` with a lifetime of its own is refused here.
const RESUME_TTL_REFUSED: &str = "resume_session: through this proxy a resumed credential keeps \
     the lifetime this window renews it with (BUS_SESSION_TTL_SECS, else the bus default), so \
     ttl_seconds is not accepted and nothing was resumed; call it without ttl_seconds";

/// A bus tool as this proxy serves it. `resume_session` is run by the
/// proxy, and its credential lives as long as every other one the window
/// holds: register, resume and each renewal ask for `BUS_SESSION_TTL_SECS`,
/// and the renewal schedule is computed from it. A lifetime asked for on one
/// call would last until the next renewal, so the proxy does not advertise
/// `ttl_seconds` and refuses it.
fn as_proxied(mut tool: Tool) -> Tool {
    if tool.name != "resume_session" {
        return tool;
    }
    let mut schema = (*tool.input_schema).clone();
    if let Some(Value::Object(properties)) = schema.get_mut("properties") {
        properties.remove("ttl_seconds");
    }
    if let Some(Value::Array(required)) = schema.get_mut("required") {
        required.retain(|r| r != "ttl_seconds");
    }
    schema.insert("additionalProperties".into(), Value::Bool(false));
    tool.input_schema = Arc::new(schema);
    let said = tool.description.as_deref().unwrap_or_default();
    tool.description = Some(
        format!(
            "{said} Through this proxy, the proxy runs the resume and keeps the new secret; \
             the lifetime is this window's (BUS_SESSION_TTL_SECS), so ttl_seconds is not \
             accepted."
        )
        .trim_start()
        .to_owned()
        .into(),
    );
    tool
}

fn tool_error(msg: String) -> CallToolResult {
    CallToolResult::error(vec![rmcp::model::ContentBlock::text(msg)])
}

impl ServerHandler for Proxy {
    async fn initialize(
        &self,
        request: rmcp::model::InitializeRequestParams,
        context: RequestContext<RoleServer>,
    ) -> Result<rmcp::model::InitializeResult, ErrorData> {
        // Bound and connected before answering, so the instructions the
        // host hands the model name the right window.
        self.admit(Some(&request.client_info)).await;
        if request.capabilities.roots.is_some()
            && self.opts.inputs.project_dir.is_none()
            && self.opts.inputs.profile.is_none()
            && let Ok(gate) = self.roots_gate.clone().try_lock_owned()
            && let Ok(mut hold) = self.roots_hold.lock()
        {
            *hold = Some(gate);
        }
        context.peer.set_peer_info(request.clone());
        self.negotiate_initialize(&request)
    }

    async fn discover(
        &self,
        context: RequestContext<RoleServer>,
    ) -> Result<rmcp::model::DiscoverResult, ErrorData> {
        self.admit(context.meta.client_info().as_ref()).await;
        Ok(rmcp::model::DiscoverResult::from_server_info(
            self.supported_protocol_versions().into_owned(),
            self.get_info(),
        ))
    }

    async fn on_initialized(&self, context: rmcp::service::NotificationContext<RoleServer>) {
        // Held since `initialize` when the client advertised roots.
        let Some(gate) = self.roots_hold.lock().ok().and_then(|mut h| h.take()) else {
            return;
        };
        let this = self.clone();
        tokio::spawn(async move {
            this.discover_roots(&context.peer).await;
            drop(gate);
        });
    }

    fn get_info(&self) -> ServerConfig {
        let mut info = ServerConfig::new(
            ServerCapabilities::builder()
                .enable_tools()
                .enable_tool_list_changed()
                .build(),
        );
        // `get_info` is synchronous; the state lock is uncontended at
        // initialize time, and a contended read simply yields the
        // disconnected wording until the next call.
        let text = match self.state.try_read() {
            Ok(st) => self.instructions(&st),
            Err(_) => format!("[ai-crew-sync] initialising; call {STATUS_TOOL} for details."),
        };
        info.instructions = Some(text);
        info
    }

    async fn list_tools(
        &self,
        _request: Option<PaginatedRequestParams>,
        context: RequestContext<RoleServer>,
    ) -> Result<ListToolsResult, ErrorData> {
        self.admit(context.meta.client_info().as_ref()).await;
        self.await_roots().await;
        let mut tools = local_tools();
        if let Some(c) = &self.state.read().await.connected {
            tools.extend(c.tools.iter().cloned().map(as_proxied));
        }
        Ok(crate::tools::catalogue(tools))
    }

    async fn call_tool(
        &self,
        request: CallToolRequestParams,
        context: RequestContext<RoleServer>,
    ) -> Result<CallToolResponse, ErrorData> {
        // The transport lifts `_meta` out of the params into the context.
        self.admit(context.meta.client_info().as_ref()).await;
        self.await_roots().await;
        self.observe_meta(&context.meta).await?;
        // Every host call is the model at work; a wait is the model blocked.
        let _turn = ToolTurn::enter(self, request.name.as_ref());
        match request.name.as_ref() {
            STATUS_TOOL => {
                let status = self.status().await;
                let value = serde_json::to_value(status)
                    .map_err(|e| ErrorData::internal_error(e.to_string(), None))?;
                Ok(CallToolResult::structured(value).into())
            }
            CONFIGURE_TOOL => {
                let args: ConfigureArgs = match request.arguments {
                    Some(map) => serde_json::from_value(Value::Object(map))
                        .map_err(|e| ErrorData::invalid_params(e.to_string(), None))?,
                    None => ConfigureArgs::default(),
                };
                match self.configure(args).await {
                    Ok(result) => {
                        let value = serde_json::to_value(result)
                            .map_err(|e| ErrorData::internal_error(e.to_string(), None))?;
                        Ok(CallToolResult::structured(value).into())
                    }
                    // The model must read this one, so it is a tool error
                    // rather than a protocol error.
                    Err(e) => Ok(tool_error(format!("{e:#}")).into()),
                }
            }
            // Delivery has to mean something. The bus hands over references
            // and records nothing; this writes them to disk, fsyncs, and
            // only then tells the bus they are held. A crash in between
            // costs a redelivery, which is idempotent by design.
            "fetch_conversation_inbox" => {
                let result = self.forward(request, context.ct.clone()).await?;
                Ok(self.spool_and_confirm(result, context.ct).await.into())
            }
            // A message with no channel and no recipient goes to this
            // window's channel. The state was decorative until now: the
            // instructions and session_status promised a default the bus
            // never saw.
            // A long wait is one call for the model and many for the bus:
            // the bus answers within a minute (HTTP intermediaries), and a
            // host that re-asks every minute spends a model turn each time.
            "wait_for_updates" => Ok(self.long_wait(request, context.ct).await?.into()),
            // A resume rotates the secret this proxy sends; the proxy has to
            // be the one that keeps the new one (#214). The lifetime is the
            // window's, not the call's: see `as_proxied`.
            "resume_session" => {
                if request
                    .arguments
                    .as_ref()
                    .and_then(|a| a.get("ttl_seconds"))
                    .is_some_and(|v| !v.is_null())
                {
                    return Ok(tool_error(RESUME_TTL_REFUSED.to_owned()).into());
                }
                match self.resume_in_place().await {
                    Some(Ok(value)) => Ok(CallToolResult::structured(value).into()),
                    Some(Err(said)) => Ok(tool_error(said).into()),
                    None => Ok(self.forward(request, context.ct).await?.into()),
                }
            }
            "post_message" => {
                let mut request = request;
                if let Some(channel) = self.default_channel().await {
                    let args = request.arguments.get_or_insert_with(Default::default);
                    let addressed = args.contains_key("channel") || args.contains_key("to");
                    if !addressed {
                        args.insert("channel".into(), Value::String(channel));
                    }
                }
                Ok(self.forward(request, context.ct).await?.into())
            }
            _ => Ok(self.forward(request, context.ct).await?.into()),
        }
    }
}

/// Run the proxy over stdio until the host closes the pipe.
pub async fn run(opts: ProxyOptions) -> anyhow::Result<()> {
    let proxy = Proxy::start(opts).await;
    let ct = CancellationToken::new();
    let keepalive = tokio::spawn(proxy.clone().keepalive(ct.child_token()));

    let running = proxy
        .clone()
        .serve(rmcp::transport::stdio())
        .await
        .context("MCP initialize over stdio failed")?;
    let quit = running.waiting().await;
    tracing::debug!(?quit, "host closed the connection");

    ct.cancel();
    let _ = keepalive.await;
    proxy.shutdown().await;
    Ok(())
}

#[cfg(test)]
mod renewal_tests {
    #[test]
    fn renewal_lead_is_half_the_lifetime_unless_overridden_and_inside_it() {
        assert_eq!(super::renewal_lead_secs(24 * 3600, None), 12 * 3600);
        assert_eq!(super::renewal_lead_secs(60, None), 30);
        assert_eq!(super::renewal_lead_secs(60, Some(50)), 50);
        assert_eq!(
            super::renewal_lead_secs(60, Some(600)),
            59,
            "never past the lifetime"
        );
        assert_eq!(
            super::renewal_lead_secs(1, None),
            1,
            "a degenerate lifetime still yields a lead"
        );
    }
}

#[cfg(test)]
mod unauthorized_tests {
    use rmcp::{
        RoleClient,
        model::{ErrorCode, ErrorData},
        transport::{
            DynamicTransportError, StreamableHttpClientTransport,
            streamable_http_client::{AuthRequiredError, StreamableHttpError},
        },
    };

    use super::*;

    fn transport(e: StreamableHttpError<reqwest::Error>) -> ServiceError {
        ServiceError::TransportSend(DynamicTransportError::new::<
            StreamableHttpClientTransport<reqwest::Client>,
            RoleClient,
        >(e))
    }

    #[test]
    fn a_rejected_bearer_is_the_transport_saying_so() {
        let e = transport(StreamableHttpError::AuthRequired(AuthRequiredError::new(
            "Bearer".into(),
        )));
        assert!(unauthorized(&e));
    }

    #[test]
    fn a_refusal_that_spells_401_is_still_a_refusal() {
        // The bus quoting a session label, a lease or an id that contains
        // "401" is talking to the model, not rejecting its credential.
        let e = ServiceError::McpError(ErrorData::invalid_request(
            "you do not hold the claim on 'api#1': it is held by joaquin (session \
             's-a7da401d8d70'), the lease expires in 401s",
            None,
        ));
        assert!(!unauthorized(&e));
        let e = ServiceError::McpError(ErrorData::internal_error("Auth required", None));
        assert!(
            !unauthorized(&e),
            "not even when it borrows the transport's words"
        );
    }

    #[test]
    fn a_missing_tool_is_the_code_saying_so() {
        // rmcp 3.x, unknown tool.
        let e = ServiceError::McpError(ErrorData::invalid_params("tool not found", None));
        assert!(no_such_tool(&e));
        assert_eq!(verdict(&e), Some(Verdict::NoSuchTool));
        // A JSON-RPC layer without the method at all.
        let e = ServiceError::McpError(ErrorData::new(
            ErrorCode::METHOD_NOT_FOUND,
            "Method not found",
            None,
        ));
        assert!(no_such_tool(&e));
    }

    #[test]
    fn a_refusal_that_spells_a_missing_method_is_still_a_refusal() {
        // The bus's conflict on a live label quotes the label, and a label
        // can hash to `s-32601…`. Same code as "tool not found".
        let e = ServiceError::McpError(ErrorData::invalid_params(
            "conflict: session 's-32601f03e877' is already registered and still live. \
             Holding the agent token does not make you that window: Method aside, \
             reconnect it with resume_session",
            None,
        ));
        assert!(!no_such_tool(&e));
        assert_eq!(verdict(&e), None);
        // The bus's own not-found errors share the code too.
        let e = ServiceError::McpError(ErrorData::invalid_params("not found: message 32601", None));
        assert!(!no_such_tool(&e));
        // Exactly rmcp's wording, not a prefix of it: a tool that exists
        // could open its refusal with the same three words.
        let e = ServiceError::McpError(ErrorData::invalid_params(
            "tool not found: the deploy tool named in `depends_on` does not exist",
            None,
        ));
        assert!(!no_such_tool(&e));
        assert!(!no_such_tool(&ServiceError::TransportClosed));
    }

    #[test]
    fn a_verdict_survives_the_anyhow_chain() {
        let e = anyhow::Error::new(Verdict::NoSuchTool).context("register_session failed");
        assert_eq!(verdict_of(&e), Some(Verdict::NoSuchTool));
        let e = anyhow::anyhow!("register_session failed: tool not found -32601 Method");
        assert_eq!(verdict_of(&e), None, "words are not a verdict");
    }

    #[test]
    fn an_http_refusal_is_read_for_its_shape() {
        let answered = |msg: &str| {
            transport(StreamableHttpError::UnexpectedServerResponse(
                msg.to_owned().into(),
            ))
        };
        // The bus's middleware body, at any status: its words, nothing ran.
        let e = answered(r#"HTTP 429 Too Many Requests: {"error":"rate limit exceeded"}"#);
        assert_eq!(
            refusal(&e),
            Some(Refusal {
                status: 429,
                said: Some("rate limit exceeded".into())
            })
        );
        assert!(remote_error_text(&e).contains("rate limit exceeded"));
        assert!(!unauthorized(&e));
        // A bare 4xx from something in front of the bus still ran nothing.
        let e = answered("HTTP 404 Not Found: <html>nope</html>");
        assert_eq!(
            refusal(&e),
            Some(Refusal {
                status: 404,
                said: None
            })
        );
        // A bare 5xx may be a gateway that gave up on a call that ran.
        assert_eq!(refusal(&answered("HTTP 504 Gateway Timeout: ")), None);
        assert_eq!(
            refusal(&answered("invalid www-authenticate header value")),
            None
        );
        assert_eq!(refusal(&ServiceError::TransportClosed), None);
    }

    #[test]
    fn another_transport_failure_is_not_a_rejected_bearer() {
        let e = transport(StreamableHttpError::UnexpectedContentType(Some(
            "text/html; 401".into(),
        )));
        assert!(!unauthorized(&e));
        assert!(!unauthorized(&ServiceError::TransportClosed));
    }
}

#[cfg(test)]
mod attention_tests {
    use super::{ATTENTION_REPORT_EVERY, Attention, AttentionReport};
    use std::time::{Duration, Instant};

    fn ago(d: Duration) -> Instant {
        Instant::now() - d
    }

    #[test]
    fn nothing_seen_reports_nothing_even_on_a_refresh() {
        let mut a = Attention::default();
        assert_eq!(a.due(true), None, "a keepalive is not news about the model");
    }

    #[test]
    fn a_call_reports_attending_and_a_burst_is_throttled() {
        let call = Instant::now();
        let mut a = Attention {
            calls: 1,
            unreported_call: Some(call),
            ..Attention::default()
        };
        let r = a.due(false).expect("first call");
        assert_eq!(r.state, "attending");
        a.sent(r);
        a.unreported_call = Some(Instant::now());
        assert_eq!(a.due(false), None, "same state inside the window");
        a.reported = Some(("attending", ago(ATTENTION_REPORT_EVERY)));
        assert_eq!(a.due(false).map(|r| r.state), Some("attending"));
    }

    #[test]
    fn a_keepalive_never_sends_a_stale_call() {
        // A throttled call, then the turn ended (a Stop hook said idle).
        // Minutes later the keepalive must not report it as current.
        let mut a = Attention {
            unreported_call: Some(ago(Duration::from_secs(120))),
            reported: Some(("attending", ago(Duration::from_secs(130)))),
            ..Attention::default()
        };
        assert_eq!(a.due(true), None);
        assert_eq!(a.unreported_call, None, "the stale call is dropped");
    }

    #[test]
    fn a_failed_send_is_retried() {
        let mut a = Attention {
            calls: 1,
            unreported_call: Some(Instant::now()),
            ..Attention::default()
        };
        let r = a.due(false).expect("due");
        // Not committed: the bus refused it.
        assert_eq!(a.due(false), Some(r));
        a.sent(r);
        assert_eq!(a.due(false), None);
    }

    #[test]
    fn a_wait_reports_waiting_and_a_refresh_keeps_it_fresh() {
        let mut a = Attention {
            calls: 1,
            waits: 1,
            unreported_call: Some(Instant::now()),
            ..Attention::default()
        };
        let r = a.due(false).expect("wait");
        assert_eq!(r.state, "waiting");
        a.sent(r);
        assert_eq!(a.due(false), None);
        assert_eq!(
            a.due(true),
            Some(AttentionReport {
                state: "waiting",
                call: None
            })
        );
        // Back from the wait: attending again at once.
        a.waits = 0;
        a.unreported_call = Some(Instant::now());
        assert_eq!(a.due(false).map(|r| r.state), Some("attending"));
    }
}