agentmux_srv/server/app_api/

agent_io.rs

use super::*;

pub fn register(engine: &Arc<WshRpcEngine>, state: &AppState) {
    register_agent_send(engine, state);
    register_agent_stop(engine, state);
    register_agent_status(engine, state);
    register_agent_list(engine, state);
    register_agent_output(engine, state);
    register_agent_process_list(engine, state);
    register_agent_tracked_blocks(engine, state);
    register_swarm_other_instances(engine, state);
    register_ambient_outcomes(engine);
    register_agent_open_panes(engine, state);
    register_agent_kill_process(engine, state);
    register_agent_kill_tree(engine, state);
}

fn register_agent_process_list(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let process_tracker = state.process_tracker.clone();
    engine.register_typed(COMMAND_AGENT_PROCESS_LIST, move |cmd: AgentProcessListCommand, _ctx| {
        let process_tracker = process_tracker.clone();
        async move {
                let members = process_tracker.list_block(&cmd.block_id);
                let confidence = process_tracker.confidence_of(&cmd.block_id);
                let processes: Vec<AgentProcessInfo> = members
                    .into_iter()
                    .map(|m| AgentProcessInfo {
                        pid: m.pid,
                        command: m.command,
                        rss_bytes: m.rss_bytes,
                        started_at_ms: m.started_at_ms,
                    })
                    .collect();
                Ok(AgentProcessListResult {
                    block_id: cmd.block_id,
                    confidence,
                    processes,
                })
        }
    });
}

fn register_agent_tracked_blocks(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let process_broker = state.process_broker.clone();
    // No argument: `Value` takes whatever the caller sends.
    engine.register_typed(COMMAND_AGENT_TRACKED_BLOCKS, move |_data: serde_json::Value, _ctx| {
        let process_broker = process_broker.clone();
        async move {
                // Was: `process_tracker.list_all_blocks()` unioned with
                // `reactive::get_global_handler().list_active_blocks()` — two
                // structurally different, independently-populated registries
                // concatenated with no reconciliation (a provider-chain
                // anti-pattern — see
                // docs/specs/REPORT_PROCESS_ARCHITECTURE_STATE_AND_RETHINK_2026_07_22.md
                // §1/§3.0.2). `ProcessBroker::list_agent_panes()` sources
                // discovery from `blockcontroller::get_all_controllers()`
                // instead (authoritative for every controller type, closing
                // the coverage gap the old `process_tracker`-only half had
                // for `shell`/`acp` blocks) filtered down to agent panes —
                // `list()` alone would also include plain terminals, which
                // this RPC's contract has never included (reagent/codex P1).
                let block_ids: Vec<String> = process_broker
                    .list_agent_panes()
                    .into_iter()
                    .map(|status| status.block_id)
                    .collect();
                Ok(AgentTrackedBlocksResult { block_ids })
        }
    });
}

/// `swarm.other-instances`: this machine's other channels (the host-global shared
/// registry) and the LAN hosts (the LAN discovery peer list), with their agents'
/// names (Phases 1 and 2 of SPEC_SWARM_OTHER_HOSTS_AND_CHANNELS_2026_10_02.md)
/// and, where their instance reports it, their state (Phase 4). No content, no
/// credentials and no addresses.
fn register_swarm_other_instances(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let hostname = state.hostname.clone();
    let own_url = state.local_web_url.clone();
    let lan_discovery = state.lan_discovery.clone();
    let http = state.http_client.clone();
    // The response stays `Value`: the frontend models the snapshot with its own
    // types (swarm-remote.ts), which the stub keeps (generated-manifest.json `returns`).
    engine.register_typed(COMMAND_SWARM_OTHER_INSTANCES, move |_data: serde_json::Value, _ctx| {
        let hostname = hostname.clone();
        let own_url = own_url.clone();
        let lan_peers = lan_discovery.get_instances();
        let http = http.clone();
        async move {
                // A directory read; off the async runtime's worker threads.
                let entries = tokio::task::spawn_blocking(|| {
                    crate::registry::resolve_shared_reactive_dir()
                        .map(|dir| crate::backend::reactive::registry::list_all_shared(&dir))
                        .unwrap_or_default()
                })
                .await
                .unwrap_or_default();
                let own_addrs = crate::backend::lan_listeners::cached_local_addresses();
                let own_channel = crate::backend::reactive::registry::local_channel_id();
                let now_ms = agentmux_common::time::now_ms_u64();
                let targets = crate::backend::swarm_remote::channel_status_targets(&entries, &own_channel, &own_url, now_ms);
                let statuses = crate::backend::swarm_remote::channel_statuses(&http, &targets).await;
                let snapshot = crate::backend::swarm_remote::snapshot(
                    &entries,
                    &statuses,
                    &lan_peers,
                    &own_addrs,
                    &hostname,
                    &own_channel,
                    &own_url,
                    now_ms,
                );
                serde_json::to_value(&snapshot).map_err(|e| format!("swarm.other-instances: {e}"))
        }
    });
}

