agentmux_srv/server/agent_handlers/

input.rs

// Copyright 2025-2026, AgentMux Corp.
// SPDX-License-Identifier: Apache-2.0

use std::sync::Arc;

use crate::backend::blockcontroller;
use crate::backend::obj::Block;
use crate::backend::rpc::engine::WshRpcEngine;
use crate::backend::rpc_types::{
    AttachmentRef, CommandAgentInputData, CommandAgentStopData, CommandSubprocessSpawnData, COMMAND_AGENT_INPUT,
    COMMAND_AGENT_STOP, COMMAND_SUBPROCESS_SPAWN,
};

use super::super::AppState;

/// Heal a container agent's argv IN PLACE when — and only when — it still
/// carries the persistent controller's flags.
///
/// A container agent runs one `docker exec` per turn: `container_spawn.rs`
/// writes the raw message, closes stdin, and reads until EOF. There is no
/// long-lived stdin and no control channel. The PERSISTENT argv is wrong for
/// that in three separate ways, and it is what stale blocks carry:
///
///   * `--input-format stream-json` — makes the CLI parse every stdin line as
///     a JSON envelope, so it meets the startup markdown and dies with
///     `JSON Parse error: Unrecognized token '#'`. This is the crash that kept
///     container agents from EVER starting (verified live, 2026-08-31).
///   * missing `-p` — the one-shot print flag. Without it the CLI is not in
///     the mode this path drives at all.
///   * `--permission-prompt-tool stdio` (+ a non-bypass `--permission-mode`) —
///     routes tool permissions through the control protocol, which only the
///     persistent controller speaks. A container turn has nothing to answer
///     `can_use_tool`, so it rejects or hangs on the first permission check.
///
/// An earlier cut of this deleted only `--input-format` and left the other two
/// (codex P1 on PR #2867) — that unblocks the immediate crash while leaving the
/// pane in the wrong CLI mode, which is a worse failure because it looks like
/// it works.
///
/// The cut after THAT rebuilt the argv wholesale from `launch_args`, which was
/// worse again (reagent P1 on the same PR): this runs on EVERY container turn,
/// not once per stale block, so a rebuild also threw away everything
/// `agent-model.ts` legitimately appends after the base args —
/// `agent.provider_flags` (user-configurable) and `--fork-session` (gated on
/// `providerId === "claude"`, not on `agentMode`, so container agents reach it).
/// It would have permanently broken both for container agents, on blocks whose
/// argv this fix had already assembled correctly.
///
/// So: subtract, don't rebuild. Remove the flags the PERSISTENT argv carries
/// that the one-shot argv does not, restore any one-shot flag that's missing,
/// and leave every other token — the user's `--model`/`--effort`, their
/// `provider_flags`, `--fork-session` — exactly where it was. Which flags are
/// "persistent-only" is derived from the provider catalog rather than hardcoded,
/// so a future flag moving between the two lists doesn't silently strand this.
///
/// An argv carrying NO persistent-only flag is returned untouched: it was built
/// by the fixed `selectLaunchArgs` path and there is nothing to heal.
///
/// The root cause is fixed at the source in `launch-args.ts`; this is the
/// self-heal for blocks ALREADY persisted with the bad argv, since `cmd:args`
/// lives in block meta and would otherwise stay wrong forever. Applied at the
/// point of use so a block that never re-runs `resync_controller` can't bypass
/// it.
pub(crate) fn container_argv(argv: Vec<String>, provider_id: &str) -> Vec<String> {
    let Some(provider) = crate::backend::providers::get_provider(provider_id) else {
        // Unknown provider — no catalog to diff against, so remove the one flag
        // that is outright fatal rather than guessing at the rest.
        return strip_flag_with_value(argv, "--input-format");
    };
    let Some(persistent) = provider.persistent_launch_args else {
        // Subprocess-shaped provider: its only argv IS the one-shot argv, so a
        // container block could never have been given persistent flags.
        return argv;
    };

    let one_shot = flags_with_arity(provider.launch_args);
    let persistent_only: Vec<(&str, bool)> = flags_with_arity(persistent)
        .into_iter()
        .filter(|(flag, _)| !one_shot.iter().any(|(f, _)| f == flag))
        .collect();

    // Already one-shot-shaped — the frontend assembled this argv, hands off.
    if !argv
        .iter()
        .any(|a| persistent_only.iter().any(|(f, _)| f == a))
    {
        return argv;
    }

    let mut out = argv;
    for (flag, takes_value) in &persistent_only {
        out = if *takes_value {
            strip_flag_with_value(out, flag)
        } else {
            out.into_iter().filter(|a| a != flag).collect()
        };
    }

    // Restore the one-shot flags the persistent argv never had (`-p` above all).
    // Prepended so the provider's own baseline keeps its leading position.
    let mut restored: Vec<String> = Vec::new();
    for (flag, takes_value) in &one_shot {
        if out.iter().any(|a| a == flag) {
            continue;
        }
        restored.push(flag.to_string());
        if *takes_value {
            if let Some(pos) = provider.launch_args.iter().position(|a| a == flag) {
                if let Some(v) = provider.launch_args.get(pos + 1) {
                    restored.push(v.to_string());
                }
            }
        }
    }
    restored.extend(out);
    restored
}

/// Split an args list into `(flag, takes_a_value)` pairs.
///
/// Arity is read off the list itself — a flag whose next token is not another
/// flag takes a value. That's what lets `container_argv` strip
/// `--permission-prompt-tool stdio` (two tokens) and `--verbose` (one)
/// correctly without a hardcoded table of every provider's flag shapes.
fn flags_with_arity(args: &'static [&'static str]) -> Vec<(&'static str, bool)> {
    let mut out = Vec::new();
    for (i, tok) in args.iter().enumerate() {
        if !tok.starts_with('-') {
            continue;
        }
        let takes_value = args.get(i + 1).is_some_and(|next| !next.starts_with('-'));
        out.push((*tok, takes_value));
    }
    out
}

/// Remove `flag` and the value token following it, everywhere it appears.
/// Tolerates a trailing `flag` with no value rather than panicking.
fn strip_flag_with_value(args: Vec<String>, flag: &str) -> Vec<String> {
    let mut out = Vec::with_capacity(args.len());
    let mut it = args.into_iter();
    while let Some(a) = it.next() {
        if a == flag {
            let _ = it.next();
            continue;
        }
        out.push(a);
    }
    out
}

/// Per-agent git commit identity env vars for a spawned agent process
/// (2026-08-22, docs/retro/retro-shared-git-identity-committer-misattribution-2026-08-22.md).
///
/// Without this, `git commit` falls through to whatever user.name/
/// user.email happens to be sitting in a shared multi-agent host's
/// `~/.gitconfig` -- one agent's real identity, silently baked into every
/// OTHER agent's commits too. `agentmux-cloud`'s review-notification
/// consumer resolves "who committed this" via GitHub's own commit->account
/// auto-linking (`commit.author.login`), not the raw git commit metadata
/// directly -- so whichever agent's *real, GitHub-verified* email is
/// sitting in the shared config silently receives every other agent's
/// committer notifications too (confirmed live: every commit across 9+
/// PRs from 4 different agents on one host resolved to a single agent's
/// account this way).
///
/// `GIT_AUTHOR_*`/`GIT_COMMITTER_*` env vars take precedence over
/// config-file `user.name`/`user.email` at commit time (git's own
/// behavior) -- scoped to just the spawned process tree, same pattern as
/// `AGENTMUX_AGENT_ID`. The email intentionally uses a non-existent
/// `.local` domain: GitHub can't link it to any real account, so an agent
/// with no dedicated PAT registered (the common case -- see this repo's
/// `CLAUDE.md`, "Which GitHub account am I acting as?") simply drops out
/// of `commit.author.login` lookups entirely instead of resolving to
/// someone else's real, verified identity.
fn git_identity_env_vars(agent_id: &str) -> [(&'static str, String); 4] {
    let git_email = format!("{}@agentmux.local", agent_id.to_lowercase());
    [
        ("GIT_AUTHOR_NAME", agent_id.to_string()),
        ("GIT_COMMITTER_NAME", agent_id.to_string()),
        ("GIT_AUTHOR_EMAIL", git_email.clone()),
        ("GIT_COMMITTER_EMAIL", git_email),
    ]
}

/// Identity M0 — `SPEC_AGENT_IDENTITY_CARRIED_NOT_DERIVED_2026_09_23.md`
/// §9.1: make the display name and the slug two explicit variables, present
/// on every launch path.
///
/// **Why.** `AGENTMUX_AGENT_ID` means two different things depending on who
/// spawned the agent (spec §0.2): the frontend's `launchAgentDefinition`
/// writes the definition *slug* into it (`agent-model.ts`), while this
/// server path's fallback wrote the *display name*
/// (`block.meta["agentName"]`) until `carry_agent_id_env` (2026-10-12), which
/// writes the slug too when the row has one. Nothing downstream could tell
/// which it got. The frontend path already sets
/// `AGENTMUX_AGENT_DISPLAY` and `AGENTMUX_AGENT_SLUG` alongside it; this
/// path set neither, so a reader that wanted the unambiguous form had
/// nothing to read. After this, both are always present when the server
/// knows them, and later phases can migrate readers off `AGENTMUX_AGENT_ID`
/// one at a time.
///
/// **Precedence.** The server's own knowledge wins, `cmd:env` is the
/// fallback — the reverse of the "user-provided values take precedence"
/// rule the surrounding identity variables follow, deliberately:
///
/// - `display_name` is `block.meta["agentName"]`, the value the primary
///   reactive registration is keyed on (see the Register-tail of
///   `run_agent_turn`) and the name the pane header shows. A `cmd:env` copy
///   is a launch-time snapshot that a rename (`setViewName`) leaves stale.
/// - `persisted_slug` is `db_agents.slug` for this block's agent — the
///   collision-resolved value `agent_def_insert` minted. A `cmd:env` copy is
///   whatever the frontend computed before that row existed.
///
/// When the two disagree it is logged, not silently resolved: a disagreement
/// is exactly the kind of signal spec §9.2 says a phase must be able to
/// observe.
///
/// **What this does not do.** `AGENTMUX_AGENT_ID` is never read or written
/// here. Spec §9.4 keeps its present value on both paths through M4; the
/// git identity, `MUXBUS_AGENT_ID`, the PR-body tag and the OSC prompt hook
/// all still derive from it, unchanged.
pub(crate) fn split_agent_identity_env(
    env_vars: &mut std::collections::HashMap<String, String>,
    display_name: &str,
    persisted_slug: Option<&str>,
) {
    let set_authoritative = |env_vars: &mut std::collections::HashMap<String, String>,
                             key: &str,
                             value: &str| {
        let value = value.trim();
        if value.is_empty() {
            return;
        }
        if let Some(existing) = env_vars
            .get(key)
            .map(|v| v.trim())
            .filter(|v| !v.is_empty())
        {
            if existing != value {
                tracing::info!(
                    key,
                    cmd_env = %existing,
                    authoritative = %value,
                    "spawn env: cmd:env disagrees with the server's value — server wins (identity M0)"
                );
            }
        }
        env_vars.insert(key.to_string(), value.to_string());
    };
    set_authoritative(env_vars, "AGENTMUX_AGENT_DISPLAY", display_name);
    if let Some(slug) = persisted_slug {
        set_authoritative(env_vars, "AGENTMUX_AGENT_SLUG", slug);
    }
}

/// `AGENTMUX_AGENT_WORKDIR`: the agent's own workspace, which the host-only
/// "Your workspace" Operator Config entry tells it to clone and work in
/// (SPEC_LAUNCH_CONTEXT_WORKSPACE_RULE_AND_STARTUP_FILES_2026_09_30.md §3.4).
/// Server-set, like `AGENTMUX_AUTH_KEY`: it is the launch cwd (`cmd:cwd`),
/// not user configuration, so a `cmd:env` copy never wins. Host agents only —
/// a container's cwd is a host path that means nothing inside the image (the
/// key is also on `CONTAINER_ENV_DENYLIST`).
///
/// Exported as the absolute path the process actually starts in: `~` is
/// expanded and a relative `cmd:cwd` is resolved against srv's own cwd, as
/// `core::apply_working_dir` does, so `cd "$AGENTMUX_AGENT_WORKDIR"` works
/// from anywhere (Codex on #4113).
pub(crate) fn carry_agent_workdir_env(
    env_vars: &mut std::collections::HashMap<String, String>,
    block_meta: &crate::backend::obj::MetaMapType,
) {
    const KEY: &str = "AGENTMUX_AGENT_WORKDIR";
    let agent_mode = crate::backend::obj::meta_get_string(block_meta, "agentMode", "host");
    let work_dir = crate::backend::obj::meta_get_string(block_meta, "cmd:cwd", "");
    if crate::backend::operator_config_seed::agent_kind(&agent_mode) == "host" && !work_dir.trim().is_empty() {
        let expanded = crate::backend::blockcontroller::core::expand_home_dir(&work_dir);
        let absolute = std::path::absolute(&expanded).map_or(expanded, |p| p.to_string_lossy().into_owned());
        env_vars.insert(KEY.to_string(), absolute);
    } else {
        env_vars.remove(KEY);
    }
}

/// What the server knows about the agent on a block, read from its
/// `db_agents` row: the identity that will be carried into the process.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct PersistedAgentIdentity {
    /// `db_agents.id` — the UID (spec §3). A UUID for every non-template row.
    pub uid: String,
    /// `db_agents.slug`, the collision-resolved value `agent_def_insert`
    /// minted. `None` if the row's slug is empty.
    pub slug: Option<String>,
    /// The agent's local identity token (`storage/agent_tokens.rs`), minted
    /// on first spawn. `None` only if minting failed — logged, and the
    /// spawn proceeds without it, because M1 has no reader for it yet.
    pub token: Option<String>,
}