/// `ambient.outcomes`: how ambient calls have ended since srv started, so "the
/// swarm has been blank for an hour" is visible without a database copy
/// (SPEC_AMBIENT_SWARM_SUMMARY_HARDENING_2026_10_02.md section 5.8).
fn register_ambient_outcomes(engine: &Arc<WshRpcEngine>) {
    // `Value` response, as for swarm.other-instances: the frontend keeps its own type.
    engine.register_typed(COMMAND_AMBIENT_OUTCOMES, |_data: serde_json::Value, _ctx| async move {
        serde_json::to_value(crate::ambient::outcome::snapshot()).map_err(|e| format!("ambient.outcomes: {e}"))
    });
}

/// `agent.open-panes`: every agent pane in this instance, wherever it is.
/// A renderer only knows the panes in its own window, so "is this agent
/// open?" asked of a renderer misses one in a floating or other window —
/// My agents then offered a reattach for an agent that was live, and the
/// second pane could never run (SPEC_AGENT_SAME_PROCESS_DUPLICATE_PANE_
/// RECOVERY_2026_10_01.md §2 G0). Same source as `agent.tracked-blocks`.
fn register_agent_open_panes(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let process_broker = state.process_broker.clone();
    let mstore = state.mstore.clone();
    engine.register_typed(COMMAND_AGENT_OPEN_PANES, move |_data: serde_json::Value, _ctx| {
        let process_broker = process_broker.clone();
        let mstore = mstore.clone();
        async move {
                let block_ids: Vec<String> = process_broker
                    .list_agent_panes()
                    .into_iter()
                    .map(|status| status.block_id)
                    .collect();
                let panes = tokio::task::spawn_blocking(move || open_agent_panes(&mstore, &block_ids))
                    .await
                    .map_err(|e| format!("agent.open-panes: {e}"))?;
                Ok(AgentOpenPanesResult { panes })
        }
    });
}

/// Where each of `block_ids` is open: its agent, tab, and the windows
/// showing its workspace. The agent is resolved like the rest of srv
/// resolves a block's agent (`instance_get_active_for_block`: `agentId`,
/// the legacy `agent:id`, a template-launched row, the latest launch on the
/// block) so a pane from an older build is not reported closed (Codex P1 on
/// #4127); the raw meta id is the fallback when that finds no row. Blocks
/// with no resolvable agent or no parent tab are skipped. Workspaces and
/// windows are read once for the whole list.
pub(crate) fn open_agent_panes(store: &Store, block_ids: &[String]) -> Vec<AgentOpenPane> {
    let workspaces = store.get_all::<Workspace>().unwrap_or_default();
    let windows = store.get_all::<obj::Window>().unwrap_or_default();
    let mut panes: Vec<AgentOpenPane> = block_ids
        .iter()
        .filter_map(|block_id| {
            let block = store.get::<Block>(block_id).ok().flatten()?;
            let agent_id = match store.instance_get_active_for_block(block_id) {
                Ok(Some(row)) if !row.id.is_empty() => row.id,
                _ => {
                    let meta_id = obj::meta_get_string(&block.meta, "agentId", "");
                    if meta_id.is_empty() {
                        obj::meta_get_string(&block.meta, "agent:id", "")
                    } else {
                        meta_id
                    }
                }
            };
            if agent_id.is_empty() {
                return None;
            }
            let tab_id = block.parentoref.strip_prefix("tab:")?.to_string();
            let tab = store.get::<Tab>(&tab_id).ok().flatten()?;
            let mut window_ids: Vec<String> = workspaces
                .iter()
                .find(|w| w.tabids.contains(&tab_id) || w.pinnedtabids.contains(&tab_id))
                .map(|ws| {
                    windows
                        .iter()
                        .filter(|w| w.workspaceid == ws.oid)
                        .map(|w| w.oid.clone())
                        .collect()
                })
                .unwrap_or_default();
            window_ids.sort();
            Some(AgentOpenPane { block_id: block_id.clone(), agent_id, tab_id, tab_name: tab.name, window_ids })
        })
        .collect();
    panes.sort_by(|a, b| a.block_id.cmp(&b.block_id));
    panes
}

fn register_agent_kill_process(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let process_tracker = state.process_tracker.clone();
    engine.register_typed(COMMAND_AGENT_KILL_PROCESS, move |cmd: AgentKillProcessCommand, _ctx| {
        let process_tracker = process_tracker.clone();
        async move {
                tracing::info!(
                    block_id = %cmd.block_id,
                    pid = cmd.pid,
                    "agent.kill-process"
                );
                let ok = process_tracker.kill_pid(&cmd.block_id, cmd.pid);
                Ok(AgentKillResult { ok })
        }
    });
}

fn register_agent_kill_tree(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let process_tracker = state.process_tracker.clone();
    engine.register_typed(COMMAND_AGENT_KILL_TREE, move |cmd: AgentKillTreeCommand, _ctx| {
        let process_tracker = process_tracker.clone();
        async move {
                tracing::info!(block_id = %cmd.block_id, "agent.kill-tree");
                let ok = process_tracker.kill_tree(&cmd.block_id);
                Ok(AgentKillResult { ok })
        }
    });
}

/// The environment an `agent.send` spawn is given: `cmd:env` through
/// `build_persistent_spawn_env`, exactly as the `agentinput` path builds it
/// — identity gate, MuxBus token, wrapper auth, agent identity (UID + token
/// for a row-backed block, `AGENTMUX_AGENT_ID` from `agentName` when
/// `cmd:env` lacks it), git identity, tools PATH. Identity M4b-1 (spec
/// §6.5.8); factored out so the carriage is testable.
pub(crate) async fn agent_send_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: &Block,
    block_id: &str,
    auth_key: &str,
) -> Result<std::collections::HashMap<String, String>, crate::identity::resolver::SpawnGateError> {
    let env_vars = crate::backend::blockcontroller::cmd_env_of(&block.meta);
    crate::server::agent_handlers::input::build_persistent_spawn_env(
        mstore,
        id_store,
        identity_store,
        broker,
        &block.meta,
        block_id,
        auth_key,
        env_vars,
    )
    .await
}