/// The `db_agents` row for the agent shown on `block_id`, if it has one,
/// plus that agent's token.
///
/// Resolves the block → row the same way the identity injector does
/// (`identity/resolver/inject.rs`): `block.meta.agentId` first, then the
/// latest active launch on this block. A quick-launch pane (`launchAgent`
/// in `agent-model.ts`, whose `agentId` is a provider key like `"claude"`)
/// has no row and gets `None` — it has no slug or UID in the identity
/// sense, and deriving either from its display name would recreate the
/// lossy path spec §1.1 measured as the root defect.
///
/// The token is minted here, at spawn, rather than at row creation: every
/// row that exists today predates the table, so "mint at creation" would
/// leave them all without one. `agent_token_ensure` is idempotent, so a
/// row created in the future gets the same token on every spawn.
///
/// A store error is logged and treated as "unknown", never as a spawn
/// failure: this is disambiguation, not a gate, and spec §10 constraint 1
/// forbids a store read from making a healthy agent unreachable.
pub(crate) fn persisted_agent_identity(
    mstore: &crate::backend::storage::store::Store,
    block_id: &str,
) -> Option<PersistedAgentIdentity> {
    let instance = match mstore.instance_get_active_for_block(block_id) {
        Ok(Some(instance)) if !instance.id.trim().is_empty() => instance,
        Ok(_) => return None,
        Err(e) => {
            tracing::warn!(block_id, error = %e, "spawn env: instance lookup failed — no persisted identity from store");
            return None;
        }
    };
    let slug = match mstore.agent_def_get(&instance.definition_id) {
        Ok(Some(def)) if !def.slug.trim().is_empty() => Some(def.slug),
        Ok(_) => None,
        Err(e) => {
            tracing::warn!(block_id, error = %e, "spawn env: definition lookup failed — no AGENTMUX_AGENT_SLUG from store");
            None
        }
    };
    let token = match mstore.agent_token_ensure(&instance.id) {
        Ok(token) => Some(token),
        Err(e) => {
            tracing::warn!(block_id, uid = %instance.id, error = %e, "spawn env: token mint failed — spawning without AGENTMUX_AGENT_TOKEN");
            None
        }
    };
    // Identity M4d-2: the UID's own signing keys, copied from its name's on
    // ownership evidence or minted. Only token-carrying spawns get them (the
    // ones that will sign v2, M4d-6); nothing reads them yet. Like the token,
    // a store error is logged, never a spawn failure.
    if token.is_some() {
        if let Err(e) = mstore.agent_uid_keys_ensure(&instance.id) {
            tracing::warn!(block_id, uid = %instance.id, error = %e, "spawn env: UID key ensure failed");
        }
    }
    Some(PersistedAgentIdentity {
        uid: instance.id,
        slug,
        token,
    })
}

/// Identity M1 — spec §9.1: carry the UID and the token into the process.
///
/// `AGENTMUX_AGENT_UID` and `AGENTMUX_AGENT_TOKEN` are RESERVED,
/// server-controlled values in exactly the sense `AGENTMUX_AUTH_KEY` and
/// `AGENTMUX_BLOCKID` are at the top of `build_persistent_spawn_env`:
/// unconditional `insert`, never `entry().or_insert()`. A persisted
/// `cmd:env` carrying a stale value for either — a pane reused for a
/// different agent, a copied block — must not survive into this spawn, and
/// for the token specifically a stale value would be *another agent's*
/// credential. So when the server knows the identity it overwrites, and
/// when it does not (no row: quick-launch pane, or a lookup failure) it
/// **removes** both rather than leaving whatever was there.
///
/// Spec §4.1: identity is asserted by whoever minted it. The frontend is
/// not consulted (its `agentInstanceId` is documented as going stale on
/// pane reuse), and the agent is not asked. The row is the source.
///
/// Descendant inheritance is real and not claimed otherwise (spec §6.4):
/// the agent's own subprocesses — its shell tool, its MCP servers, build
/// scripts — inherit the token. That is a smaller blast radius than the
/// instance-wide `AGENTMUX_AUTH_KEY` they already inherit, not zero. The
/// `.mcp.json` `agent_config.rs` writes into the working directory carries
/// neither variable (it copies only `AGENTMUX_AGENT_ID`/`_BUS_ID`), so the
/// token never lands in a file a human may commit.
pub(crate) fn carry_agent_uid_env(
    env_vars: &mut std::collections::HashMap<String, String>,
    identity: Option<&PersistedAgentIdentity>,
) {
    const UID: &str = "AGENTMUX_AGENT_UID";
    const TOKEN: &str = "AGENTMUX_AGENT_TOKEN";
    match identity {
        Some(id) => {
            env_vars.insert(UID.to_string(), id.uid.clone());
            match &id.token {
                Some(token) => {
                    env_vars.insert(TOKEN.to_string(), token.clone());
                }
                None => {
                    env_vars.remove(TOKEN);
                }
            }
        }
        None => {
            env_vars.remove(UID);
            env_vars.remove(TOKEN);
        }
    }
}

/// #3577: the id of the deleted agent `block_id` names, when the block
/// resolved to no row (`identity` is `None`) and its agent was deleted —
/// the builder's last-read re-check of the gate's refusal (Codex P1 on
/// #3591). `None` whenever a row resolved, or the block names nothing
/// deleted.
pub(crate) fn deleted_agent_of_rowless_block(
    mstore: &crate::backend::storage::store::Store,
    block_id: &str,
    identity: Option<&PersistedAgentIdentity>,
) -> Option<String> {
    if identity.is_some() {
        return None;
    }
    crate::identity::resolver::block_agent_id(mstore, block_id)
        .filter(|agent_id| mstore.agent_was_deleted(agent_id))
}

/// Identity M4b-2 (spec §6.5.8): carry `block_id`'s row UID and token into
/// the env of a controller that builds its own (App Server, ACP) instead of
/// going through [`build_persistent_spawn_env`]. Same resolution and the
/// same reserved-value rules as that builder (stale values are overwritten,
/// or removed when the block has no row). A synchronous store read and an
/// idempotent token mint, so on a multi-thread runtime worker it runs under
/// `block_in_place`, as eager resume does (incident #1782).
pub(crate) fn carry_block_identity_env(
    mstore: &crate::backend::storage::store::Store,
    block_id: &str,
    env_vars: &mut std::collections::HashMap<String, String>,
) {
    let lookup = || persisted_agent_identity(mstore, block_id);
    let identity = match tokio::runtime::Handle::try_current() {
        Ok(handle) if handle.runtime_flavor() == tokio::runtime::RuntimeFlavor::MultiThread => {
            tokio::task::block_in_place(lookup)
        }
        _ => lookup(),
    };
    carry_agent_uid_env(env_vars, identity.as_ref());
}

/// The slice of [`AppState`] an agent turn needs in order to be started.
///
/// Exists so [`run_agent_turn`] can be driven from somewhere other than the
/// `agentinput` RPC handler it was originally inlined in. The second caller is
/// the reactive handler's message sender (`bootstrap::install_agent_turn_delivery`):
/// a `SubprocessController` starts a turn ONLY via this path, so without a
/// non-RPC entry point every inter-agent message addressed to one was dropped.
/// See `docs/reports/REPORT_JEKT_DELIVERY_DROPS_SUBPROCESS_AGENTS_2026_09_02.md`.
///
/// Field names match the local bindings the extracted body already used, so the
/// body itself is unchanged by the extraction.
#[derive(Clone)]
pub struct AgentTurnDeps {
    pub mstore: Arc<crate::backend::storage::store::Store>,
    pub id_store: Arc<crate::backend::storage::store::Store>,
    pub identity_store: Arc<crate::backend::storage::store::Store>,
    /// Streaming-bash wrapper auth key — see SPEC_STREAMING_BASH_RUNNER_2026_05_11.md §7.
    pub auth_key: String,
    pub broker: Arc<crate::backend::mps::Broker>,
    pub container_manager: Arc<crate::backend::container::ContainerRuntimeHandle>,
    /// Named `filestore_gate` because the spawn-gate error frame MUST be
    /// persisted through it, not merely live-broadcast (reagent P1, PR #2164).
    pub filestore_gate: Arc<crate::backend::storage::filestore::FileStore>,
    pub local_web_url: String,
    pub event_bus_gate: Arc<crate::backend::eventbus::EventBus>,
}

impl AgentTurnDeps {
    pub fn from_state(state: &AppState) -> Self {
        Self {
            mstore: state.mstore.clone(),
            id_store: state.id_store.clone(),
            identity_store: state.identity_store.clone(),
            auth_key: state.auth_key.clone(),
            broker: state.broker.clone(),
            container_manager: state.container_manager.clone(),
            filestore_gate: state.filestore.clone(),
            local_web_url: state.local_web_url.clone(),
            event_bus_gate: state.event_bus.clone(),
        }
    }
}

/// Start one agent turn for `block_id` with `message`.
///
/// This is the whole of what `AgentInputCommand` does, verbatim: re-read the
/// spawn config from block meta, inject identity/muxbus/bashwrap env (failing
/// closed on the oauth spawn gate), then dispatch to the persistent controller,
/// the host subprocess path, or the container `docker exec` path as the block's
/// `agentMode` and controller type require — and finally re-register the agent
/// for reactive delivery.
///
/// Extracted from the `agentinput` handler body with no behavior change so a
/// second, non-RPC caller can reach it; see [`AgentTurnDeps`].
/// Whether starting a turn should also (re-)register the agent for reactive
/// delivery.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TurnRegistration {
    /// The RPC / UI path. Registers, exactly as `AgentInputCommand` always has.
    Register,
    /// The reactive-delivery path (`bootstrap::install_agent_turn_delivery`).
    /// Skipping is required, for two independent reasons:
    ///
    /// 1. **Redundant.** That caller resolved this block *by looking the agent
    ///    up in the reactive handler's own registry* (`uid_to_block` /
    ///    `name_to_blocks`), so the agent
    ///    is registered by construction. There is nothing to re-register.
    /// 2. **Deadlock.** `ReactiveHandler::inject_message` holds the global
    ///    `Mutex<Handler>` across the message-sender call, and that sender
    ///    drives this function to completion on the same thread. The
    ///    registration below calls `get_global_handler().register_agent(...)`,
    ///    which locks that same `std::sync::Mutex` — which is NOT reentrant.
    ///    Registering here would block the thread on a lock it already holds,
    ///    wedging the reactive handler process-wide, on essentially every
    ///    successful delivery to a subprocess agent (reagent P0 on PR #2930).
    Skip,
}