fn register_agent_send(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let mstore = state.mstore.clone();
    let id_store = state.id_store.clone();
    let identity_store = state.identity_store.clone();
    let broker = state.broker.clone();
    let container_manager = state.container_manager.clone();
    let filestore = state.filestore.clone();
    // codex P1, PR #2802: needed to persist_last_failure/publish
    // EVENT_AGENT_FAILURE for a pre-spawn identity gate refusal — see the
    // same fix in agent_handlers/input.rs's agentinput handler.
    let event_bus = state.event_bus.clone();
    // Identity M4b-1: `build_persistent_spawn_env` re-injects the wrapper's
    // auth key, as `agentinput` does.
    let auth_key = state.auth_key.clone();

    engine.register_typed(COMMAND_AGENT_SEND, move |cmd: CommandAgentSendData, _ctx| {
        let mstore = mstore.clone();
        let id_store = id_store.clone();
        let identity_store = identity_store.clone();
        let broker = broker.clone();
        let container_manager = container_manager.clone();
        let filestore = filestore.clone();
        let event_bus = event_bus.clone();
        let auth_key = auth_key.clone();
        async move {
                tracing::info!(block_id = %cmd.block_id, "agent.send");

                let ctrl = blockcontroller::get_controller(&cmd.block_id)
                    .ok_or_else(|| format!("NOT_RUNNING: no controller for block {}", cmd.block_id))?;

                // Re-read spawn config from block metadata (same pattern as agentinput)
                let block: Block = mstore
                    .get(&cmd.block_id)
                    .map_err(|e| format!("agent.send: {e}"))?
                    .ok_or_else(|| format!("BLOCK_NOT_FOUND: {}", cmd.block_id))?;

                let cli_command = 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![],
                };
                // Same backstop as `agentinput`: this path has no frontend rebuild.
                let cli_args = crate::backend::agent_runtime::with_runtime_flags(&block.meta, cli_args);
                let working_dir = obj::meta_get_string(&block.meta, "cmd:cwd", "");
                // Identity M4b-1 (spec §6.5.8): the same env builder as
                // `agentinput`, so a row-backed block's process carries its
                // UID + token — and the wrapper auth, MuxBus, identity and
                // PATH vars `agentinput` gives — instead of a hand copy that
                // had only the identity gate. Same gate error, same arm below.
                let mut env_vars = match agent_send_spawn_env(
                    mstore.clone(),
                    id_store.clone(),
                    identity_store.clone(),
                    Some(broker.clone()),
                    &block,
                    &cmd.block_id,
                    &auth_key,
                )
                .await
                {
                    Ok(env) => env,
                    Err(gate) => {
                        let error_frame = serde_json::json!({
                            "type": "result",
                            "is_error": true,
                            "subtype": "error_during_execution",
                            "error": {"message": format!("[AgentMux] {gate}")}
                        })
                        .to_string();
                        // Some(filestore): 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,
                        // which is what a pane with an agentId reloads from
                        // (Phase 5a-3c).
                        let global_zone = crate::backend::blockcontroller::shell::resolve_global_output_zone(
                            &Some(mstore.clone()),
                            &cmd.block_id,
                        );
                        crate::backend::blockcontroller::shell::handle_append_block_file(
                            &broker,
                            &cmd.block_id,
                            crate::backend::blockcontroller::subprocess::SUBPROCESS_OUTPUT_SUBJECT,
                            format!("{error_frame}\n").as_bytes(),
                            Some(&filestore),
                            global_zone.as_deref(),
                        );
                        // codex P1, PR #2802: same fix as agent_handlers/
                        // input.rs's agentinput handler — the frame above
                        // only lands in the block's raw output log, which
                        // the recovery banner does NOT read from. Classify
                        // + persist + publish so the structured card
                        // (agent:last_failure meta + EVENT_AGENT_FAILURE)
                        // populates for this pre-spawn refusal too.
                        let gate_failure = crate::agents::failure::classify(
                            None,
                            None,
                            &gate.to_string(),
                            None,
                        );
                        crate::backend::blockcontroller::core::persist_last_failure(
                            &cmd.block_id,
                            Some(&gate_failure),
                            &Some(mstore.clone()),
                            &Some(event_bus.clone()),
                        );
                        broker.publish_event::<crate::backend::mps_events::AgentFailed>(&gate_failure, vec![format!("block:{}", cmd.block_id)], 1);
                        return Err(format!("identity spawn gate: {gate}"));
                    }
                };
                let session_id_field = obj::meta_get_string(
                    &block.meta, "agent:session_id_field", "session_id",
                );

                // Dispatch to persistent or subprocess controller
                let mut session_id = None;
                // App Server owns a persistent thread/turn process and has its own
                // typed protocol path — mirrors the websocket agentinput dispatch in
                // agent_handlers/input.rs, which was already updated for this
                // controller type. Without this branch, an app-server-backed agent
                // opened via agent.open could never receive a turn via agent.send
                // (ReAgent P1, PR #3215).
                if let Some(app_server_ctrl) = ctrl
                    .as_any()
                    .downcast_ref::<blockcontroller::app_server_controller::AppServerController>()
                {
                    app_server_ctrl.send_message(cmd.message)?;
                } 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 = 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 parity with the subprocess path: pass the resume
                    // flag + captured session id so a respawn (e.g. after a
                    // /model change) continues the same conversation.
                    let resume_flag = obj::meta_get_string(
                        &block.meta, "agent:resume_flag", "--resume",
                    );
                    let persisted_session_id = 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: None,
                    };
                    persistent_ctrl.send_message(cmd.message, config)?;
                    session_id = persistent_ctrl.session_id();
                } else if let Some(subprocess_ctrl) = ctrl
                    .as_any()
                    .downcast_ref::<blockcontroller::subprocess::SubprocessController>()
                {
                    let resume_flag = obj::meta_get_string(
                        &block.meta, "agent:resume_flag", "--resume",
                    );
                    let resume_strategy = 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"
                        },
                    );
                    let prompt_arg_flag =
                        obj::meta_get_string(&block.meta, "agent:prompt_arg_flag", "");
                    // Picker reattach (parallel of the websocket-path
                    // logic): hydrate the persisted session id from
                    // block meta so spawn_turn appends --resume <sid>
                    // on the FIRST turn after reattach.
                    let persisted_session_id = 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 = obj::meta_get_string(&block.meta, "agentMode", "host");
                    // Cross-process session-lease key (registry::LeaseStore) —
                    // read once for both branches below. Only the host
                    // branch's spawn_turn enforces it in this PR; the
                    // container branch's field is unused for now.
                    let instance_id = 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 = 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 = obj::meta_get_string(&block.meta, "agentId", "");
                        let container_name = crate::backend::container::container_name_for_slug(&agent_id);
                        let volumes_json = obj::meta_get_string(&block.meta, "agent:container_volumes", "[]");
                        let volumes: Vec<String> = serde_json::from_str(&volumes_json).unwrap_or_default();

                        // Same credential/workspace mounts the `agentinput` path
                        // builds — see its comment. `CLAUDE_CONFIG_DIR` is a host
                        // path stripped by CONTAINER_ENV_DENYLIST, so mounting it
                        // is the only way the in-container CLI sees credentials.
                        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()),
                        };

                        // Same AGENTMUX_LOCAL_URL rewrite the `agentinput` path
                        // does — see its comment (input.rs). Without this a
                        // container agent opened via MuxAPI path gets no
                        // sidecar URL at all, same as the RPC path before this fix.
                        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),
                            );
                        }

                        let container_command = obj::meta_get_string(
                            &block.meta, "agent:container_command", "claude",
                        );
                        let agent_provider = obj::meta_get_string(&block.meta, "agentProvider", "claude");

                        // Start the container and make sure the provider CLI is
                        // installed in it; a failure is written to the pane first.
                        crate::server::agent_handlers::container_turn::prepare(
                            crate::server::agent_handlers::container_turn::ContainerTurnPrep {
                                cm: &cm,
                                block_id: &cmd.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,
                                mstore: &mstore,
                            },
                        )
                        .await?;


                        tracing::info!(
                            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]. The command
                        // is the provider CLI 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 (docker exec would fail "no such file or
                        // directory"). cli_args are format flags + provider flags —
                        // no host paths, safe as-is. spawn_container_turn appends
                        // --resume <sid> internally.
                        let mut base_cmd = vec![container_command];
                        base_cmd.extend(cli_args);

                        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: cmd.message,
                            resume_flag,
                            resume_strategy,
                            prompt_arg_flag,
                            session_id_field,
                            message_id: None,
                            session_id: if persisted_session_id.is_empty() {
                                None
                            } else {
                                Some(persisted_session_id)
                            },
                            instance_id: instance_id.clone(),
                        };
                        subprocess_ctrl.spawn_container_turn(cm.clone(), container_name, base_cmd, config)?;
                    } 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: cmd.message,
                            resume_flag,
                            resume_strategy,
                            prompt_arg_flag,
                            session_id_field,
                            message_id: None,
                            session_id: if persisted_session_id.is_empty() {
                                None
                            } else {
                                Some(persisted_session_id)
                            },
                            instance_id,
                        };
                        subprocess_ctrl.spawn_turn(config)?;
                    }
                } else if let Some(acp_ctrl) = ctrl
                    .as_any()
                    .downcast_ref::<blockcontroller::acp::AcpController>()
                {
                    // If the agent exited, start it again; the message waits
                    // for the new session.
                    acp_ctrl.ensure_started(block.meta.clone())?;
                    acp_ctrl.send_message(cmd.message, None)?;
                } else {
                    return Err("NOT_RUNNING: controller type not supported".to_string());
                }

                Ok(AgentSendResult {
                    block_id: cmd.block_id,
                    status: "running".to_string(),
                    session_id,
                })
        }
    });
}