/// Builds the full spawn environment for a persistent-controller CLI
/// process: `cmd:env` base → Layer 3 identity/credential gate → the two
/// RESERVED wrapper variables (unconditionally overwritten —
/// `AGENTMUX_AUTH_KEY`/`AGENTMUX_BLOCKID`) → the two agent-identity
/// variables and per-agent git identity (all user-overridable via
/// `cmd:env`) → bundled/user tools PATH → the reserved guards (`gh_guard`,
/// `account_login_guard`: no agent gets a human's `gh` login or the
/// account's cloud login).
///
/// Shared by every real spawn of a persistent-controller CLI — this
/// function's own logic used to live inline in `run_agent_turn` below, and
/// only there. `PersistentSubprocessController::start()`'s eager-resume
/// path (`SPEC_PERSISTENT_CONTROLLER_EAGER_RESUME_ON_RECONNECT_2026_09_20.md`)
/// built its own second, independent copy of a SUBSET of this — codex P1 on
/// PR #3513 found it had already drifted from this one: missing PATH and
/// (then) MuxBus-token injection entirely, and using `entry().or_insert()` instead
/// of an unconditional overwrite for the two reserved wrapper variables
/// (meaning a stale persisted `cmd:env` value for either would silently
/// survive across an eager resume, unlike a live message send). One
/// function, not two copies kept in sync by hand.
///
/// `Err(SpawnGateError)` means the caller must NOT spawn — same contract as
/// `inject_identity_env_async` itself, which this wraps.
pub(crate) async fn build_persistent_spawn_env(
    mstore: Arc<crate::backend::storage::store::Store>,
    id_store: Arc<crate::backend::storage::store::Store>,
    identity_store: Arc<crate::backend::storage::store::Store>,
    broker: Option<Arc<crate::backend::mps::Broker>>,
    block_meta: &crate::backend::obj::MetaMapType,
    block_id: &str,
    auth_key: &str,
    base_env_vars: std::collections::HashMap<String, String>,
) -> Result<std::collections::HashMap<String, String>, crate::identity::resolver::SpawnGateError> {
    // Identity injection: look up the active AgentInstance for this block,
    // resolve its identity_id's bindings, and merge each per-provider env
    // var into the spawn map. Api-key-class failures are logged and
    // skipped — the agent CLI launches with whatever resolved cleanly plus
    // the static cmd:env block. Oauth-class resolution failures are
    // BLOCKING unless the agent opted into ambient login (layer-3 spawn
    // gate, SPEC_ACCOUNT_DELETE_DEAUTH_LAYERS_2_4_2026_07_14.md §2.2).
    // Kept for the identity-M0 slug lookup below; `mstore` itself moves into
    // the identity injector.
    let mstore_for_slug = mstore.clone();
    let mut env_vars = crate::identity::resolver::inject_identity_env_async(
        mstore,
        id_store.clone(),
        identity_store,
        broker,
        block_id.to_string(),
        base_env_vars,
    )
    .await?;

    // No MuxBus account login here: agents reach the cloud through the srv,
    // which uses its own stored login (see `account_login_guard`).

    // Streaming-bash wrapper auth + discovery
    // (SPEC_STREAMING_BASH_RUNNER_2026_05_11.md §7).
    //
    // 1. AGENTMUX_AUTH_KEY — config.rs's `Config::from_env_and_args` removed
    //    it from the process env at startup (security PR #801). Re-inject for this spawn so the
    //    wrapper (running inside Claude's bash subprocess tree) can
    //    authenticate against the auth_middleware-gated
    //    /agentmux/wps/publish endpoint via X-AuthKey.
    // 2. PATH — prepend the bundled tools/bin dir so `agentmux-bashwrap.exe`
    //    resolves when the PreToolUse hook (auto-injected by
    //    agent_config.rs) rewrites the command to invoke it.
    //    AGENTMUX_LOCAL_URL is already in the inherited process env
    //    (set by `bind_listeners_and_network` in bootstrap/network.rs).
    //
    // Unconditional `insert`, NOT `entry().or_insert()` — these two are
    // RESERVED, server-controlled values (the current key, the real block
    // id), not user configuration. A persisted `cmd:env` carrying a stale
    // value for either (e.g. from a prior server restart with a different
    // key) must not survive into this spawn.
    env_vars.insert("AGENTMUX_AUTH_KEY".to_string(), auth_key.to_string());
    // Block id so the wrapper can scope its MPS publishes to `block:<id>`.
    // Without this, chunks publish without a scope and the frontend's
    // per-block subscription doesn't receive them.
    env_vars.insert("AGENTMUX_BLOCKID".to_string(), block_id.to_string());

    // Identity M0 + M1. M0: name and slug as two explicit variables, on this
    // path too — the frontend's `launchAgentDefinition` already sets both;
    // this path never did, so a reader could only get at either by guessing
    // what `AGENTMUX_AGENT_ID` held (`split_agent_identity_env`). M1: the
    // UID and the agent's token, from the row the server itself created
    // (`carry_agent_uid_env`). One store round-trip serves both.
    {
        let display_name = crate::backend::obj::meta_get_string(block_meta, "agentName", "");
        let identity = {
            let mstore = mstore_for_slug.clone();
            let owned_block_id = block_id.to_string();
            match tokio::task::spawn_blocking(move || {
                let identity = persisted_agent_identity(&mstore, &owned_block_id);
                // #3577, re-checked here: this is the last store read before
                // the process starts. The gate refused a deleted agent, but
                // the agent can be deleted after the gate's lookup succeeded
                // (the delete handlers share no lock with the spawn path) —
                // then no row resolves here, and without this the spawn
                // would go ahead with no identity (Codex P1 on #3591). A
                // delete after THIS read is a delete of a running agent.
                let deleted =
                    deleted_agent_of_rowless_block(&mstore, &owned_block_id, identity.as_ref());
                (identity, deleted)
            })
            .await
            {
                Ok((_, Some(agent_id))) => {
                    return Err(crate::identity::resolver::SpawnGateError::AgentDeleted {
                        agent_id,
                    });
                }
                Ok((identity, None)) => identity,
                Err(e) => {
                    tracing::warn!(
                        block_id = %block_id,
                        error = %e,
                        "spawn env: persisted-identity lookup task failed — no UID/token this spawn; AGENTMUX_AGENT_SLUG falls back to cmd:env"
                    );
                    None
                }
            }
        };
        split_agent_identity_env(
            &mut env_vars,
            &display_name,
            identity.as_ref().and_then(|i| i.slug.as_deref()),
        );
        carry_agent_uid_env(&mut env_vars, identity.as_ref());
        super::identity_env::carry_agent_id_env(&mut env_vars, &display_name, identity.as_ref().and_then(|i| i.slug.as_deref()));
    }
    carry_agent_workdir_env(&mut env_vars, block_meta);
    crate::server::browser_uploads::record_launch_workspace(block_id, &env_vars);
    // Per-agent git commit identity -- see git_identity_env_vars() doc
    // comment. Still overridable per the same "user-provided values take
    // precedence" rule as every other var here.
    if let Some(agent_id) = env_vars.get("AGENTMUX_AGENT_ID").cloned() {
        for (key, value) in git_identity_env_vars(&agent_id) {
            if !env_vars.contains_key(key) {
                env_vars.insert(key.to_string(), value);
            }
        }
    }
    // PATH gets the bundled tools dir (portable builds, runtime/tools/bin/)
    // and the user tools dir (~/.agentmux/tools/bin/), both ahead of the
    // inherited PATH: bundled is None in `task dev` (tool_store excludes
    // target/debug), so the wrapper is in the user dir and a system copy must
    // not shadow it. See `UserToolsPrecedence`.
    {
        let existing = env_vars
            .get("PATH")
            .cloned()
            .or_else(|| std::env::var("PATH").ok())
            .unwrap_or_default();
        if let Some(path) = crate::backend::tool_store::tools_path(
            &existing,
            crate::backend::tool_store::UserToolsPrecedence::BeforeExisting,
        ) {
            env_vars.insert("PATH".to_string(), path);
        }
    }
    super::bash_prefix::apply_for_spawn(&mut env_vars, block_id);
    // Plain `gh` must not act as a human's gh login (see `gh_guard`), and no
    // agent holds the account's cloud login (see `account_login_guard`). Last,
    // so nothing above — a persisted `cmd:env`, an identity binding — can set
    // them back: these are reserved variables, not defaults.
    crate::backend::gh_guard::apply_gh_guard(&mut env_vars);
    crate::backend::account_login_guard::strip_account_login(&mut env_vars);

    Ok(env_vars)
}

/// Why `run_agent_turn` refused to spawn, before any process started.
enum TurnGate {
    /// Another AgentMux instance on this host runs this agent.
    Admission(String),
    /// The identity/credential spawn gate refused.
    Identity(crate::identity::resolver::SpawnGateError),
}

impl std::fmt::Display for TurnGate {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            TurnGate::Admission(refusal) => f.write_str(refusal),
            TurnGate::Identity(e) => write!(f, "{e}"),
        }
    }
}

/// A held message's join hint, armed for one dispatch. Dropped without
/// `keep` (the dispatch failed), it is cleared, so no later message can take
/// it (#4492).
struct ArmedJoinHint(Option<std::sync::Arc<crate::backend::blockcontroller::health::TurnActivityTracker>>);

impl ArmedJoinHint {
    fn arm(
        joins_turn: Option<u64>,
        tracker: Option<std::sync::Arc<crate::backend::blockcontroller::health::TurnActivityTracker>>,
    ) -> Self {
        match (joins_turn, tracker) {
            (Some(turn_id), Some(tracker)) => {
                tracker.hint_join(turn_id);
                Self(Some(tracker))
            }
            _ => Self(None),
        }
    }

    /// The dispatch went through: the pass it starts takes the hint.
    fn keep(mut self) {
        self.0 = None;
    }
}

impl Drop for ArmedJoinHint {
    fn drop(&mut self) {
        if let Some(tracker) = self.0.take() {
            tracker.clear_join_hint();
        }
    }
}

pub async fn run_agent_turn(
    deps: &AgentTurnDeps,
    block_id: String,
    message: String,
    message_id: Option<String>,
    registration: TurnRegistration,
    origin: crate::backend::blockcontroller::health::TurnOrigin,
    attachments: Vec<AttachmentRef>,
) -> Result<(), String> {
    run_agent_turn_joining(deps, block_id, message, message_id, registration, origin, attachments, None).await
}