/// Core `agent.stop` logic, factored out so `fleet.bulk-stop`
/// (`server/app_api/fleet.rs`) can loop it per target instead of
/// duplicating the controller-lookup/stop/status-read sequence.
pub(crate) async fn stop_one_agent_block(
    state: &AppState,
    block_id: &str,
    signal: Option<&str>,
    stop_background: Option<bool>,
) -> Result<AgentStopResult, String> {
    tracing::info!(block_id = %block_id, signal = ?signal, ?stop_background, "agent.stop");

    let force = matches!(signal, Some("SIGKILL") | Some("SIGTERM"));
    crate::sagas::agent_teardown::stop(state, block_id, !force, stop_background).await?;

    let exit_code = blockcontroller::get_block_controller_status(block_id)
        .map(|s| s.shellprocexitcode);

    Ok(AgentStopResult {
        block_id: block_id.to_string(),
        status: "done".to_string(),
        exit_code,
    })
}

fn register_agent_stop(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let state = state.clone();
    engine.register_typed(COMMAND_AGENT_STOP_API, move |cmd: CommandAgentStopApiData, _ctx| {
        let state = state.clone();
        async move {
                stop_one_agent_block(&state, &cmd.block_id, cmd.signal.as_deref(), cmd.stop_background).await
        }
    });
}