/// [`run_agent_turn`] for a message the pane held while turn `joins_turn`
/// ran: if that turn is still settling (or only just ended), the message
/// joins it rather than starting a new one (turn-model spec §4.3, J3).
#[allow(clippy::too_many_arguments)]
pub async fn run_agent_turn_joining(
    deps: &AgentTurnDeps,
    block_id: String,
    message: String,
    message_id: Option<String>,
    registration: TurnRegistration,
    origin: crate::backend::blockcontroller::health::TurnOrigin,
    attachments: Vec<AttachmentRef>,
    joins_turn: Option<u64>,
) -> Result<(), String> {
    let AgentTurnDeps {
        mstore,
        id_store,
        identity_store,
        auth_key,
        broker,
        container_manager,
        filestore_gate,
        local_web_url,
        event_bus_gate,
    } = deps.clone();

    let ctrl = blockcontroller::get_controller(&block_id)
        .ok_or_else(|| format!("no controller for block {}", block_id))?;
    // A message the pane held during a turn joins that turn (turn-model spec
    // §4.3, J3). The persistent controller carries it on the input itself;
    // the others start passes with no labelled input, so their tracker is told
    // just before the dispatch, never earlier: a failure in the preparation
    // must not leave the hint armed for a later message (#4492).
    let arm_join_hint = || ArmedJoinHint::arm(joins_turn, ctrl.turn_tracker());

    // Re-read the spawn config from block metadata
    let block: Block = mstore
        .get(&block_id)
        .map_err(|e| format!("agentinput: load block: {e}"))?
        .ok_or_else(|| format!("block {} not found", block_id))?;

    let cli_command = crate::backend::obj::meta_get_string(&block.meta, "cmd", "claude");
    let cli_args: Vec<String> = match block.meta.get("cmd:args") {
        Some(serde_json::Value::Array(arr)) => arr
            .iter()
            .filter_map(|v| v.as_str().map(|s| s.to_string()))
            .collect(),
        _ => vec![
            "-p".to_string(),
            "--input-format".to_string(),
            "stream-json".to_string(),
            "--output-format".to_string(),
            "stream-json".to_string(),
        ],
    };
    // A pane stored without --model/--effort (saved before the launch fixes, or
    // written by a path that forgot) would spawn on the CLI default; fill what
    // its `agent:runtime` says. Never overrides a flag already there.
    let cli_args = crate::backend::agent_runtime::with_runtime_flags(&block.meta, cli_args);
    let working_dir = crate::backend::obj::meta_get_string(&block.meta, "cmd:cwd", "");
    let env_vars = crate::backend::blockcontroller::cmd_env_of(&block.meta);
    // Identity gate, MuxBus token, reserved wrapper vars, agent identity,
    // git identity, tools PATH — see `build_persistent_spawn_env`'s own doc
    // comment. Broker hand-in lets the identity resolver publish
    // `identity:no-direct-links`, and its OAuth expiry probe
    // `identityaccounts:changed` when it flips a token's status.
    //
    // Before all of that — the env build writes shared identity state — a
    // persistent pane about to spawn checks that no other AgentMux instance
    // on this host runs this agent (SPEC_AGENT_SINGLE_LIVE_INSTANCE_2026_09_24
    // Phase 1, I9). A refusal takes the gate's error path below, so it is
    // persisted and shown in the pane like any other spawn refusal.
    let admission = match ctrl
        .as_any()
        .downcast_ref::<blockcontroller::persistent::PersistentSubprocessController>()
    {
        Some(p) if p.needs_spawn() => {
            let uid = crate::backend::obj::meta_get_string(&block.meta, "agentId", "");
            let name = crate::backend::obj::meta_get_string(&block.meta, "agentName", "");
            p.check_admission(&uid, &name).await
        }
        _ => Ok(()),
    };
    let env_result = match admission {
        Err(refusal) => Err(TurnGate::Admission(refusal)),
        Ok(()) => build_persistent_spawn_env(
            mstore.clone(),
            id_store.clone(),
            identity_store.clone(),
            Some(broker.clone()),
            &block.meta,
            &block_id,
            &auth_key,
            env_vars,
        )
        .await
        .map_err(TurnGate::Identity),
    };
    // Show a refusal that happened before any process ran in the pane: the
    // raw frame in the output log, plus the classified failure the recovery
    // row reads. Used by the gate below and by a single-live-instance
    // refusal from the persistent spawn/fence path further down.
    let surface_refusal = |message: &str| {
            let error_frame = serde_json::json!({
                "type": "result",
                "is_error": true,
                "subtype": "error_during_execution",
                "error": {"message": format!("[AgentMux] {message}")}
            })
            .to_string();
            // Some(&filestore_gate): the frame must be PERSISTED
            // to the block file, not just live-broadcast —
            // otherwise the error vanishes on pane
            // reload/reconnect (reagent P1, PR #2164 round 2). And to
            // the agent's global zone: a pane with an agentId reads
            // that on reload, so without it the frame still vanished
            // (Phase 5a-3c).
            let global_zone = crate::backend::blockcontroller::shell::resolve_global_output_zone(
                &Some(mstore.clone()),
                &block_id,
            );
            crate::backend::blockcontroller::shell::handle_append_block_file(
                &broker,
                &block_id,
                crate::backend::blockcontroller::subprocess::SUBPROCESS_OUTPUT_SUBJECT,
                format!("{error_frame}\n").as_bytes(),
                Some(&filestore_gate),
                global_zone.as_deref(),
            );
            // codex P1, PR #2802: the frame above only lands in
            // the block's raw output log, which the recovery
            // banner does NOT read from — it reads the
            // structured `agent:last_failure` block-meta key
            // (persist_last_failure) + the ephemeral
            // EVENT_AGENT_FAILURE push, same as every POST-spawn
            // exit classification (host_spawn.rs). No process
            // ever ran here, so classify() gets no exit
            // code/stderr/result-frame — just the gate's own
            // Display text, exactly like health.rs's in-band-error
            // reclassification call.
            let gate_failure =
                crate::agents::failure::classify(None, None, message, None);
            crate::backend::blockcontroller::core::persist_last_failure(
                &block_id,
                Some(&gate_failure),
                &Some(mstore.clone()),
                &Some(event_bus_gate.clone()),
            );
            broker.publish_event::<crate::backend::mps_events::AgentFailed>(&gate_failure, vec![format!("block:{}", block_id)], 1);
    };
    let mut env_vars = match env_result {
        Ok(env) => env,
        Err(gate) => {
            surface_refusal(&gate.to_string());
            return Err(match &gate {
                TurnGate::Admission(refusal) => refusal.clone(),
                TurnGate::Identity(e) => format!("identity spawn gate: {e}"),
            });
        }
    };
    // Identity M2: the UID this turn's spawn env carries (M1a), kept for the
    // Register-tail below so registration is keyed by identity without a
    // second store read (spec §4.4.1 row 2). `None` when the block has no
    // row yet — the tail then registers by name, counted.
    let turn_uid: Option<String> = env_vars
        .get("AGENTMUX_AGENT_UID")
        .map(|u| u.trim().to_string())
        .filter(|u| !u.is_empty());
    let turn_agent_id = env_vars.get("AGENTMUX_AGENT_ID").cloned(); // the stable binding, as spawn parks it

    let session_id_field =
        crate::backend::obj::meta_get_string(&block.meta, "agent:session_id_field", "session_id");

    // Attachments (SPEC_AGENT_PANE_IMAGE_ATTACHMENTS_2026_09_26.md §6.6,
    // SPEC_AGENT_PANE_FILE_ATTACHMENTS_2026_09_26.md §7): every provider gets
    // a numbered list of the files appended to the text; Claude's persistent
    // stream-json path also gets images and short PDFs inline. Container
    // agents can't reach host paths yet (their panes copy files into the
    // working folder instead), so they refuse rather than send a list the
    // agent can't open.
    let mut inline_images: Vec<serde_json::Value> = Vec::new();
    let message = if attachments.is_empty() {
        message
    } else {
        let agent_mode = crate::backend::obj::meta_get_string(&block.meta, "agentMode", "host");
        // Refusals go through surface_refusal like every other early exit
        // here, so the reason survives a pane reload.
        if agent_mode == "container" {
            let msg = "Attachments aren't supported for container agents yet.";
            surface_refusal(msg);
            return Err(msg.to_string());
        }
        let Some(svc) = crate::backend::attachments::get() else {
            let msg = "Image attachments are unavailable: the attachment service isn't running.";
            surface_refusal(msg);
            return Err(msg.to_string());
        };
        let is_persistent = ctrl
            .as_any()
            .downcast_ref::<blockcontroller::persistent::PersistentSubprocessController>()
            .is_some();
        let limits = svc.limits();
        let inline_max = if is_persistent { limits.claude_inline_max } else { 0 };
        // Claude Code keeps every inline image in its own session transcript,
        // so inline bytes are budgeted per session too (spec §6.6). A new
        // session has no id yet; its first turn counts against the block.
        // Once the id is known, that first turn's count moves to the session.
        let sid = crate::backend::obj::meta_get_string(&block.meta, "agent:sessionid", "");
        let new_session_key = format!("block:{block_id}:new");
        let session_key = if sid.is_empty() { new_session_key.clone() } else { format!("session:{sid}") };
        // Re-derive any send-copy whose fingerprint is stale (a changed send
        // edge) under the service's CPU and memory limits, before the
        // blocking resolve below looks the files up.
        for a in &attachments {
            svc.ensure_derived(&a.id).await;
        }
        let prepared = tokio::task::spawn_blocking(move || {
            use crate::backend::attachments::prompt;
            let (found, missing) = prompt::resolve(&svc, &attachments);
            // Read, spend and record the session budget as one locked step.
            svc.with_session_inline_budget(
                &session_key,
                Some(&new_session_key),
                limits.claude_session_inline_bytes,
                |budget| {
                    let prepared = prompt::prepare(&found, &missing, inline_max, budget);
                    let spent = prompt::inline_bytes(&prepared.inline);
                    (prepared, spent)
                },
            )
        })
        .await
        .map_err(|e| format!("attachments: {e}"))?;
        inline_images = prepared.inline;
        crate::backend::attachments::prompt::append(&message, &prepared.list)
    };

    // App Server owns a persistent thread/turn process and has its own typed
    // protocol path. Keep it ahead of the legacy stream-json branches.
    //
    // KNOWN GAP (ReAgent P1, PR #3215, tracked for a follow-up PR, not fixed
    // here): env_vars above (including this turn's freshly-resolved identity
    // bindings from inject_identity_env_async) is computed but never reaches
    // this branch — send_message carries no env, and the App Server child was
    // already spawned earlier using only command_from_meta's cmd:env snapshot.
    // Unlike the subprocess/persistent controllers, this isn't a missing
    // plumbing call: env vars are fixed at process exec() time, so an
    // already-running App Server process cannot pick up a per-turn identity
    // change at all without being torn down and respawned. Deciding when to
    // force that respawn (e.g. on every identity-binding change vs. only
    // between turns) is a real design question, not a mechanical fix, so it's
    // deliberately left for a follow-up PR rather than rushed in here.
    if let Some(app_server_ctrl) =
        ctrl.as_any()
            .downcast_ref::<blockcontroller::app_server_controller::AppServerController>()
    {
        let hint = arm_join_hint();
        app_server_ctrl.send_message(message)?;
        hint.keep();
    }
    // Try persistent controller first, fall back to subprocess
    else if let Some(persistent_ctrl) =
        ctrl.as_any()
            .downcast_ref::<blockcontroller::persistent::PersistentSubprocessController>()
    {
        // Container agents use per-turn docker exec — incompatible with a
        // long-lived persistent subprocess. Fail loudly instead of silently
        // spawning the CLI on the host.
        let agent_mode = crate::backend::obj::meta_get_string(&block.meta, "agentMode", "host");
        if agent_mode == "container" {
            return Err("container agents require a subprocess controller; this provider uses a persistent controller".to_string());
        }
        // Resume support: a /model (or effort/permission) change
        // respawns the persistent CLI with new flags; pass the
        // resume flag + captured session id so the respawn continues
        // the same conversation. Same meta keys the subprocess path
        // reads below. Without this, switching model on a persistent
        // agent would either no-op (old behavior) or lose context.
        let resume_flag =
            crate::backend::obj::meta_get_string(&block.meta, "agent:resume_flag", "--resume");
        let persisted_session_id =
            crate::backend::obj::meta_get_string(&block.meta, "agent:sessionid", "");
        let config = blockcontroller::persistent::PersistentSpawnConfig {
            cli_command,
            cli_args,
            working_dir,
            env_vars,
            session_id_field,
            resume_flag,
            session_id: persisted_session_id,
            message_id: message_id.clone(),
        };
        let input = crate::backend::blockcontroller::health::TurnInput { origin, text: message.clone(), joins_turn };
        if let Err(e) = persistent_ctrl.send_message_with_images_from(message, inline_images, config, Some(input)) {
            // A single-live-instance refusal from inside the controller (the
            // spawn-time claim losing to another instance, or the pre-turn
            // fence) gets the same pane treatment as the early check above.
            if crate::backend::agent_admission::is_admission_refusal(&e) {
                surface_refusal(&e);
            }
            return Err(e);
        }
    } else if let Some(subprocess_ctrl) =
        ctrl.as_any()
            .downcast_ref::<blockcontroller::subprocess::SubprocessController>()
    {
        let resume_flag =
            crate::backend::obj::meta_get_string(&block.meta, "agent:resume_flag", "--resume");
        let resume_strategy = crate::backend::obj::meta_get_string(
            &block.meta,
            "agent:resume_strategy",
            if session_id_field == "thread_id" {
                "codex-exec"
            } else if resume_flag.is_empty() {
                "none"
            } else {
                "flag"
            },
        );
        // Antigravity takes its prompt as `-p <prompt>`, not on stdin.
        let prompt_arg_flag =
            crate::backend::obj::meta_get_string(&block.meta, "agent:prompt_arg_flag", "");
        // Picker reattach: the frontend writes the prior
        // block's session id here when launching with
        // `continueOfInstanceId`. spawn_turn hydrates its
        // inner.session_id from this on the first turn so
        // --resume <sid> lands on the very first launch.
        let persisted_session_id =
            crate::backend::obj::meta_get_string(&block.meta, "agent:sessionid", "");

        // Container agent branch: use Docker socket API exec (P1a: no
        // secrets in argv). Host agent branch: regular CLI subprocess.
        let agent_mode = crate::backend::obj::meta_get_string(&block.meta, "agentMode", "host");
        // Cross-process session-lease key (registry::LeaseStore) —
        // read once here for both branches below. Only the host
        // branch's spawn_turn actually enforces it in this PR;
        // the container branch's config field is unused for now
        // (struct-completeness — see host_spawn.rs's doc comment).
        let instance_id = crate::backend::obj::meta_get_string(&block.meta, "agentId", "");
        if agent_mode == "container" {
            let cm = container_manager.get().await.ok_or_else(|| {
                "Docker not available on this host; cannot start container agent".to_string()
            })?;
            let container_image =
                crate::backend::obj::meta_get_string(&block.meta, "agent:container_image", "");
            // Use agentId (UUID) — always valid as a Docker name; display names can have spaces.
            let agent_id = crate::backend::obj::meta_get_string(&block.meta, "agentId", "");
            let container_name = crate::backend::container::container_name_for_slug(&agent_id);
            let volumes_json =
                crate::backend::obj::meta_get_string(&block.meta, "agent:container_volumes", "[]");
            let volumes: Vec<String> = serde_json::from_str(&volumes_json).unwrap_or_default();

            // Mount the bound account's credentials and the agent's working
            // directory. `CLAUDE_CONFIG_DIR` was resolved into `env_vars` by the
            // identity injection above, but it is a HOST path and so is stripped
            // by CONTAINER_ENV_DENYLIST before exec — mounting it is the only way
            // the in-container CLI ever sees `.credentials.json`. Without this a
            // container agent authenticates never, no matter how many times the
            // operator logs in via Connectors → Accounts.
            let mount_spec = crate::backend::container::ContainerMountSpec {
                claude_config_host_dir: env_vars
                    .get("CLAUDE_CONFIG_DIR")
                    .and_then(|d| crate::backend::container::credentials_dir_if_file_backed(d)),
                workspace_host_dir: Some(working_dir.clone()).filter(|d| !d.is_empty()),
            };

            // AGENTMUX_LOCAL_URL is never explicitly set in env_vars for a host
            // agent either (see the PATH-block comment above) — a host
            // subprocess just inherits it from this process's own env
            // (bootstrap/network.rs). `docker exec` over the Docker socket never
            // inherits host process env, so a container agent gets no sidecar
            // URL at all unless it's injected here. Rewritten to
            // host.docker.internal (routable from inside the container via
            // the extra_hosts entry `create_and_start` sets — see
            // container.rs) since the loopback address this process sees is
            // not reachable from the container's own network namespace.
            // #2939 workstream 1 (host integration) — the other half is the
            // agentmux-mcp/agentmux-bashwrap binaries that
            // docker/Dockerfile.agent-agentmux builds into the image.
            if let Ok(local_url) = std::env::var("AGENTMUX_LOCAL_URL") {
                env_vars.insert(
                    "AGENTMUX_LOCAL_URL".to_string(),
                    crate::backend::container::rewrite_local_url_for_container(&local_url),
                );
            }

            // The command and provider the turn runs, resolved INSIDE the image
            // (on PATH, e.g. `claude`) -- not `cli_command`/`cmd`, which is the
            // host-resolved absolute npm path and does not exist in the
            // container.
            let container_command = crate::backend::obj::meta_get_string(
                &block.meta,
                "agent:container_command",
                "claude",
            );
            let agent_provider =
                crate::backend::obj::meta_get_string(&block.meta, "agentProvider", "claude");

            // Start the container (pulling the image if needed) and make sure
            // the provider CLI is installed in it. A failure is written to the
            // pane as a plain-language error before it is returned.
            super::container_turn::prepare(super::container_turn::ContainerTurnPrep {
                cm: &cm,
                block_id: &block_id,
                container_name: &container_name,
                image: &container_image,
                volumes: &volumes,
                mount_spec: &mount_spec,
                provider_id: &agent_provider,
                container_command: &container_command,
                broker: &broker,
                filestore: &filestore_gate,
                mstore: &mstore,
            })
            .await?;

            tracing::info!(
                block_id = %block_id,
                container = %container_name,
                image = %crate::backend::container_image::resolve_container_image(&container_image),
                "container agent turn: bollard exec (env via Docker socket, not argv)",
            );

            // Env is passed via CreateExecOptions.env (Docker socket API),
            // NOT as -e KEY=VALUE argv args — this prevents CWE-214 exposure.
            // spawn_container_turn filters config.env_vars (denylist) per
            // turn, so cmd:cwd (host path) and host-path vars never reach
            // the container, and each queued turn uses its own env.

            // Base cmd: [container_command, ...cli_args]. cli_args are format
            // flags (-p, --input-format ...) + provider flags -- no host
            // paths, safe as-is. spawn_container_turn appends --resume <sid>
            // internally. `agent_provider` drives the one-shot argv rebuild.
            let mut base_cmd = vec![container_command];
            base_cmd.extend(container_argv(cli_args, &agent_provider));

            let config = blockcontroller::subprocess::SubprocessSpawnConfig {
                cli_command: String::new(), // unused by spawn_container_turn
                cli_args: vec![],           // unused by spawn_container_turn
                working_dir: String::new(), // unused — container has own cwd
                env_vars,
                message: message,
                resume_flag,
                resume_strategy,
                prompt_arg_flag,
                session_id_field,
                message_id: message_id,
                session_id: if persisted_session_id.is_empty() {
                    None
                } else {
                    Some(persisted_session_id)
                },
                instance_id: instance_id.clone(),
            };
            let hint = arm_join_hint();
            subprocess_ctrl.spawn_container_turn(cm.clone(), container_name, base_cmd, config)?;
            hint.keep();
        } else {
            // Host agent: regular CLI subprocess (env set on child process, not in argv).
            let config = blockcontroller::subprocess::SubprocessSpawnConfig {
                cli_command,
                cli_args,
                working_dir,
                env_vars,
                message: message,
                resume_flag,
                resume_strategy,
                prompt_arg_flag,
                session_id_field,
                message_id: message_id,
                session_id: if persisted_session_id.is_empty() {
                    None
                } else {
                    Some(persisted_session_id)
                },
                instance_id,
            };
            let hint = arm_join_hint();
            subprocess_ctrl.spawn_turn(config)?;
            hint.keep();
        }
    } else if let Some(acp_ctrl) =
        ctrl.as_any().downcast_ref::<blockcontroller::acp::AcpController>()
    {
        // ACP agents (Copilot, OpenClaw, Pi): the message becomes a
        // `session/prompt` on the agent's running session.
        // If the agent's process exited or crashed, this starts it again from
        // the pane's meta; the message waits for the new session.
        acp_ctrl.ensure_started(block.meta.clone())?;
        let hint = arm_join_hint();
        acp_ctrl.send_message(message, message_id.as_deref())?;
        hint.keep();
    } else {
        return Err(
            "controller is not a SubprocessController, PersistentSubprocessController or AcpController"
                .to_string(),
        );
    }

    // Registration is skipped on the reactive-delivery path — see
    // `TurnRegistration::Skip` for why that is both redundant AND a hard
    // deadlock requirement, not an optimisation.
    if matches!(registration, TurnRegistration::Register) {
        // Register with cloud subscriber + reactive handler so cloud-injected
        // messages (e.g. GitHub PR review notifications) reach this agent.
        // Uses agentName (the logical display name, e.g. "smike") as the name
        // binding — matching the namespace used by `handle_reactive_register`
        // in server/reactive.rs (`req.agent_id`) and the delivery path's
        // lowercased `name_to_blocks` map — with this turn's UID as the key.
        // PR bodies embed $AGENTMUX_AGENT_ID, the stable binding (a fork's
        // `loap-3` under `Loap-3`), so both are subscribed (idempotently).
        // As its route name for every use below ("Loap #3" → "Loap-3"), or the
        // controller's own id disagrees with what a jekt resolves to (#2695).
        let display = crate::backend::obj::meta_get_string(&block.meta, "agentName", "");
        let agent_name = crate::backend::reactive::sanitize::route_name(&display).unwrap_or_default();
        if !agent_name.is_empty() {
            // Identity M2: the display binding is `agentName`, the key is
            // the UID carried in this turn's spawn env. A block that spawned
            // before its row existed (continuation eager-resume, spec §4.4.4
            // Q1) is upgraded in place here on its first turn.
            let registered = crate::backend::reactive::handler::get_global_handler()
                .register_agent_full(
                    &agent_name,
                    &block_id,
                    None,
                    0,
                    turn_agent_id.as_deref(),
                    turn_uid.as_deref(),
                    "registration.no_uid.register_tail",
                );
            if registered.is_ok() {
                // Refresh the block's OWN captured identity too
                // (reagentx P1, round 2 on #2697): this call runs on
                // EVERY turn, using block.meta["agentName"] as the
                // source of truth — which can diverge from whatever
                // a PersistentSubprocessController captured at its
                // original spawn_process call (rename, reconfigured
                // cmd:env) without the block ever respawning. Without
                // this, inject_message_inner's recipient-identity
                // check (#2695) would compare the current (correct)
                // target_agent against that stale spawn-time value
                // and falsely reject the agent's own, correctly-
                // addressed jekts as an identity mismatch. A no-op
                // for controller types that don't override
                // set_agent_id (e.g. SubprocessController).
                if let Some(ctrl) = crate::backend::blockcontroller::get_controller(&block_id) {
                    ctrl.set_agent_id(Some(agent_name.clone()));
                }
                if let Some(sub) = crate::muxbus::cloud_subscriber::get_global_subscriber() {
                    let names = crate::backend::reactive::handler::get_global_handler().names_for_block(&block_id);
                    names.iter().for_each(|n| sub.add_agent(n));
                }
                // Also mirror into the per-channel + host-global
                // shared file registries — this AgentInput/
                // SubprocessController path (unlike ShellController/
                // PersistentSubprocessController) previously only
                // ever registered in the in-memory Tier-1 map,
                // leaving it permanently unreachable via Tier 2/2b
                // cross-instance/cross-channel delivery (reagent P1,
                // third round on PR #2350).
                let data_dir = crate::backend::base::get_mux_data_dir();
                crate::backend::reactive::registry::write(
                    &data_dir,
                    &agent_name,
                    &local_web_url,
                    &block_id,
                );
                crate::backend::reactive::registry::write_shared_from_env(
                    &agent_name,
                    &local_web_url,
                    &block_id,
                );
            }
        }
    }

    Ok(())
}

pub fn register_agent_input_handlers(engine: &Arc<WshRpcEngine>, state: &AppState) {
    // subprocessspawn → spawn agent CLI as subprocess for a single turn
    let mstore_spawn = state.mstore.clone();
    let broker_spawn = state.broker.clone();
    let event_bus_spawn = state.event_bus.clone();
    let filestore_spawn = state.filestore.clone();
    let boot_id_spawn = state.boot_id.clone();
    engine.register_typed(
        COMMAND_SUBPROCESS_SPAWN,
        move |cmd: CommandSubprocessSpawnData, _ctx| {
            let mstore = mstore_spawn.clone();
            let broker = broker_spawn.clone();
            let event_bus = event_bus_spawn.clone();
            let filestore = filestore_spawn.clone();
            let boot_id = boot_id_spawn.clone();
            async move {
                tracing::info!(
                    block_id = %cmd.blockid,
                    cli = %cmd.cli_command,
                    "SubprocessSpawn"
                );

                // Get or create a SubprocessController for this block
                let ctrl = match blockcontroller::get_controller(&cmd.blockid) {
                    Some(c)
                        if c.controller_type() == blockcontroller::BLOCK_CONTROLLER_SUBPROCESS =>
                    {
                        c
                    }
                    _ => {
                        // Create and register a new SubprocessController
                        let registry = mstore.shared_agent_registry();
                        let ctrl = blockcontroller::subprocess::SubprocessController::new(
                            cmd.tabid.clone(),
                            cmd.blockid.clone(),
                            Some(broker),
                            Some(event_bus),
                            Some(mstore),
                            Some(filestore),
                            registry,
                            boot_id,
                        );
                        let ctrl = std::sync::Arc::new(ctrl);
                        ctrl.set_self_ref();
                        blockcontroller::register_controller(&cmd.blockid, ctrl.clone());
                        ctrl as std::sync::Arc<dyn blockcontroller::Controller>
                    }
                };

                // Downcast to SubprocessController to call spawn_turn
                let subprocess_ctrl = ctrl
                    .as_any()
                    .downcast_ref::<blockcontroller::subprocess::SubprocessController>()
                    .ok_or_else(|| "controller is not a SubprocessController".to_string())?;

                let config = blockcontroller::subprocess::SubprocessSpawnConfig {
                    cli_command: cmd.cli_command,
                    cli_args: cmd.cli_args,
                    working_dir: cmd.working_dir,
                    env_vars: cmd.env_vars,
                    message: cmd.message,
                    resume_flag: "--resume".to_string(),
                    resume_strategy: "flag".to_string(),
                    prompt_arg_flag: String::new(),
                    session_id_field: "session_id".to_string(),
                    message_id: None,
                    // Direct-spawn legacy command — caller doesn't
                    // carry a reattach context. Greenfield session id
                    // is None; spawn_turn captures it from CLI stdout
                    // on the first turn as before.
                    session_id: None,
                    // COMMAND_SUBPROCESS_SPAWN has no block to read
                    // agentId from (dead code — unused by the
                    // frontend, per grep); empty disables leasing.
                    instance_id: String::new(),
                };
                subprocess_ctrl.spawn_turn(config)?;
                Ok(())
            }
        },
    );

    // agentinput → send message to agent (persistent or per-turn subprocess).
    // Everything the turn needs from AppState now travels as one
    // `AgentTurnDeps` (see its doc comment for why it is a named struct rather
    // than nine separate closure captures).
    let deps_ai = AgentTurnDeps::from_state(state);
    engine.register_typed(
        COMMAND_AGENT_INPUT,
        move |cmd: CommandAgentInputData, _ctx| {
            let deps = deps_ai.clone();
            async move {
                tracing::info!(block_id = %cmd.blockid, "AgentInput");
                // Authoritative hidden-window gate for ambient digest reads
                // (next_prompt_suggestion / activity_summary / activity_watcher)
                // — set here, not reconstructed from transcript bytes. See
                // crate::ambient::digest::set_hidden_reinjection_active's doc comment.
                crate::ambient::digest::set_hidden_reinjection_active(
                    &cmd.blockid,
                    cmd.hidden.unwrap_or(false),
                );
                // A re-delivery srv composed (`memorydelivery:compose`): its
                // card is written once the message has gone to the agent
                // (below), never for one that didn't start.
                let fallback_card = cmd
                    .hidden
                    .unwrap_or(false)
                    .then_some(())
                    .and(cmd.delivery_id.as_deref())
                    .and_then(|id| crate::server::memory_delivery_handlers::take_fallback_frame(&cmd.blockid, id));
                // A post-compaction memory reinjection also carries the
                // agent's running summary (SPEC_DURABLE_CONVERSATION_MEMORY
                // §4.4, "Compaction reuse").
                let message = if cmd.hidden.unwrap_or(false) {
                    crate::backend::continuity_state::with_state_after_compaction(&deps.mstore, &cmd.blockid, cmd.message)
                } else {
                    cmd.message
                };
                // The pane's composer: the user — except a hidden memory
                // reinjection, which the frontend sends on its own behalf.
                let origin = if cmd.hidden.unwrap_or(false) {
                    crate::backend::blockcontroller::health::TurnOrigin::System
                } else {
                    crate::backend::blockcontroller::health::TurnOrigin::User
                };
                let block_id = cmd.blockid.clone();
                run_agent_turn_joining(
                    &deps,
                    cmd.blockid,
                    message,
                    cmd.message_id,
                    TurnRegistration::Register,
                    origin,
                    cmd.attachments.unwrap_or_default(),
                    cmd.joins_turn,
                )
                .await?;
                // Right after the hidden message's own transcript line: replay
                // shows this card in its place (parseHistoryLines.ts).
                if let Some(frame) = fallback_card {
                    let line = format!("{frame}\n");
                    let zone = crate::backend::blockcontroller::shell::resolve_global_output_zone(
                        &Some(deps.mstore.clone()),
                        &block_id,
                    );
                    crate::backend::blockcontroller::shell::handle_append_block_file(
                        &deps.broker,
                        &block_id,
                        crate::backend::blockcontroller::persistent::PERSISTENT_OUTPUT_SUBJECT,
                        line.as_bytes(),
                        Some(&deps.filestore_gate),
                        zone.as_deref(),
                    );
                }
                Ok(())
            }
        },
    );

    // agentstop → stop the running agent subprocess
    engine.register_typed(
        COMMAND_AGENT_STOP,
        |cmd: CommandAgentStopData, _ctx| {
            async move {
                tracing::info!(block_id = %cmd.blockid, force = cmd.force, "AgentStop");
                match blockcontroller::get_controller(&cmd.blockid) {
                    Some(_) => {
                        // The one teardown path, Stop policy (SPEC_AGENT_TEARDOWN_SINGLE_PATH §5).
                        crate::sagas::agent_teardown::stop_now(&cmd.blockid, !cmd.force)?;
                        // Deregister: unregister_block cleans up both agent_to_block and
                        // block_to_agent maps; remove_agent then removes the cloud poll entry
                        // using the logical agent_id recovered from block_to_agent.
                        let handler = crate::backend::reactive::handler::get_global_handler();
                        // Identity M2: a block can be bound under two names
                        // (display + stable); tear down the name-keyed side
                        // state for EACH, not just the display name — the
                        // stable name's file entry was written at spawn and
                        // its cloud subscription would otherwise leak.
                        let names = handler.unregister_block(&cmd.blockid);
                        let data_dir = crate::backend::base::get_mux_data_dir();
                        let sub = crate::muxbus::cloud_subscriber::get_global_subscriber();
                        for name in &names {
                            // Symmetric teardown for the registry writes added
                            // alongside SubprocessSpawn's register_agent call.
                            crate::backend::reactive::registry::remove(&data_dir, name);
                            crate::backend::reactive::registry::remove_shared_from_env(name);
                            if let Some(sub) = sub {
                                sub.remove_agent(name);
                            }
                        }
                        Ok(())
                    }
                    // Stopping a block with no controller is a no-op, not an
                    // error: the UI cannot tell "already stopped" from "never
                    // started", and either way the caller's intent is
                    // satisfied.
                    None => Ok(()),
                }
            }
        },
    );
}