fn register_agent_status(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let mstore = state.mstore.clone();

    engine.register_typed(COMMAND_AGENT_STATUS, move |cmd: CommandAgentStatusData, _ctx| {
        let mstore = mstore.clone();
        async move {
                let block: Block = mstore
                    .get(&cmd.block_id)
                    .map_err(|e| format!("agent.status: {e}"))?
                    .ok_or_else(|| format!("BLOCK_NOT_FOUND: {}", cmd.block_id))?;

                let agent_id = obj::meta_get_string(&block.meta, "agentId", "");
                let provider = obj::meta_get_string(&block.meta, "agentProvider", "");
                let controller_type = obj::meta_get_string(&block.meta, "controller", "");

                let runtime = blockcontroller::get_block_controller_status(&cmd.block_id);
                let status = runtime.as_ref()
                    .map(|s| s.shellprocstatus.clone())
                    .unwrap_or_else(|| "init".to_string());
                let exit_code = runtime.as_ref()
                    .and_then(|s| if s.shellprocstatus == "done" { Some(s.shellprocexitcode) } else { None });
                let pid = None; // PID not currently exposed in status struct

                // Get session ID from block meta
                let session_id = block.meta.get("agent:sessionid")
                    .and_then(|v| v.as_str())
                    .map(|s| s.to_string());

                Ok(AgentStatusResult {
                    block_id: cmd.block_id,
                    agent_id,
                    provider,
                    controller_type,
                    status,
                    session_id,
                    pid,
                    exit_code,
                })
        }
    });
}

fn register_agent_list(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let mstore = state.mstore.clone();

    // No argument: `Value` takes whatever the caller sends.
    engine.register_typed(COMMAND_AGENT_LIST, move |_data: serde_json::Value, _ctx| {
        let mstore = mstore.clone();
        async move {
                let tabs: Vec<Tab> = mstore.get_all::<Tab>()
                    .map_err(|e| format!("agent.list: {e}"))?;

                let mut agents = Vec::new();
                for tab in &tabs {
                    for block_id in &tab.blockids {
                        if let Ok(Some(block)) = mstore.get::<Block>(block_id) {
                            let agent_id = obj::meta_get_string(&block.meta, "agentId", "");
                            if agent_id.is_empty() {
                                continue;
                            }
                            let provider = obj::meta_get_string(&block.meta, "agentProvider", "");
                            let status = blockcontroller::get_block_controller_status(block_id)
                                .map(|s| s.shellprocstatus)
                                .unwrap_or_else(|| "init".to_string());
                            let session_id = block.meta.get("agent:sessionid")
                                .and_then(|v| v.as_str())
                                .map(|s| s.to_string());

                            agents.push(AgentListEntry {
                                block_id: block_id.clone(),
                                tab_id: tab.oid.clone(),
                                agent_id,
                                provider,
                                status,
                                session_id,
                            });
                        }
                    }
                }

                Ok(AgentListResult { agents })
        }
    });
}

fn register_agent_output(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let broker = state.broker.clone();

    engine.register_typed(COMMAND_AGENT_OUTPUT, move |cmd: CommandAgentOutputData, _ctx| {
        let broker = broker.clone();
        async move {
                let scope = format!("block:{}", cmd.block_id);
                let max = cmd.max_lines.unwrap_or(1000);
                let after = cmd.after_line.unwrap_or(0);

                // Read persisted blockfile events from broker history
                let mut all_lines: Vec<String> = Vec::new();
                {
                    let events = broker.read_event_history(
                        crate::backend::mps::EVENT_BLOCK_FILE,
                        &scope,
                        max + after, // read enough to cover offset
                    );
                    for event in events {
                        if let Some(ref data) = event.data {
                            if let Some(data64) = data.get("data64").and_then(|v| v.as_str()) {
                                if let Ok(bytes) = base64::engine::general_purpose::STANDARD
                                    .decode(data64)
                                {
                                    let text = String::from_utf8_lossy(&bytes);
                                    for line in text.lines() {
                                        if !line.trim().is_empty() {
                                            all_lines.push(line.to_string());
                                        }
                                    }
                                }
                            }
                        }
                    }
                }

                let total = all_lines.len();
                let lines: Vec<String> = all_lines.into_iter()
                    .skip(after)
                    .take(max)
                    .collect();
                let has_more = after + lines.len() < total;

                Ok(AgentOutputResult {
                    block_id: cmd.block_id,
                    lines,
                    total_lines: total,
                    has_more,
                })
        }
    });
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::backend::storage::agents::test_agent_def;
    use crate::backend::storage::store::Store;

    fn block_with(store: &Store, block_id: &str, agent_id: &str) -> Block {
        let mut block = Block {
            oid: block_id.to_string(),
            parentoref: String::new(),
            version: 0,
            runtimeopts: None,
            stickers: None,
            meta: {
                let mut m = obj::MetaMapType::new();
                m.insert("view".to_string(), serde_json::json!("agent"));
                m.insert("agentId".to_string(), serde_json::json!(agent_id));
                m.insert("agentName".to_string(), serde_json::json!("AgentY"));
                m.insert("cmd:env".to_string(), serde_json::json!({"KEEP_ME": "1"}));
                m
            },
            subblockids: None,
        };
        store.insert(&mut block).unwrap();
        block
    }

    async fn env_for(
        store: &Arc<Store>,
        block: &Block,
    ) -> std::collections::HashMap<String, String> {
        agent_send_spawn_env(
            store.clone(),
            store.clone(),
            store.clone(),
            None,
            block,
            &block.oid,
            "test-key",
        )
        .await
        .expect("no spawn gate for a provider without credentials")
    }

    /// Identity M4b-1 (spec §6.5.8): an `agent.send` spawn on a row-backed
    /// block carries that row's UID and its own token — the env
    /// `agentinput` builds — plus the block's `cmd:env` and the wrapper
    /// auth key. Before M4b-1 it got `cmd:env` + identity injection only.
    #[tokio::test]
    async fn an_agent_send_spawn_carries_the_rows_uid_and_token() {
        let store = Arc::new(Store::open_in_memory().unwrap());
        store.apply_identity_schema_for_tests().unwrap();
        // A provider with no credential class, so the spawn gate (which
        // refuses an OAuth-class agent with no bound account) stays out of
        // a test about identity carriage.
        let mut def = test_agent_def(
            "uid-send-y",
            "AgentY",
            "test-no-credentials",
            "agent",
            1,
            "",
        );
        def.slug = "agenty".to_string();
        store.agent_def_insert(&mut def).unwrap();
        let block = block_with(&store, "block-send-y", "uid-send-y");

        let env = env_for(&store, &block).await;
        assert_eq!(
            env.get("AGENTMUX_AGENT_UID").map(String::as_str),
            Some("uid-send-y")
        );
        let token = store
            .agent_token_load("uid-send-y")
            .unwrap()
            .expect("minted");
        assert_eq!(env.get("AGENTMUX_AGENT_TOKEN"), Some(&token));
        assert_eq!(
            env.get("KEEP_ME").map(String::as_str),
            Some("1"),
            "cmd:env kept"
        );
        assert_eq!(
            env.get("AGENTMUX_AUTH_KEY").map(String::as_str),
            Some("test-key")
        );
    }

    /// Plain-`gh` guard: a spawn built through `build_persistent_spawn_env` —
    /// the path `agentinput`, eager resume and `agent.send` all share — points
    /// `GH_CONFIG_DIR` at the agent's login-free dir, overriding a value the
    /// block's `cmd:env` carries (e.g. a human's real gh config).
    #[tokio::test]
    async fn an_agent_send_spawn_overrides_gh_config_dir_with_the_guard_dir() {
        let store = Arc::new(Store::open_in_memory().unwrap());
        store.apply_identity_schema_for_tests().unwrap();
        let mut def = test_agent_def("uid-gh-y", "AgentY", "test-no-credentials", "agent", 1, "");
        def.slug = "agenty".to_string();
        store.agent_def_insert(&mut def).unwrap();
        let mut block = block_with(&store, "block-gh-y", "uid-gh-y");
        block.meta.insert(
            "cmd:env".to_string(),
            serde_json::json!({"GH_CONFIG_DIR": "/home/human/.config/gh"}),
        );
        store.update(&mut block).unwrap();

        let env = env_for(&store, &block).await;
        let dir = std::path::PathBuf::from(env.get("GH_CONFIG_DIR").expect("always set"));
        assert_eq!(dir.parent(), Some(crate::backend::gh_guard::guard_config_home().as_path()));
        assert!(
            dir.file_name().unwrap().to_string_lossy().starts_with("gh-"),
            "{}",
            dir.display()
        );
        assert_ne!(dir, std::path::PathBuf::from("/home/human/.config/gh"));
    }

    /// A block with no row (a provider-key pane) gets neither.
    #[tokio::test]
    async fn an_agent_send_spawn_without_a_row_carries_no_uid_or_token() {
        let store = Arc::new(Store::open_in_memory().unwrap());
        store.apply_identity_schema_for_tests().unwrap();
        let block = block_with(&store, "block-send-none", "claude");
        let env = env_for(&store, &block).await;
        assert!(!env.contains_key("AGENTMUX_AGENT_UID"));
        assert!(!env.contains_key("AGENTMUX_AGENT_TOKEN"));
    }
}

#[cfg(test)]
mod open_panes_tests {
    use super::*;
    use crate::backend::storage::store::Store;