#[cfg(test)]
mod tests {
    use super::git_identity_env_vars;
    use super::{container_argv, flags_with_arity, strip_flag_with_value};

    fn v(xs: &[&str]) -> Vec<String> {
        xs.iter().map(|s| s.to_string()).collect()
    }

    /// The exact argv a container agent carried before the fix — the persistent
    /// controller's args, copied verbatim from the live broken block (Moras,
    /// 2026-08-31).
    fn stale_persistent_argv() -> Vec<String> {
        v(&[
            "--input-format",
            "stream-json",
            "--output-format",
            "stream-json",
            "--verbose",
            "--include-partial-messages",
            "--permission-prompt-tool",
            "stdio",
            "--permission-mode",
            "default",
            "--model",
            "sonnet",
            "--effort",
            "high",
        ])
    }

    /// All three persistent-only defects must go, not just the fatal one
    /// (codex P1 on PR #2867): the parse crash, the missing one-shot `-p`, and
    /// the control-protocol permission flags a container turn cannot answer.
    #[test]
    fn rebuilds_a_stale_persistent_argv_into_the_one_shot_form() {
        let got = container_argv(stale_persistent_argv(), "claude");

        assert!(
            !got.iter().any(|a| a == "--input-format"),
            "the fatal parse flag must go"
        );
        assert!(
            got.iter().any(|a| a == "-p"),
            "one-shot print mode must be present"
        );
        assert!(
            !got.iter().any(|a| a == "--permission-prompt-tool"),
            "the container turn has no control channel to answer can_use_tool",
        );
        assert!(
            !got.iter().any(|a| a == "stdio"),
            "its value token must go with it"
        );
        assert!(
            !got.iter().any(|a| a == "--permission-mode"),
            "non-bypass permission mode needs the control protocol",
        );
        assert!(
            got.iter().any(|a| a == "--output-format"),
            "output-format stream-json is how the pane parses the turn at all",
        );
    }

    /// A user's model/effort selections must survive the heal — silently
    /// resetting someone's model on every stale pane would be its own bug.
    #[test]
    fn preserves_the_users_model_and_effort_choices() {
        let got = container_argv(stale_persistent_argv(), "claude");
        let pos = |f: &str| got.iter().position(|a| a == f);
        assert_eq!(got[pos("--model").expect("--model kept") + 1], "sonnet");
        assert_eq!(got[pos("--effort").expect("--effort kept") + 1], "high");
    }

    /// reagent P1 on PR #2867. This runs on EVERY container turn, not once per
    /// stale block, so it must not touch an argv the frontend already built
    /// correctly — a rebuild-from-baseline silently deleted `provider_flags`
    /// and `--fork-session` on every single turn, forever.
    #[test]
    fn leaves_an_already_correct_one_shot_argv_completely_untouched() {
        let correct = v(&[
            "-p",
            "--output-format",
            "stream-json",
            "--verbose",
            "--include-partial-messages",
            "--dangerously-skip-permissions",
            "--exclude-dynamic-system-prompt-sections",
            "--model",
            "opus",
            "--my-custom-provider-flag",
            "42",             // agent.provider_flags
            "--fork-session", // resolveForkSessionArgs
        ]);
        assert_eq!(container_argv(correct.clone(), "claude"), correct);
    }

    /// …and when it DOES heal a stale argv, those same user-owned tokens still
    /// have to come through. Healing is subtraction, not reconstruction.
    #[test]
    fn a_heal_preserves_provider_flags_and_fork_session_too() {
        let mut stale = stale_persistent_argv();
        stale.extend(v(&["--my-custom-provider-flag", "42", "--fork-session"]));

        let got = container_argv(stale, "claude");

        assert!(
            got.iter().any(|a| a == "--my-custom-provider-flag"),
            "provider_flags survive"
        );
        assert_eq!(
            got[got
                .iter()
                .position(|a| a == "--my-custom-provider-flag")
                .unwrap()
                + 1],
            "42",
            "…with its value",
        );
        assert!(
            got.iter().any(|a| a == "--fork-session"),
            "fork-session survives"
        );
        assert!(
            !got.iter().any(|a| a == "--input-format"),
            "while still being healed"
        );
        assert!(got.iter().any(|a| a == "-p"));
    }

    /// An unknown provider has no baseline to diff against, so fall back to
    /// removing only the outright-fatal flag rather than guessing.
    #[test]
    fn an_unknown_provider_falls_back_to_removing_only_the_fatal_flag() {
        let got = container_argv(
            v(&["--input-format", "stream-json", "--custom"]),
            "no-such-provider",
        );
        assert_eq!(got, v(&["--custom"]));
    }

    /// A subprocess-shaped provider has no persistent argv, so nothing it was
    /// ever launched with could need healing.
    #[test]
    fn a_provider_with_no_persistent_variant_is_passed_through() {
        let argv = v(&["--json", "--whatever"]);
        assert_eq!(container_argv(argv.clone(), "codex"), argv);
    }

    /// Arity comes off the catalog, not a hardcoded table — this is what lets
    /// a valued flag drop its value and a bare flag not eat the next one.
    #[test]
    fn flags_with_arity_reads_valued_and_bare_flags_off_the_list() {
        static ARGS: &[&str] = &["-p", "--output-format", "stream-json", "--verbose"];
        assert_eq!(
            flags_with_arity(ARGS),
            vec![
                ("-p", false),
                ("--output-format", true),
                ("--verbose", false)
            ],
        );
    }

    #[test]
    fn strip_flag_with_value_removes_every_occurrence_and_tolerates_a_missing_value() {
        assert_eq!(
            strip_flag_with_value(v(&["--x", "1", "--keep", "--x", "2"]), "--x"),
            v(&["--keep"]),
        );
        assert_eq!(
            strip_flag_with_value(v(&["--keep", "--x"]), "--x"),
            v(&["--keep"])
        );
        assert_eq!(strip_flag_with_value(vec![], "--x"), Vec::<String>::new());
    }