    fn insert_tab(store: &Store, name: &str) -> Tab {
        let mut tab = Tab { oid: uuid::Uuid::new_v4().to_string(), name: name.to_string(), ..Default::default() };
        store.insert(&mut tab).unwrap();
        tab
    }

    fn insert_workspace(store: &Store, tab_ids: &[&str]) -> Workspace {
        let mut ws = Workspace {
            oid: uuid::Uuid::new_v4().to_string(),
            tabids: tab_ids.iter().map(|t| t.to_string()).collect(),
            ..Default::default()
        };
        store.insert(&mut ws).unwrap();
        ws
    }

    fn insert_window(store: &Store, workspace_id: &str) -> obj::Window {
        let mut w = obj::Window { oid: uuid::Uuid::new_v4().to_string(), workspaceid: workspace_id.to_string(), ..Default::default() };
        store.insert(&mut w).unwrap();
        w
    }

    fn insert_block(store: &Store, tab_id: &str, agent_id: Option<&str>) -> String {
        let mut meta = obj::MetaMapType::new();
        if let Some(a) = agent_id {
            meta.insert("agentId".to_string(), serde_json::json!(a));
        }
        let mut block = Block {
            oid: uuid::Uuid::new_v4().to_string(),
            parentoref: format!("tab:{tab_id}"),
            meta,
            ..Default::default()
        };
        store.insert(&mut block).unwrap();
        block.oid
    }

    /// The 2026-10-01 case: Korp torn off into a floating window (its own
    /// workspace and `Window`), a second agent in the main window. Both are
    /// reported, each with the window that shows it.
    #[test]
    fn lists_agent_panes_in_every_window_including_a_floating_one() {
        let store = Store::open_in_memory().unwrap();
        let main_tab = insert_tab(&store, "Tab 2");
        let main_ws = insert_workspace(&store, &[&main_tab.oid]);
        let main_win = insert_window(&store, &main_ws.oid);
        let float_tab = insert_tab(&store, "");
        let float_ws = insert_workspace(&store, &[&float_tab.oid]);
        let float_win = insert_window(&store, &float_ws.oid);

        let agent3 = insert_block(&store, &main_tab.oid, Some("agent3-def"));
        let korp = insert_block(&store, &float_tab.oid, Some("korp-def"));

        let panes = open_agent_panes(&store, &[agent3.clone(), korp.clone()]);
        let korp_pane = panes.iter().find(|p| p.block_id == korp).expect("floating pane listed");
        assert_eq!(korp_pane.agent_id, "korp-def");
        assert_eq!(korp_pane.tab_id, float_tab.oid);
        assert_eq!(korp_pane.window_ids, vec![float_win.oid]);
        let a3 = panes.iter().find(|p| p.block_id == agent3).expect("main pane listed");
        assert_eq!(a3.tab_name, "Tab 2");
        assert_eq!(a3.window_ids, vec![main_win.oid]);
    }

    /// A pane from an older build carries only the legacy `agent:id` key.
    #[test]
    fn resolves_a_legacy_agent_id_pane_to_its_agent_row() {
        let store = Store::open_in_memory().unwrap();
        let mut def = crate::backend::storage::agents::test_agent_def(
            "uid-legacy",
            "Korp",
            "test-no-credentials",
            "agent",
            1,
            "",
        );
        store.agent_def_insert(&mut def).unwrap();
        let tab = insert_tab(&store, "t");
        insert_workspace(&store, &[&tab.oid]);
        let mut meta = obj::MetaMapType::new();
        meta.insert("agent:id".to_string(), serde_json::json!("uid-legacy"));
        let mut block = Block {
            oid: uuid::Uuid::new_v4().to_string(),
            parentoref: format!("tab:{}", tab.oid),
            meta,
            ..Default::default()
        };
        store.insert(&mut block).unwrap();

        let panes = open_agent_panes(&store, &[block.oid.clone()]);
        assert_eq!(panes.len(), 1);
        assert_eq!(panes[0].agent_id, "uid-legacy");
    }

    #[test]
    fn skips_non_agent_and_orphan_blocks_and_reports_a_windowless_workspace() {
        let store = Store::open_in_memory().unwrap();
        let tab = insert_tab(&store, "t");
        insert_workspace(&store, &[&tab.oid]); // open in no window
        let terminal = insert_block(&store, &tab.oid, None);
        let agent = insert_block(&store, &tab.oid, Some("def"));
        let missing = "no-such-block".to_string();

        let panes = open_agent_panes(&store, &[terminal, agent.clone(), missing]);
        assert_eq!(panes.len(), 1);
        assert_eq!(panes[0].block_id, agent);
        assert!(panes[0].window_ids.is_empty());
    }
}