    #[test]
    fn maps_agent_id_to_name_and_placeholder_email() {
        let vars = git_identity_env_vars("korp");
        let map: std::collections::HashMap<&str, String> = vars.into_iter().collect();
        assert_eq!(map["GIT_AUTHOR_NAME"], "korp");
        assert_eq!(map["GIT_COMMITTER_NAME"], "korp");
        assert_eq!(map["GIT_AUTHOR_EMAIL"], "korp@agentmux.local");
        assert_eq!(map["GIT_COMMITTER_EMAIL"], "korp@agentmux.local");
    }

    #[test]
    fn lowercases_email_but_preserves_display_name_casing() {
        // AGENTMUX_AGENT_ID is natural display casing (e.g. "Korp", per
        // SPEC_PR_TITLE_AGENT_HOST_PREFIX_2026_08_22.md's distinction
        // between the tag's lowercase machine-key and the title's natural
        // casing) -- the git *name* field should read naturally too, but
        // the email's local-part must stay lowercase so it can never
        // collide with a differently-cased but same agent (git/GitHub
        // treat email local-parts as effectively case-sensitive strings
        // for linking purposes; we want exactly one canonical email per
        // agent regardless of what casing happened to be in the block's
        // agentName metadata at spawn time).
        let vars = git_identity_env_vars("Korp");
        let map: std::collections::HashMap<&str, String> = vars.into_iter().collect();
        assert_eq!(map["GIT_AUTHOR_NAME"], "Korp");
        assert_eq!(map["GIT_AUTHOR_EMAIL"], "korp@agentmux.local");
    }

    #[test]
    fn distinct_agents_get_distinct_non_colliding_identities() {
        let a = git_identity_env_vars("agenty");
        let b = git_identity_env_vars("smike");
        let a_map: std::collections::HashMap<&str, String> = a.into_iter().collect();
        let b_map: std::collections::HashMap<&str, String> = b.into_iter().collect();
        assert_ne!(a_map["GIT_AUTHOR_EMAIL"], b_map["GIT_AUTHOR_EMAIL"]);
        assert_ne!(a_map["GIT_AUTHOR_NAME"], b_map["GIT_AUTHOR_NAME"]);
    }

    // ---- identity M0: split_agent_identity_env ---------------------------
    //
    // Fixture per SPEC_AGENT_IDENTITY_CARRIED_NOT_DERIVED_2026_09_23.md §9.3:
    // two agents whose names collide under `derive_slug` ("AgentY" and
    // "AGENTY" both lowercase to "agenty"), which `agent_def_insert` gave
    // the suffixed slugs `agenty-2` / `agenty-3`. Every assertion below is
    // made against that pair so a change that only works for one agent
    // cannot pass.

    use super::split_agent_identity_env;
    use std::collections::HashMap;

    fn env(pairs: &[(&str, &str)]) -> HashMap<String, String> {
        pairs
            .iter()
            .map(|(k, v)| (k.to_string(), v.to_string()))
            .collect()
    }

    /// The server launch path (`input.rs` fallback) — `cmd:env` carries
    /// nothing identity-shaped, so both must be filled from what the server
    /// knows. Positive case first (retro §5): prove they are SET before
    /// proving anything about what they are not.
    #[test]
    fn server_path_sets_both_display_and_slug_from_server_knowledge() {
        let mut e = env(&[("AGENTMUX_AGENT_ID", "AgentY")]);
        split_agent_identity_env(&mut e, "AgentY", Some("agenty-3"));
        assert_eq!(e["AGENTMUX_AGENT_DISPLAY"], "AgentY");
        assert_eq!(e["AGENTMUX_AGENT_SLUG"], "agenty-3");
    }

    /// The frontend launch path (`launchAgentDefinition`) already set both
    /// to the same values the server holds — nothing changes, nothing is
    /// duplicated under a different key.
    #[test]
    fn frontend_path_with_agreeing_values_is_left_as_is() {
        let before = env(&[
            ("AGENTMUX_AGENT_ID", "agenty-2"),
            ("AGENTMUX_AGENT_SLUG", "agenty-2"),
            ("AGENTMUX_AGENT_DISPLAY", "AGENTY"),
        ]);
        let mut e = before.clone();
        split_agent_identity_env(&mut e, "AGENTY", Some("agenty-2"));
        assert_eq!(e, before);
    }

    /// §9.4: this phase must not change what `AGENTMUX_AGENT_ID` holds on
    /// either path. Slug-valued on the frontend path, name-valued on the
    /// server path — both survive byte-for-byte.
    #[test]
    fn agentmux_agent_id_is_never_touched_on_either_path() {
        let mut frontend = env(&[("AGENTMUX_AGENT_ID", "agenty-2")]);
        split_agent_identity_env(&mut frontend, "AGENTY", Some("agenty-2"));
        assert_eq!(frontend["AGENTMUX_AGENT_ID"], "agenty-2");

        let mut server = env(&[("AGENTMUX_AGENT_ID", "AgentY")]);
        split_agent_identity_env(&mut server, "AgentY", Some("agenty-3"));
        assert_eq!(server["AGENTMUX_AGENT_ID"], "AgentY");

        let mut none = env(&[]);
        split_agent_identity_env(&mut none, "AgentY", Some("agenty-3"));
        assert!(
            !none.contains_key("AGENTMUX_AGENT_ID"),
            "must not invent one either"
        );
    }

    /// The server's knowledge wins over a `cmd:env` snapshot: a rename after
    /// launch moves `block.meta["agentName"]` but not the frozen env copy,
    /// and the persisted slug is the collision-resolved one, not whatever
    /// the frontend computed before the row existed.
    #[test]
    fn server_values_override_stale_cmd_env_snapshots() {
        let mut e = env(&[
            ("AGENTMUX_AGENT_DISPLAY", "AgentY"),
            ("AGENTMUX_AGENT_SLUG", "agenty"),
        ]);
        split_agent_identity_env(&mut e, "AgentY (renamed)", Some("agenty-3"));
        assert_eq!(e["AGENTMUX_AGENT_DISPLAY"], "AgentY (renamed)");
        assert_eq!(e["AGENTMUX_AGENT_SLUG"], "agenty-3");
    }

    /// A quick-launch pane (`launchAgent`, provider key as `agentId`) has no
    /// `db_agents` row and therefore no slug. The display name is still set;
    /// the slug is NOT derived from it — that derivation is the lossy step
    /// spec §1.1 identifies as the defect. A `cmd:env` slug survives as the
    /// fallback.
    #[test]
    fn no_persisted_row_sets_display_only_and_keeps_any_cmd_env_slug() {
        let mut bare = env(&[]);
        split_agent_identity_env(&mut bare, "AgentY", None);
        assert_eq!(bare["AGENTMUX_AGENT_DISPLAY"], "AgentY");
        assert!(!bare.contains_key("AGENTMUX_AGENT_SLUG"));

        let mut with_fallback = env(&[("AGENTMUX_AGENT_SLUG", "agenty-3")]);
        split_agent_identity_env(&mut with_fallback, "AgentY", None);
        assert_eq!(with_fallback["AGENTMUX_AGENT_SLUG"], "agenty-3");
    }

    /// An empty or whitespace display name (a pane with no `agentName` meta
    /// yet) sets nothing and does not clobber a `cmd:env` value with "".
    #[test]
    fn empty_display_name_sets_nothing_and_preserves_cmd_env() {
        let mut bare = env(&[]);
        split_agent_identity_env(&mut bare, "   ", None);
        assert!(!bare.contains_key("AGENTMUX_AGENT_DISPLAY"));

        let mut with_fallback = env(&[("AGENTMUX_AGENT_DISPLAY", "AgentY")]);
        split_agent_identity_env(&mut with_fallback, "", Some("  "));
        assert_eq!(with_fallback["AGENTMUX_AGENT_DISPLAY"], "AgentY");
        assert!(!with_fallback.contains_key("AGENTMUX_AGENT_SLUG"));
    }

    /// The whole point: after the split, the two colliding agents are
    /// distinguishable by BOTH variables, whereas anything derived from the
    /// display name alone is not.
    #[test]
    fn colliding_names_stay_separable_after_the_split() {
        let mut a = env(&[("AGENTMUX_AGENT_ID", "AgentY")]);
        let mut b = env(&[("AGENTMUX_AGENT_ID", "AGENTY")]);
        split_agent_identity_env(&mut a, "AgentY", Some("agenty-3"));
        split_agent_identity_env(&mut b, "AGENTY", Some("agenty-2"));

        assert_ne!(a["AGENTMUX_AGENT_SLUG"], b["AGENTMUX_AGENT_SLUG"]);
        assert_ne!(a["AGENTMUX_AGENT_DISPLAY"], b["AGENTMUX_AGENT_DISPLAY"]);
        // …and the reason the slug has to come from the store rather than
        // from the name: lowercased, the names are the same string.
        assert_eq!(
            a["AGENTMUX_AGENT_DISPLAY"].to_lowercase(),
            b["AGENTMUX_AGENT_DISPLAY"].to_lowercase(),
        );
    }

    // ---- identity M1: carry_agent_uid_env --------------------------------

    use super::{carry_agent_uid_env, PersistedAgentIdentity};

    fn identity(uid: &str, token: Option<&str>) -> PersistedAgentIdentity {
        PersistedAgentIdentity {
            uid: uid.to_string(),
            slug: Some("agenty-3".to_string()),
            token: token.map(str::to_string),
        }
    }

    /// Positive case first: a known row carries both values into the env.
    #[test]
    fn carry_sets_uid_and_token_from_the_row() {
        let mut e = env(&[("AGENTMUX_AGENT_ID", "AgentY")]);
        carry_agent_uid_env(&mut e, Some(&identity("4f3c-a91", Some("tok-a"))));
        assert_eq!(e["AGENTMUX_AGENT_UID"], "4f3c-a91");
        assert_eq!(e["AGENTMUX_AGENT_TOKEN"], "tok-a");
        assert_eq!(e["AGENTMUX_AGENT_ID"], "AgentY", "§9.4: untouched");
    }

    /// Reserved-variable rule: a persisted `cmd:env` carrying ANOTHER
    /// agent's UID and token (pane reuse, copied block) is overwritten, not
    /// respected. For the token this is the difference between an agent
    /// holding its own credential and holding someone else's.
    #[test]
    fn carry_overwrites_stale_cmd_env_values_unconditionally() {
        let mut e = env(&[
            ("AGENTMUX_AGENT_UID", "9b2e-7d4"),
            ("AGENTMUX_AGENT_TOKEN", "tok-of-the-other-agent"),
        ]);
        carry_agent_uid_env(&mut e, Some(&identity("4f3c-a91", Some("tok-a"))));
        assert_eq!(e["AGENTMUX_AGENT_UID"], "4f3c-a91");
        assert_eq!(e["AGENTMUX_AGENT_TOKEN"], "tok-a");
    }

    /// No row (quick-launch pane, lookup failure): both are REMOVED. Leaving
    /// a stale token in place would be worse than having none, and the
    /// server has nothing truthful to put there.
    #[test]
    fn carry_removes_both_when_the_server_knows_no_identity() {
        let mut e = env(&[
            ("AGENTMUX_AGENT_ID", "AgentY"),
            ("AGENTMUX_AGENT_UID", "stale"),
            ("AGENTMUX_AGENT_TOKEN", "stale"),
        ]);
        carry_agent_uid_env(&mut e, None);
        assert!(!e.contains_key("AGENTMUX_AGENT_UID"));
        assert!(!e.contains_key("AGENTMUX_AGENT_TOKEN"));
        assert_eq!(e["AGENTMUX_AGENT_ID"], "AgentY");
    }

    /// A row whose token could not be minted still carries its UID, and a
    /// stale token is removed rather than kept — never a half-truth.
    #[test]
    fn carry_with_uid_but_no_token_sets_uid_and_removes_token() {
        let mut e = env(&[("AGENTMUX_AGENT_TOKEN", "stale")]);
        carry_agent_uid_env(&mut e, Some(&identity("4f3c-a91", None)));
        assert_eq!(e["AGENTMUX_AGENT_UID"], "4f3c-a91");
        assert!(!e.contains_key("AGENTMUX_AGENT_TOKEN"));
    }

    // ---- identity M0/M1: persisted_agent_identity against a real store ---

    use super::persisted_agent_identity;
    use crate::backend::storage::agents::AgentDefinition;
    use crate::backend::storage::store::Store;

    fn agent_def(id: &str, name: &str) -> AgentDefinition {
        AgentDefinition {
            conversation_visibility:
                crate::backend::storage::agents::default_conversation_visibility(),
            id: id.to_string(),
            // Empty on purpose: `agent_def_insert` derives it from `name`
            // and collision-resolves, exactly as the launch flow does.
            slug: String::new(),
            name: name.to_string(),
            icon: "✦".to_string(),
            provider: "claude".to_string(),
            description: String::new(),
            working_directory: String::new(),
            shell: String::new(),
            provider_flags: String::new(),
            auto_start: 0,
            restart_on_crash: 0,
            idle_timeout_minutes: 0,
            created_at: 0,
            agent_type: "host".to_string(),
            environment: String::new(),
            agent_bus_id: String::new(),
            is_seeded: 0,
            accounts: String::new(),
            parent_id: String::new(),
            branch_label: String::new(),
            updated_at: 0,
            user_hidden: 0,
            container_image: String::new(),
            container_volumes: "[]".to_string(),
            container_name: String::new(),
            use_ambient_login: 0,
            auto_continue_enabled: 0,
            model_vendor_base_url: String::new(),
            memory_id: String::new(),
        }
    }

    fn block_with_agent_id(store: &Store, block_id: &str, agent_id: &str) {
        let mut block = crate::backend::obj::Block {
            oid: block_id.to_string(),
            parentoref: String::new(),
            version: 0,
            runtimeopts: None,
            stickers: None,
            meta: {
                let mut m = crate::backend::obj::MetaMapType::new();
                m.insert("view".to_string(), serde_json::json!("agent"));
                m.insert("agentId".to_string(), serde_json::json!(agent_id));
                m
            },
            subblockids: None,
        };
        store.insert(&mut block).unwrap();
    }

    /// The §1.1 scenario end to end: two agents whose names collide under
    /// `derive_slug` get `agenty` and `agenty-2` from `agent_def_insert`, and
    /// the lookup returns each block's OWN row — its UID, its suffixed slug
    /// (which no derivation from its name could produce), and a token that
    /// is that row's and nobody else's.
    #[test]
    fn persisted_identity_returns_the_rows_uid_slug_and_own_token() {
        let store = Store::open_in_memory().unwrap();
        let mut first = agent_def("uid-first", "AgentY");
        store.agent_def_insert(&mut first).unwrap();
        let mut second = agent_def("uid-second", "AGENTY");
        store.agent_def_insert(&mut second).unwrap();
        assert_eq!(
            first.slug, "agenty",
            "fixture: first insert keeps the bare slug"
        );
        assert_eq!(
            second.slug, "agenty-2",
            "fixture: second insert is suffixed"
        );

        block_with_agent_id(&store, "block-first", "uid-first");
        block_with_agent_id(&store, "block-second", "uid-second");

        let a = persisted_agent_identity(&store, "block-first").expect("first resolves");
        let b = persisted_agent_identity(&store, "block-second").expect("second resolves");
        assert_eq!(a.uid, "uid-first");
        assert_eq!(b.uid, "uid-second");
        assert_eq!(a.slug.as_deref(), Some("agenty"));
        assert_eq!(b.slug.as_deref(), Some("agenty-2"));

        // The token is the persisted one for that UID, minted on this first
        // spawn, and distinct between the two colliding agents.
        let tok_a = a.token.clone().expect("token minted");
        let tok_b = b.token.clone().expect("token minted");
        assert_ne!(tok_a, tok_b);
        assert_eq!(
            store.agent_token_load("uid-first").unwrap().as_deref(),
            Some(tok_a.as_str())
        );
        assert_eq!(
            store.agent_token_load("uid-second").unwrap().as_deref(),
            Some(tok_b.as_str())
        );

        // A second spawn of the same agent carries the SAME token — long-lived
        // for the agent's lifetime (spec §6.3), not re-minted per spawn.
        let a_again = persisted_agent_identity(&store, "block-first").unwrap();
        assert_eq!(a_again.token.as_deref(), Some(tok_a.as_str()));
    }

    /// Identity M4b-2: the helper App Server and ACP use carries the
    /// block's row UID and token into their own env, overwriting a stale
    /// token, and strips both for a block with no row — outside a runtime
    /// and, through `block_in_place`, on a multi-thread runtime worker.
    #[test]
    fn carry_agent_workdir_env_is_the_launch_cwd_for_host_agents_only() {
        let meta = |pairs: &[(&str, &str)]| -> crate::backend::obj::MetaMapType {
            pairs.iter().map(|(k, v)| (k.to_string(), serde_json::Value::String(v.to_string()))).collect()
        };
        let stale = || {
            std::collections::HashMap::from([("AGENTMUX_AGENT_WORKDIR".to_string(), "C:/stale".to_string())])
        };

        let workdir = |e: &std::collections::HashMap<String, String>| std::path::PathBuf::from(&e["AGENTMUX_AGENT_WORKDIR"]);
        let ws = std::env::temp_dir().join("clamk-0612a");
        let ws_str = ws.to_string_lossy().into_owned();

        for mode in ["host", "standalone"] {
            let mut e = stale();
            super::carry_agent_workdir_env(&mut e, &meta(&[("agentMode", mode), ("cmd:cwd", &ws_str)]));
            assert_eq!(workdir(&e), ws, "{mode}");
        }
        // No agentMode is a host agent, srv's default.
        let mut e = stale();
        super::carry_agent_workdir_env(&mut e, &meta(&[("cmd:cwd", &ws_str)]));
        assert_eq!(workdir(&e), ws);

        // Exported as the absolute path the process starts in: a relative
        // cwd resolves against srv's own cwd, `~` against the home dir.
        let mut e = stale();
        super::carry_agent_workdir_env(&mut e, &meta(&[("cmd:cwd", "projects/foo")]));
        assert_eq!(workdir(&e), std::env::current_dir().unwrap().join("projects/foo"));
        if let Some(home) = dirs::home_dir() {
            let mut e = stale();
            super::carry_agent_workdir_env(&mut e, &meta(&[("cmd:cwd", "~/.agentmux/agents/x-0930a")]));
            assert_eq!(workdir(&e), home.join(".agentmux/agents/x-0930a"));
        }

        let mut e = stale();
        super::carry_agent_workdir_env(&mut e, &meta(&[("agentMode", "container"), ("cmd:cwd", &ws_str)]));
        assert!(!e.contains_key("AGENTMUX_AGENT_WORKDIR"), "container");
        let mut e = stale();
        super::carry_agent_workdir_env(&mut e, &meta(&[("agentMode", "host"), ("cmd:cwd", "  ")]));
        assert!(!e.contains_key("AGENTMUX_AGENT_WORKDIR"), "no cwd");
    }

    #[test]
    fn carry_block_identity_env_carries_or_strips() {
        let store = Store::open_in_memory().unwrap();
        let mut def = agent_def("uid-carry", "Carrier");
        store.agent_def_insert(&mut def).unwrap();
        block_with_agent_id(&store, "block-carry", "uid-carry");
        block_with_agent_id(&store, "block-carry-none", "claude");

        let mut env = std::collections::HashMap::from([(
            "AGENTMUX_AGENT_TOKEN".to_string(),
            "stale-token".to_string(),
        )]);
        super::carry_block_identity_env(&store, "block-carry", &mut env);
        assert_eq!(
            env.get("AGENTMUX_AGENT_UID").map(String::as_str),
            Some("uid-carry")
        );
        assert_eq!(
            env.get("AGENTMUX_AGENT_TOKEN"),
            store.agent_token_load("uid-carry").unwrap().as_ref()
        );

        let mut none = env.clone();
        super::carry_block_identity_env(&store, "block-carry-none", &mut none);
        assert!(
            !none.contains_key("AGENTMUX_AGENT_UID") && !none.contains_key("AGENTMUX_AGENT_TOKEN")
        );

        let rt = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .build()
            .unwrap();
        let store = std::sync::Arc::new(store);
        let on_worker = rt.block_on({
            let store = store.clone();
            async move {
                tokio::spawn(async move {
                    let mut env = std::collections::HashMap::new();
                    super::carry_block_identity_env(&store, "block-carry", &mut env);
                    env
                })
                .await
                .unwrap()
            }
        });
        assert_eq!(
            on_worker.get("AGENTMUX_AGENT_UID").map(String::as_str),
            Some("uid-carry")
        );
    }

    /// A quick-launch pane carries a provider key as `agentId` and has no
    /// `db_agents` row; a block that does not exist has nothing either.
    /// Both are "no identity", never an error and never a derived value —
    /// and, load-bearing for the token: no row means no token is minted.
    #[test]
    fn persisted_identity_is_none_for_quick_launch_panes_and_missing_blocks() {
        let store = Store::open_in_memory().unwrap();
        block_with_agent_id(&store, "block-quick", "claude");
        assert_eq!(persisted_agent_identity(&store, "block-quick"), None);
        assert_eq!(persisted_agent_identity(&store, "block-missing"), None);
        assert_eq!(
            store.agent_token_load("claude").unwrap(),
            None,
            "no token minted for a provider key"
        );
    }

    /// Codex P1 on #3591: the builder re-checks deletion at its last read —
    /// an agent deleted after the gate's lookup (no row resolves now, and the
    /// shared registry holds only a retired record) is reported, so the spawn
    /// is refused; a live agent is not.
    #[test]
    fn the_builder_recheck_reports_an_agent_deleted_after_the_gate() {
        let tmp = tempfile::tempdir().unwrap();
        let store = Store::open_in_memory().unwrap();
        let defs = crate::registry::DefinitionStore::open(tmp.path().join("definitions")).unwrap();
        store.set_def_registry(std::sync::Arc::new(defs));
        let mut def = agent_def("uid-late-del", "LateDel");
        store.agent_def_insert(&mut def).unwrap();
        block_with_agent_id(&store, "block-late-del", "uid-late-del");
        let live = persisted_agent_identity(&store, "block-late-del");
        assert!(live.is_some());
        assert_eq!(
            super::deleted_agent_of_rowless_block(&store, "block-late-del", live.as_ref()),
            None
        );

        assert!(store.agent_def_delete("uid-late-del").unwrap());
        let after = persisted_agent_identity(&store, "block-late-del");
        assert_eq!(after, None);
        assert_eq!(
            super::deleted_agent_of_rowless_block(&store, "block-late-del", after.as_ref())
                .as_deref(),
            Some("uid-late-del")
        );
    }

    /// Spec §6.3: revoked on deletion. Deleting the agent through either
    /// deletion entry point takes its token with it, so a later spawn on a
    /// reused block cannot resurrect the old credential.
    #[test]
    fn deleting_the_agent_revokes_its_token() {
        let store = Store::open_in_memory().unwrap();
        let mut def = agent_def("uid-del", "Deleted");
        store.agent_def_insert(&mut def).unwrap();
        block_with_agent_id(&store, "block-del", "uid-del");
        let minted = persisted_agent_identity(&store, "block-del").unwrap().token;
        assert!(minted.is_some());

        assert!(store.agent_def_delete("uid-del").unwrap());
        assert_eq!(store.agent_token_load("uid-del").unwrap(), None);
        assert_eq!(persisted_agent_identity(&store, "block-del"), None);

        // …and via the launch-side entry point too.
        let mut def2 = agent_def("uid-del2", "Deleted2");
        store.agent_def_insert(&mut def2).unwrap();
        block_with_agent_id(&store, "block-del2", "uid-del2");
        assert!(persisted_agent_identity(&store, "block-del2")
            .unwrap()
            .token
            .is_some());
        assert!(store.instance_delete("uid-del2").unwrap());
        assert_eq!(store.agent_token_load("uid-del2").unwrap(), None);
    }

    #[test]
    fn email_domain_is_not_a_real_github_verifiable_domain() {
        // Load-bearing for the fix's safety property: this must NOT be a
        // domain GitHub could ever link to a real account (see the
        // git_identity_env_vars doc comment) -- asserting the exact
        // domain here so a future edit can't accidentally change it to
        // something real (e.g. "users.noreply.github.com") without this
        // test catching it.
        let vars = git_identity_env_vars("camper");
        let map: std::collections::HashMap<&str, String> = vars.into_iter().collect();
        assert!(map["GIT_AUTHOR_EMAIL"].ends_with("@agentmux.local"));
        assert!(map["GIT_COMMITTER_EMAIL"].ends_with("@agentmux.local"));
    }
}