agentmux_srv/server/

websocket.rs

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

use std::sync::Arc;

use axum::{
    extract::{
        ws::{Message, WebSocket},
        State, WebSocketUpgrade,
    },
    response::Response,
};
use base64::Engine as _;
use serde::Deserialize;
use serde_json::json;

use crate::backend::background_task_feed::publish_background_task_updated;
use crate::backend::blockcontroller;
use crate::backend::rpc::engine::WshRpcEngine;
use crate::backend::rpc_types::{
    AiRateLimitResult, AppInfoResult, NoArgsReq,
    CommandBlockInputData, CommandControllerResyncData, CommandCreateSubBlockData,
    CommandDeleteSubBlockData, CommandEventReadHistoryData,
    CommandGetMetaData, CommandSetMetaData, CommandToolDecisionData,
    RpcMessage, COMMAND_CONTROLLER_INPUT,
    COMMAND_CONTROLLER_RESYNC, COMMAND_CREATE_SUB_BLOCK, COMMAND_DELETE_SUB_BLOCK,
    COMMAND_EVENT_READ_HISTORY, COMMAND_EVENT_SUB, COMMAND_EVENT_UNSUB,
    COMMAND_EVENT_UNSUB_ALL, COMMAND_GET_FULL_CONFIG, COMMAND_GET_META,
    COMMAND_GET_AI_RATE_LIMIT, COMMAND_ROUTE_ANNOUNCE, COMMAND_ROUTE_UNANNOUNCE,
    COMMAND_SET_META, COMMAND_SET_CONFIG, COMMAND_APP_INFO,
    COMMAND_TOOL_DECISION, COMMAND_AGENT_ANSWER, COMMAND_AGENT_CANCEL,
    COMMAND_AGENT_SHUTDOWN_KEEP, CommandAgentShutdownKeepData, AgentShutdownKeepResult,
    CommandAgentAnswerData, CommandAgentCancelData,
    COMMAND_AMBIENT_NARRATE, CommandAmbientNarrateData,
    COMMAND_DOCK_NODE_STATUS, CommandDockNodeStatusData,
    COMMAND_BACKGROUND_TASK_COMPLETION, CommandBackgroundTaskCompletionData,
    COMMAND_BACKGROUND_TASK_PID, CommandBackgroundTaskPidData,
    COMMAND_LIST_BACKGROUND_TASKS, CommandListBackgroundTasksData,
};
use crate::backend::base::normalize_working_dir;
use crate::backend::obj::{Block, TermSize, MuxObjUpdate, mux_obj_to_value};
use super::service::{update_object_meta, schedule_agent_zoom_mirror};

use super::AppState;

/// Incoming WebSocket message envelope.
/// Supports both ping/pong messages and wscommand-based RPC.
#[derive(Deserialize)]
struct WSIncoming {
    #[serde(rename = "type")]
    msg_type: Option<String>,
    #[allow(dead_code)]
    stime: Option<i64>,
    wscommand: Option<String>,
    message: Option<RpcMessage>,
    // Fields for setblocktermsize / blockinput
    blockid: Option<String>,
    inputdata64: Option<String>,
    termsize: Option<serde_json::Value>,
    // Fields for bus:* commands
    agent_id: Option<String>,
    from: Option<String>,
    to: Option<String>,
    target: Option<String>,
    payload: Option<String>,
    #[serde(rename = "bus_message")]
    bus_message_text: Option<String>,
    priority: Option<String>,
}

/// The `srvinfo` event sent once on every WebSocket connect, right after
/// `config`: srv's version, which the UI compares with its own, and the
/// machine srv runs on (`crate::srv_info`).
fn srv_info_event(info: crate::srv_info::SrvInfo) -> serde_json::Value {
    crate::backend::eventbus::EventBus::typed_frame::<crate::backend::mps_events::SrvInfo>(&info)
}

pub(super) async fn handle_ws(
    State(state): State<AppState>,
    ws: WebSocketUpgrade,
) -> Response {
    ws.on_upgrade(move |socket| handle_ws_connection(socket, state))
}

async fn handle_ws_connection(mut socket: WebSocket, state: AppState) {
    let ws_start = std::time::Instant::now();
    let conn_id = uuid::Uuid::new_v4().to_string();
    let tab_id = String::new();

    tracing::info!(conn_id = %conn_id, "WebSocket client connected");

    // Two egress lanes per connection: interactive (terminal echo, RPC-routed
    // wave events, obj updates) and background (droppable perf telemetry —
    // sysinfo/blockstats). The select! below drains priority before background
    // so typing never waits behind a perf tick. See
    // docs/specs/SPEC_TERMINAL_INPUT_PRIORITY_OVER_SYSINFO_2026_06_16.md.
    let crate::backend::eventbus::WsReceivers {
        priority: mut priority_rx,
        background: mut background_rx,
    } = state.event_bus.register_ws(&conn_id, &tab_id);
    tracing::info!("[ws-perf] register_ws: {:.2}ms", ws_start.elapsed().as_secs_f64() * 1000.0);

    // Optional messagebus receiver — activated when pane sends bus:register
    let mut bus_rx: Option<tokio::sync::mpsc::UnboundedReceiver<crate::backend::messagebus::BusMessage>> = None;
    let mut bus_agent_id: Option<String> = None;

    // Send initial "config" wave event via the RPC eventrecv path so the frontend
    // populates fullConfigAtom (and shows the widget bar).
    // Frontend only processes events via: {"eventtype":"rpc","data":{"command":"eventrecv","data":{"event":"config","data":{...}}}}
    {
        let t = std::time::Instant::now();
        let config = state.config_watcher.get_full_config();
        if let Ok(mut config_val) = serde_json::to_value(config.as_ref()) {
            crate::backend::wconfig::redact_full_config_for_renderer(&mut config_val);
            let config_event = crate::backend::eventbus::EventBus::typed_frame::<crate::backend::mps_events::Config>(&crate::backend::mps_events::ConfigEvent { fullconfig: config_val });
            if let Ok(msg) = serde_json::to_string(&config_event) {
                let _ = socket.send(Message::Text(msg.into())).await;
            }
        }
        tracing::info!("[ws-perf] send_initial_config: {:.2}ms", t.elapsed().as_secs_f64() * 1000.0);
    }

    // Tell the client which srv it reached: the version, so a UI from another
    // version (an update replaced one side) can ask the user to reload, and
    // the machine, whose names and paths the UI shows and builds.
    let info = crate::srv_info::srv_info(&state.version, &state.hostname);
    if let Ok(msg) = serde_json::to_string(&srv_info_event(info)) {
        let _ = socket.send(Message::Text(msg.into())).await;
    }

    // Create RPC engine for this connection
    let t = std::time::Instant::now();
    let (engine, mut rpc_output_rx) = WshRpcEngine::new();

    // Register handlers
    register_handlers(&engine, state.clone(), conn_id.clone());
    tracing::info!("[ws-perf] create_engine+register_handlers: {:.2}ms", t.elapsed().as_secs_f64() * 1000.0);
    tracing::info!("[ws-perf] TOTAL ws_setup: {:.2}ms", ws_start.elapsed().as_secs_f64() * 1000.0);

    // Periodic ping interval (10 seconds)
    let mut ping_interval = tokio::time::interval(std::time::Duration::from_secs(10));
    ping_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);

    // Fairly-ordered priority-lane events awaiting their turn to be sent, one
    // per `'ws` loop iteration (see the priority-lane branch below and the
    // pending-priority-flush branch after it). Reagent P1 on this PR's first
    // revision: forwarding an entire drained batch inside one `select!`
    // branch body — even a fairly-ordered one — starves `socket.recv()`
    // (this connection's OWN incoming keystrokes) for the whole batch, since
    // `select!` only re-polls sibling branches after the chosen branch's body
    // returns. Trickling the batch out one event per iteration keeps every
    // send interleaved with a fresh `socket.recv()` poll, exactly like the
    // pre-fix one-event-per-iteration cadence, while still applying the fair
    // pane ordering to however the batch is drained.
    let mut pending_priority: std::collections::VecDeque<serde_json::Value> =
        std::collections::VecDeque::new();

    'ws: loop {
        tokio::select! {
            // Biased: poll branches top-to-bottom so interactive terminal I/O
            // always wins over droppable perf telemetry. Order = incoming
            // keystrokes → RPC replies → agent bus → priority events (terminal
            // echo) → background events (sysinfo/blockstats) → keepalive ping.
            // See docs/specs/SPEC_TERMINAL_INPUT_PRIORITY_OVER_SYSINFO_2026_06_16.md.
            biased;

            // Incoming WebSocket messages → parse & dispatch (keystrokes first)
            msg = socket.recv() => {
                match msg {
                    Some(Ok(Message::Close(_))) | None => break,
                    Some(Ok(Message::Ping(data))) => {
                        let _ = socket.send(Message::Pong(data)).await;
                    }
                    Some(Ok(Message::Text(text))) => {
                        match handle_incoming_text(&text, &engine, &state, &mut socket).await {
                            Err(true) => break,
                            Ok(Some((new_rx, agent_id))) => {
                                // bus:register returned a new receiver
                                bus_rx = Some(new_rx);
                                bus_agent_id = Some(agent_id);
                            }
                            _ => {}
                        }
                    }
                    Some(Ok(_)) => {
                        // Binary or other message types — ignore
                    }
                    Some(Err(_)) => break,
                }
            }

            // Forward RPC engine output → WebSocket (wrapped as eventtype:rpc)
            Some(rpc_msg) = rpc_output_rx.recv() => {
                let wrapped = json!({
                    "eventtype": "rpc",
                    "data": rpc_msg,
                });
                let msg = serde_json::to_string(&wrapped).unwrap_or_default();
                if socket.send(Message::Text(msg.into())).await.is_err() {
                    break;
                }
            }

            // Forward MessageBus messages → WebSocket (if registered as agent)
            Some(bus_msg) = async {
                match bus_rx.as_mut() {
                    Some(rx) => rx.recv().await,
                    None => std::future::pending().await,
                }
            } => {
                let wrapped = json!({
                    "type": "bus:message",
                    "data": bus_msg,
                });
                let msg = serde_json::to_string(&wrapped).unwrap_or_default();
                if socket.send(Message::Text(msg.into())).await.is_err() {
                    break;
                }
            }

            // Priority event lane → WebSocket. Two sources feed it:
            //   1. MPS Broker (via EventBusBridge) — already wrapped as
            //      { eventtype: "rpc", data: { command: "eventrecv", data: MuxEvent } }
            //   2. Direct broadcasts (e.g., SetMeta's obj:update) — raw
            //      { eventtype: "waveobj:update", oref: "block:xxx", data: ... }
            // This carries terminal echo output and all interactive events.
            //
            // Fairness: this lane is shared by EVERY pane on the connection, not
            // just one. A single FIFO here means a noisy pane (heavy PTY output)
            // can queue an unbounded number of frames ahead of a different pane's
            // own keystroke echo, which arrives in the same channel. Draining all
            // immediately-available events and round-robining them by pane (see
            // `fair_drain_priority`) bounds that delay to one "round" per distinct
            // active pane instead of the length of the noisy pane's backlog. See
            // docs/analysis/ANALYSIS_CROSS_PANE_INPUT_DELAY_UNDER_OUTPUT_LOAD_2026_09_04.md.
            //
            // Only drains a fresh batch when `pending_priority` is empty — while
            // it's still being flushed (one per iteration, below), this branch
            // stays disabled so we don't pull more out of `priority_rx` than
            // we've finished sending.
            Some(event) = priority_rx.recv(), if pending_priority.is_empty() => {
                let mut fair = fair_drain_priority(event, &mut priority_rx);
                // `fair` always has >= 1 element (it contains at least `event`).
                let first = fair.remove(0);
                pending_priority.extend(fair);
                if forward_event(&mut socket, first).await {
                    break 'ws;
                }
            }

            // Flush one fairly-ordered priority event per iteration. A no-op
            // branch guarded on `pending_priority` being non-empty — see the
            // comment on `pending_priority`'s declaration above for why this
            // is its own branch instead of a loop in the arm above.
            _ = std::future::ready(()), if !pending_priority.is_empty() => {
                let ev = pending_priority.pop_front().expect("guarded non-empty");
                if forward_event(&mut socket, ev).await {
                    break 'ws;
                }
            }

            // Background event lane → WebSocket. Droppable perf telemetry
            // (sysinfo/blockstats) only; serviced when the priority lanes above
            // are momentarily idle, so it can never delay a keystroke echo.
            //
            // Tradeoff of `biased;`: a SUSTAINED priority-lane flood (e.g.
            // `! yes` pumping terminal output continuously) keeps the priority
            // branches ready, so this lane is starved and events queue here
            // until the flood pauses, then flush as a stale burst. As of B.2
            // (SPEC_MEMORY_COMMIT_ATTRIBUTION_CORRECTION_2026_07_02) this is a
            // bounded `mpsc::channel(BACKGROUND_LANE_CAPACITY)` with
            // `try_send` drop-on-full semantics (see eventbus.rs), not an
            // unbounded receiver — a starvation period long enough to fill
            // all 256 slots drops each new event as it arrives (the queued
            // events already in the channel are unaffected) rather than
            // growing commit without limit. Acceptable for now because telemetry
            // ingress is low-rate (sysinfo ~1/s) and a flood also saturates the
            // socket writes — during which the priority lanes do drain, so this
            // lane gets serviced — making true never-empty starvation the rare
            // Phase 2: coalesce all immediately-available background events to the
            // latest reading per (event-name:scope) key before forwarding. During
            // sustained priority-lane activity, sysinfo/blockstats queue here; without
            // coalescing they flush as a burst of stale frames that the browser
            // processes back-to-back, delaying queued keypress events. Coalescing
            // bounds the flush to O(distinct event×scope pairs) — at most 1 sysinfo +
            // 1 blockstats per open block — regardless of how long the lane was starved.
            // See SPEC_TERMINAL_INPUT_PRIORITY_OVER_SYSINFO_2026_06_16.md §4.2.
            Some(first_bg) = background_rx.recv() => {
                let coalesced = coalesce_background(first_bg, &mut background_rx);
                for ev in coalesced {
                    if forward_event(&mut socket, ev).await {
                        break 'ws;
                    }
                }
            }

            // Periodic ping
            _ = ping_interval.tick() => {
                let now = agentmux_common::time::now_ms();
                let ping = json!({ "type": "ping", "stime": now });
                let msg = serde_json::to_string(&ping).unwrap_or_default();
                if socket.send(Message::Text(msg.into())).await.is_err() {
                    break;
                }
            }
        }
    }

    tracing::info!(conn_id = %conn_id, "WebSocket client disconnected");
    state.event_bus.unregister_ws(&conn_id);
    state.broker.unsubscribe_all(&conn_id);
    if let Some(r) = crate::backend::notify::router::get(&state.broker) {
        r.disconnect(&conn_id);
    }

    // Unregister from messagebus if this connection was an agent
    if let Some(ref agent_id) = bus_agent_id {
        state.messagebus.unregister(agent_id);
    }
}

/// Forward a queued event-bus value to the WebSocket. Returns `true` if the
/// send failed (the caller should break the loop).
///
/// Two shapes arrive: already-RPC-wrapped values (from the MPS broker via
/// EventBusBridge) are forwarded as-is; raw event-bus values (e.g. SetMeta's
/// `waveobj:update`) are wrapped as an RPC `eventrecv` so the frontend
/// WshRouter routes them to handleMuxEvent → updateMuxObject. Shared by both
/// the priority and background egress lanes.
async fn forward_event(socket: &mut WebSocket, event: serde_json::Value) -> bool {
    let msg = if event["eventtype"] == "rpc" {
        // Already an RPC message (from MPS broker via EventBusBridge)
        serde_json::to_string(&event).unwrap_or_default()
    } else {
        // Raw event bus event — wrap as RPC eventrecv
        // Re-wraps a raw event-bus event into the frame the renderer reads. It is typed where it
        // was built (`EventBus::typed_ws_event`); check-untyped-rpc-handlers counts any raw event
        // built by hand.
        let mux_event = json!({
            "event": event["eventtype"],
            "scopes": [event["oref"]],
            "data": event["data"],
        });
        let wrapped = json!({
            "eventtype": "rpc",
            "data": {
                // untyped-relay: forwards an event built elsewhere, as it was built.
                "command": "eventrecv",
                "data": mux_event,
            },
        });
        serde_json::to_string(&wrapped).unwrap_or_default()
    };
    socket.send(Message::Text(msg.into())).await.is_err()
}

/// Extract a coalescing key from a background-lane event.
/// Background events are RPC-wrapped MPS events:
///   { "eventtype": "rpc", "data": { "command": "eventrecv", "data": { "event": "sysinfo", "scopes": ["local"] } } }
/// Key = "<event-name>:<first-scope>", e.g. "sysinfo:local" or "blockstats:block:abc123".
/// Falls back to "_:_" for events that don't match the expected shape (safe: they coalesce
/// to one slot, which is acceptable because non-sysinfo/blockstats never reach this lane).
fn background_event_key(event: &serde_json::Value) -> String {
    let inner = event.get("data").and_then(|d| d.get("data"));
    let event_name = inner
        .and_then(|d| d.get("event"))
        .and_then(|e| e.as_str())
        .unwrap_or("_");
    let scope = inner
        .and_then(|d| d.get("scopes"))
        .and_then(|s| s.get(0))
        .and_then(|s| s.as_str())
        .unwrap_or("_");
    format!("{event_name}:{scope}")
}

/// Drain all immediately-available events from `rx` (including `first`) and
/// coalesce them to the latest reading per (event-name:scope) key. Iteration
/// order in the returned Vec is first-seen (stable across coalescing rounds).
fn coalesce_background(
    first: serde_json::Value,
    rx: &mut tokio::sync::mpsc::Receiver<serde_json::Value>,
) -> Vec<serde_json::Value> {
    let mut order: Vec<String> = Vec::new();
    let mut map: std::collections::HashMap<String, serde_json::Value> = std::collections::HashMap::new();
    let key = background_event_key(&first);
    order.push(key.clone());
    map.insert(key, first);
    while let Ok(next) = rx.try_recv() {
        let k = background_event_key(&next);
        if !map.contains_key(&k) {
            order.push(k.clone());
        }
        map.insert(k, next);
    }
    order.into_iter().filter_map(|k| map.remove(&k)).collect()
}

/// Sentinel key for every priority-lane event that is NOT a `blockfile`
/// (terminal/PTY output) event. Not a valid `block:<id>` scope string, so it
/// can never collide with a real pane key.
const NON_PANE_EVENT_KEY: &str = "\u{0}non-pane-event";

/// Extract a fairness key so a flood of PTY output from one pane can't queue
/// an unbounded number of frames ahead of a different pane's own keystroke
/// echo sitting in the same lane.
///
/// Reordering is scoped to `blockfile` events ONLY — that's the one event
/// type a single pane can flood the shared lane with, and PTY bytes from
/// different panes are causally independent (only same-pane ordering
/// matters, which each key's own FIFO group preserves). Every OTHER event
/// type on this lane (`waveobj:update`, `waveobj:batchedupdates`, the initial
/// `config` push, RPC replies) returns the same `NON_PANE_EVENT_KEY` and so
/// stays in ONE FIFO group, never reordered relative to each other.
///
/// This restriction is load-bearing, not a simplification: a `waveobj:update`
/// for object X and a LATER `waveobj:batchedupdates` that also touches X must
/// never be reordered relative to each other, but an unscoped batch has no
/// single object id to key on — an earlier version of this function keyed
/// updates by their individual `oref`/scope and left unscoped batches in
/// their own group, which let round-robin interleave `[update, delete]` into
/// `[update, delete, update]`-style reorderings; the frontend doesn't
/// version-guard deletes, so a resurrected-then-redeleted object could render
/// as resurrected until reload. See the review discussion on the PR that
/// introduced this function.
fn priority_pane_key(event: &serde_json::Value) -> Option<String> {
    if event["eventtype"] != "rpc" {
        // Raw (non-rpc) events on this lane are always object-model
        // broadcasts (e.g. SetMeta's `waveobj:update`), never blockfile —
        // see `forward_event`'s shape comment.
        return Some(NON_PANE_EVENT_KEY.to_string());
    }
    let inner = event.get("data").and_then(|d| d.get("data"));
    let is_blockfile = inner.and_then(|d| d.get("event")).and_then(|e| e.as_str())
        == Some(crate::backend::mps::EVENT_BLOCK_FILE);
    if !is_blockfile {
        return Some(NON_PANE_EVENT_KEY.to_string());
    }
    inner
        .and_then(|d| d.get("scopes"))
        .and_then(|s| s.get(0))
        .and_then(|s| s.as_str())
        .map(|s| s.to_string())
        .or_else(|| Some(NON_PANE_EVENT_KEY.to_string()))
}

fn enqueue_priority_event(
    order: &mut Vec<String>,
    groups: &mut std::collections::HashMap<String, std::collections::VecDeque<serde_json::Value>>,
    event: serde_json::Value,
) {
    let key = priority_pane_key(&event).unwrap_or_default();
    if !groups.contains_key(&key) {
        order.push(key.clone());
    }
    groups.entry(key).or_default().push_back(event);
}

/// Drain all immediately-available events from `rx` (including `first`) and
/// interleave them round-robin by pane (`priority_pane_key`), instead of
/// forwarding in raw arrival order. Without this, a pane producing output
/// faster than the socket can flush it fills this same lane with its own
/// frames, and a different pane's keystroke-echo frame — queued in the same
/// FIFO — waits behind the entire backlog. Round-robining bounds that wait to
/// one slot per OTHER distinct pane with events currently queued, regardless
/// of how deep any single pane's backlog is.
///
/// Ordering guarantee: events for the SAME pane are never reordered relative
/// to each other (each pane's VecDeque is FIFO); only the INTERLEAVING across
/// different panes changes.
fn fair_drain_priority(
    first: serde_json::Value,
    rx: &mut tokio::sync::mpsc::Receiver<serde_json::Value>,
) -> Vec<serde_json::Value> {
    let mut order: Vec<String> = Vec::new();
    let mut groups: std::collections::HashMap<String, std::collections::VecDeque<serde_json::Value>> =
        std::collections::HashMap::new();

    enqueue_priority_event(&mut order, &mut groups, first);
    while let Ok(next) = rx.try_recv() {
        enqueue_priority_event(&mut order, &mut groups, next);
    }

    // Common case — only one pane (or only one event) queued this round.
    // Skip the round-robin bookkeeping and preserve arrival order directly.
    if order.len() <= 1 {
        return order
            .into_iter()
            .filter_map(|k| groups.remove(&k))
            .flat_map(|q| q.into_iter())
            .collect();
    }

    let total: usize = groups.values().map(|q| q.len()).sum();
    let mut out = Vec::with_capacity(total);
    loop {
        let mut progressed = false;
        for key in &order {
            if let Some(q) = groups.get_mut(key) {
                if let Some(ev) = q.pop_front() {
                    out.push(ev);
                    progressed = true;
                }
            }
        }
        if !progressed {
            break;
        }
    }
    out
}

/// Handle an incoming text message.
/// Returns Err(true) if the socket send failed.
/// Returns Ok(Some((rx, agent_id))) if a bus:register was processed.
async fn handle_incoming_text(
    text: &str,
    engine: &Arc<WshRpcEngine>,
    state: &AppState,
    socket: &mut WebSocket,
) -> Result<Option<(tokio::sync::mpsc::UnboundedReceiver<crate::backend::messagebus::BusMessage>, String)>, bool> {
    let incoming: WSIncoming = match serde_json::from_str(text) {
        Ok(v) => v,
        Err(e) => {
            tracing::warn!("ws: invalid JSON: {}", e);
            return Ok(None);
        }
    };

    // Handle ping/pong by type field
    if let Some(ref msg_type) = incoming.msg_type {
        match msg_type.as_str() {
            "ping" => {
                let now = agentmux_common::time::now_ms();
                let pong = json!({ "type": "pong", "stime": now });
                let msg = serde_json::to_string(&pong).unwrap_or_default();
                if socket.send(Message::Text(msg.into())).await.is_err() {
                    return Err(true);
                }
                return Ok(None);
            }
            "pong" => {
                return Ok(None);
            }
            "bus:register" => {
                if let Some(ref agent_id) = incoming.agent_id {
                    let rx = state.messagebus.register(agent_id, "websocket");
                    // Stamp agent_id on the RPC context so MuxAPI S1 checks work.
                    engine.set_rpc_context(crate::backend::rpc_types::RpcContext {
                        agent_id: agent_id.clone(),
                        ..Default::default()
                    });
                    let ack = json!({ "type": "bus:registered", "agent_id": agent_id });
                    let msg = serde_json::to_string(&ack).unwrap_or_default();
                    if socket.send(Message::Text(msg.into())).await.is_err() {
                        return Err(true);
                    }
                    return Ok(Some((rx, agent_id.clone())));
                }
                return Ok(None);
            }
            "bus:send" => {
                if let (Some(ref from), Some(ref to), Some(ref payload)) =
                    (&incoming.from, &incoming.to, &incoming.payload)
                {
                    let priority = match incoming.priority.as_deref() {
                        Some("high") => crate::backend::messagebus::Priority::High,
                        Some("urgent") => crate::backend::messagebus::Priority::Urgent,
                        _ => crate::backend::messagebus::Priority::Normal,
                    };
                    let bus_msg = crate::backend::messagebus::BusMessage::new(
                        from, to, crate::backend::messagebus::MessageType::Send, payload, priority,
                    );
                    let msg_id = bus_msg.id.clone();
                    let _ = state.messagebus.send(bus_msg);
                    let ack = json!({ "type": "bus:sent", "message_id": msg_id });
                    let msg = serde_json::to_string(&ack).unwrap_or_default();
                    if socket.send(Message::Text(msg.into())).await.is_err() {
                        return Err(true);
                    }
                }
                return Ok(None);
            }
            "bus:inject" => {
                let from = incoming.from.as_deref().unwrap_or("unknown");
                if let (Some(ref target), Some(ref message)) =
                    (&incoming.target, &incoming.bus_message_text)
                {
                    // Try direct PTY injection via ReactiveHandler first
                    let mut reactive_req = crate::backend::reactive::InjectionRequest {
                        target_agent: target.clone(),
                        message: message.clone(),
                        source_agent: Some(from.to_string()),
                        request_id: None,
                        priority: incoming.priority.clone(),
                        jekt_tier: None,   // auto-detected from keywords
                        delivery_tier: Some("host".to_string()),
                        forward_hops: 0,
                        ..Default::default()
                    };
                    // reagentx P0 on PR #2565: same bypass as
                    // messagebus.rs::handle_inject — this WS message type
                    // built InjectionRequest from client-controlled
                    // `incoming.from` and called inject_message directly,
                    // never going through host-tier signature verification.
                    super::reactive::verify_jekt_signature(state, &mut reactive_req);
                    let resp = state.reactive_handler.inject_message(reactive_req);
                    if resp.success {
                        // Sender-side echo (SPEC_JEKT_SECURITY_AND_VISIBILITY §3.2)
                        super::reactive::echo_jekt_to_sender(
                            state,
                            Some(from),
                            target,
                            message,
                            &resp.request_id,
                            resp.effective_tier.as_deref(),
                            resp.requires_stop,
                            // Always host-tier here (this is the pane's own outbound
                            // send) — reagent/lan verification never applies.
                            super::reactive::EchoTrust {
                                delivery_tier: "host",
                                sig_verified: None,
                                reagent_verified: None,
                                lan_verified: None,
                                channel_verified: None,
                            },
                            incoming.priority.as_deref().unwrap_or("normal"),
                        );
                        let ack = json!({ "type": "bus:injected", "via": "pty", "block_id": resp.block_id });
                        let msg = serde_json::to_string(&ack).unwrap_or_default();
                        if socket.send(Message::Text(msg.into())).await.is_err() {
                            return Err(true);
                        }
                        return Ok(None);
                    }

                    // Non-"agent not found" error — report it
                    let is_not_found = resp.error.as_deref().map(|e| e.contains("not found")).unwrap_or(false);
                    if !is_not_found {
                        let err = json!({ "type": "bus:error", "error": resp.error });
                        let msg = serde_json::to_string(&err).unwrap_or_default();
                        if socket.send(Message::Text(msg.into())).await.is_err() {
                            return Err(true);
                        }
                        return Ok(None);
                    }

                    // Fall back to MessageBus WebSocket push
                    let priority = match incoming.priority.as_deref() {
                        Some("high") => crate::backend::messagebus::Priority::High,
                        Some("urgent") => crate::backend::messagebus::Priority::Urgent,
                        _ => crate::backend::messagebus::Priority::Normal,
                    };
                    match state.messagebus.inject(from, "", target, message, priority) {
                        Ok(msg_id) => {
                            // Sender-side echo for the messagebus fallback path.
                            // Tier is unknown here (no ReactiveHandler wrap) — None
                            // renders as the default "coord".
                            super::reactive::echo_jekt_to_sender(
                                state,
                                Some(from),
                                target,
                                message,
                                &msg_id,
                                None,
                                None,
                                super::reactive::EchoTrust {
                                    delivery_tier: "host",
                                    sig_verified: None,
                                    reagent_verified: None,
                                    lan_verified: None,
                                    channel_verified: None,
                                },
                                incoming.priority.as_deref().unwrap_or("normal"),
                            );
                            let ack = json!({ "type": "bus:injected", "via": "messagebus", "message_id": msg_id });
                            let msg = serde_json::to_string(&ack).unwrap_or_default();
                            if socket.send(Message::Text(msg.into())).await.is_err() {
                                return Err(true);
                            }
                        }
                        Err(e) => {
                            let err = json!({ "type": "bus:error", "error": e });
                            let msg = serde_json::to_string(&err).unwrap_or_default();
                            if socket.send(Message::Text(msg.into())).await.is_err() {
                                return Err(true);
                            }
                        }
                    }
                }
                return Ok(None);
            }
            "bus:broadcast" => {
                let from = incoming.from.as_deref().unwrap_or("unknown");
                if let Some(ref payload) = incoming.payload {
                    let priority = match incoming.priority.as_deref() {
                        Some("high") => crate::backend::messagebus::Priority::High,
                        Some("urgent") => crate::backend::messagebus::Priority::Urgent,
                        _ => crate::backend::messagebus::Priority::Normal,
                    };
                    let _ = state.messagebus.broadcast(from, "", payload, priority);
                }
                return Ok(None);
            }
            _ => {}
        }
    }

    // Handle wscommand-based messages
    if let Some(ref wscommand) = incoming.wscommand {
        match wscommand.as_str() {
            "rpc" => {
                if let Some(rpc_msg) = incoming.message {
                    engine.handle_message(rpc_msg);
                } else {
                    tracing::warn!("ws: rpc wscommand missing message field");
                }
            }
            "blockinput" => {
                if let Some(ref block_id) = incoming.blockid {
                    if let Some(ref data64) = incoming.inputdata64 {
                        if !data64.is_empty() {
                            match base64::engine::general_purpose::STANDARD.decode(data64) {
                                Ok(data) => {
                                    if let Err(e) = dispatch_blockinput(block_id, data) {
                                        tracing::debug!("ws: blockinput error: {}", e);
                                    }
                                }
                                Err(e) => {
                                    tracing::warn!("ws: blockinput base64 decode error: {}", e);
                                }
                            }
                        }
                    }
                }
            }
            "setblocktermsize" => {
                if let Some(ref block_id) = incoming.blockid {
                    if let Some(ref ts_val) = incoming.termsize {
                        match serde_json::from_value::<TermSize>(ts_val.clone()) {
                            Ok(ts) => {
                                let input = blockcontroller::BlockInputUnion::resize(ts);
                                if let Err(e) = blockcontroller::send_input(block_id, input, None) {
                                    tracing::debug!("ws: setblocktermsize error: {}", e);
                                }
                            }
                            Err(e) => {
                                tracing::warn!("ws: setblocktermsize parse error: {}", e);
                            }
                        }
                    }
                }
            }
            other => {
                tracing::warn!("ws: unknown wscommand: {}", other);
            }
        }
    }

    Ok(None)
}

/// How far back the fleet list (`COMMAND_LIST_BACKGROUND_TASKS` with an empty
/// `blockid`) reaches for tasks that already ended. Comfortably longer than the
/// longest finished-row retention the renderer applies (15 s for an error), so
/// a finished row is still in the list for as long as it is shown.
const FLEET_ENDED_WINDOW_MS: i64 = 60_000;

pub(super) fn register_handlers(engine: &Arc<WshRpcEngine>, state: AppState, conn_id: String) {
    // getfullconfig → return full config as JSON
    let config_watcher = state.config_watcher.clone();
    // The RESPONSE stays `serde_json::Value`, deliberately. Redaction runs on
    // the serialized form (`redact_full_config_for_renderer` walks the JSON and
    // strips secrets), so what goes on the wire is NOT `FullConfigType` -- it
    // is `FullConfigType` minus fields that a type would still claim are
    // there. Generating `FullConfigType` here would produce a binding that
    // promises the renderer secrets it will never receive, which is worse
    // than `unknown`.
    engine.register_typed(
        COMMAND_GET_FULL_CONFIG,
        move |_req: Option<NoArgsReq>, _ctx| {
            let cw = config_watcher.clone();
            async move {
                let config = cw.get_full_config();
                let mut v = serde_json::to_value(config.as_ref())
                    .map_err(|e| format!("failed to serialize config: {}", e))?;
                crate::backend::wconfig::redact_full_config_for_renderer(&mut v);
                Ok(v)
            }
        },
    );

    // routeannounce → log + no-op (fire-and-forget, may have no reqid)
    // `Value` request: fire-and-forget, may arrive with no reqid and any
    // payload, and it only logs. A struct would start rejecting shapes this has
    // always accepted.
    engine.register_typed(
        COMMAND_ROUTE_ANNOUNCE,
        |data: serde_json::Value, _ctx| {
            async move {
                tracing::debug!("routeannounce: {:?}", data);
                Ok(())
            }
        },
    );

    // routeunannounce → no-op
    engine.register_typed(
        COMMAND_ROUTE_UNANNOUNCE,
        |_data: serde_json::Value, _ctx| async move { Ok(()) },
    );

    // eventsub → register subscription with the MPS broker
    let broker_sub = state.broker.clone();
    let conn_id_sub = conn_id.clone();
    engine.register_typed(
        COMMAND_EVENT_SUB,
        move |sub: crate::backend::mps::SubscriptionRequest, _ctx| {
            let broker = broker_sub.clone();
            let conn_id = conn_id_sub.clone();
            async move {
                tracing::debug!("eventsub: event={} scopes={:?} allscopes={}", sub.event, sub.scopes, sub.allscopes);
                broker.subscribe(&conn_id, sub);
                Ok(())
            }
        },
    );

    // eventunsub → unsubscribe from the MPS broker
    let broker_unsub = state.broker.clone();
    let conn_id_unsub = conn_id.clone();
    // The request is a BARE STRING, not an object -- the only command in the
    // app shaped that way, which is exactly the sort of thing a hand-written
    // stub gets wrong. Typing it as `String` makes the generated binding say
    // so. `Option<String>` rather than `String` because the old handler
    // answered a non-string payload with "" (no-op) rather than an error, and
    // that tolerance is cheap to keep.
    engine.register_typed(
        COMMAND_EVENT_UNSUB,
        move |data: serde_json::Value, _ctx| {
            let broker = broker_unsub.clone();
            let conn_id = conn_id_unsub.clone();
            async move {
                // `as_str().unwrap_or("")` -- byte-identical to the untyped
                // handler. A number/object/array is a silent no-op, as before,
                // NOT a deserialize error.
                let event_name = data.as_str().unwrap_or("").to_string();
                if !event_name.is_empty() {
                    broker.unsubscribe(&conn_id, &event_name);
                }
                Ok(())
            }
        },
    );

    // eventunsuball → unsubscribe all from the MPS broker
    let broker_unsub_all = state.broker.clone();
    let conn_id_unsub_all = conn_id.clone();
    engine.register_typed(
        COMMAND_EVENT_UNSUB_ALL,
        move |_req: Option<NoArgsReq>, _ctx| {
            let broker = broker_unsub_all.clone();
            let conn_id = conn_id_unsub_all.clone();
            async move {
                broker.unsubscribe_all(&conn_id);
                Ok(())
            }
        },
    );

    // setmeta → update object metadata in the DB, broadcast update event
    let mstore_sm = state.mstore.clone();
    let event_bus_sm = state.event_bus.clone();
    // `Value` request, parsed into its struct below (which uses the hand-written ORef/MetaType); the stub keeps its
    // hand-written request type (generated-manifest.json `dataType`).
    engine.register_typed(COMMAND_SET_META, move |data: serde_json::Value, _ctx| {
            let mstore = mstore_sm.clone();
            let event_bus = event_bus_sm.clone();
        async move {
                let cmd: CommandSetMetaData =
                    serde_json::from_value(data).map_err(|e| format!("setmeta: {e}"))?;
                let oref_str = cmd.oref.to_string();
                let meta_keys: Vec<&String> = cmd.meta.keys().collect();
                // debug, not info: fires on every metadata write (zoom, title,
                // view-state edits); at info this became a meaningful slice of
                // an unrotated launcher-log mirror (SPEC_WIN10_PAGEFILE_OOM_
                // CRASH_2026_06_29 P1). Default production filter is info, so
                // this is now suppressed unless RUST_LOG=debug is set.
                tracing::debug!(oref = %oref_str, keys = ?meta_keys, "SetMeta");
                crate::server::browser_owner::guard_client_meta_write(&oref_str, &cmd.meta)?;
                update_object_meta(&mstore, &oref_str, &cmd.meta)?;
                // Per-agent zoom persistence (SPEC_AGENT_ZOOM_PERSISTENCE): the
                // frontend writes term:zoom via this WebSocket path, not the HTTP
                // UpdateObjectMeta handler where the mirror was originally placed.
                // Mirror term:zoom → ui:zoom here so the zoom survives pane close.
                let oref_parsed = crate::backend::ORef::parse(&oref_str)
                    .map_err(|e| e.to_string())?;
                if oref_parsed.otype == "block" && cmd.meta.contains_key("term:zoom") {
                    if let Ok(block) = mstore.must_get::<Block>(&oref_parsed.oid) {
                        let agent_id = block.meta.get("agentId")
                            .and_then(|v| v.as_str()).unwrap_or("").to_string();
                        if !agent_id.is_empty() {
                            let zoom = cmd.meta.get("term:zoom").and_then(|v| v.as_f64());
                            schedule_agent_zoom_mirror(mstore.clone(), agent_id, zoom);
                        }
                    }
                }
                // Read the updated object and broadcast a proper MuxObjUpdate
                // so all WS clients refresh their atoms with the new data.
                let oref = crate::backend::ORef::parse(&oref_str)
                    .map_err(|e| e.to_string())?;
                // Only a block's update carries its object; any other went out with no data,
                // which the frontend ignores, so it isn't sent.
                if oref.otype == "block" {
                    if let Ok(block) = mstore.must_get::<Block>(&oref.oid) {
                        let update = MuxObjUpdate {
                            updatetype: "update".into(),
                            otype: oref.otype.clone(),
                            oid: oref.oid.clone(),
                            obj: Some(mux_obj_to_value(&block)),
                        };
                        event_bus.broadcast_event(&crate::backend::eventbus::EventBus::typed_ws_event::<crate::backend::mps_events::MuxObjUpdated>(&update, oref_str));
                    }
                }
                Ok(())
        }
    });

    // getmeta → return metadata for a wave object
    let mstore_gm = state.mstore.clone();
    // `Value` request and response: the meta map is what the frontend models as
    // MetaType (generated-manifest.json `dataType` / `returns`).
    engine.register_typed(COMMAND_GET_META, move |data: serde_json::Value, _ctx| {
            let mstore = mstore_gm.clone();
        async move {
                let cmd: CommandGetMetaData =
                    serde_json::from_value(data).map_err(|e| format!("getmeta: {e}"))?;
                let obj: Option<serde_json::Value> = mstore
                    .get_raw(&cmd.oref.otype, &cmd.oref.oid)
                    .map_err(|e| format!("getmeta: {e}"))?;
                match obj {
                    Some(val) => {
                        // Return the "meta" field if present, otherwise the full object
                        let meta = val.get("meta").cloned().unwrap_or(val);
                        Ok(meta)
                    }
                    None => Err(format!("getmeta: object {} not found", cmd.oref)),
                }
        }
    });

    // waveinfo → return version and build info
    let version_info = state.version.clone();
    engine.register_typed(
        COMMAND_APP_INFO,
        move |_req: Option<NoArgsReq>, _ctx| {
            let version = version_info.clone();
            async move { Ok(AppInfoResult { version }) }
        },
    );

    // getwaveairatelimit → AgentMux has no rate limits; return unlimited/unknown
    engine.register_typed(
        COMMAND_GET_AI_RATE_LIMIT,
        |_req: Option<NoArgsReq>, _ctx| async move {
            Ok(AiRateLimitResult {
                req: 9999,
                reqlimit: 9999,
                preq: 9999,
                preqlimit: 9999,
                resetepoch: 0,
                unknown: true,
            })
        },
    );

    // controllerresync → load block from DB, create/restart controller with PTY
    let mstore_resync = state.mstore.clone();
    let broker_resync = state.broker.clone();
    let event_bus_resync = state.event_bus.clone();
    let filestore_resync = state.filestore.clone();
    let id_store_resync = state.id_store.clone();
    let identity_store_resync = state.identity_store.clone();
    let boot_id_resync = state.boot_id.clone();
    let auth_key_resync = state.auth_key.clone();
    let config_watcher_resync = state.config_watcher.clone();
    // `Value` request, parsed into its struct below (termsize and the rest); the stub keeps its
    // hand-written request type (generated-manifest.json `dataType`).
    engine.register_typed(COMMAND_CONTROLLER_RESYNC, move |data: serde_json::Value, _ctx| {
            let mstore = mstore_resync.clone();
            let broker = broker_resync.clone();
            let event_bus = event_bus_resync.clone();
            let filestore = filestore_resync.clone();
            let id_store = id_store_resync.clone();
            let identity_store = identity_store_resync.clone();
            let boot_id = boot_id_resync.clone();
            let auth_key = auth_key_resync.clone();
            let config_watcher = config_watcher_resync.clone();
        async move {
                let cmd: CommandControllerResyncData = serde_json::from_value(data)
                    .map_err(|e| format!("controllerresync: {e}"))?;
                tracing::info!(
                    block_id = %cmd.blockid,
                    tab_id = %cmd.tabid,
                    forcerestart = cmd.forcerestart,
                    norespawn = cmd.norespawn,
                    "ControllerResync"
                );
                let block: Block = mstore
                    .get(&cmd.blockid)
                    .map_err(|e| format!("controllerresync: load block: {e}"))?
                    .ok_or_else(|| format!("controllerresync: block {} not found", cmd.blockid))?;
                let registry = mstore.shared_agent_registry();
                blockcontroller::resync_controller(
                    &block,
                    &cmd.tabid,
                    cmd.rtopts,
                    cmd.forcerestart,
                    // respawn_if_done — see `norespawn`'s own doc comment.
                    // Default (false → true here) preserves TermResyncHandler's
                    // crash-recovery behaviour for every existing caller.
                    !cmd.norespawn,
                    Some(broker),
                    Some(event_bus),
                    Some(mstore),
                    Some(filestore),
                    Some(id_store),
                    Some(identity_store),
                    registry,
                    boot_id,
                    &auth_key,
                    Some(config_watcher),
                )?;
                Ok(())
        }
    });

    // controllerinput → route keyboard input / signals / resize to block controller
    //
    // Drops input for a block currently under an active PtyShell agent lock
    // (`term:agentlockuntil`, docs/specs/SPEC_AGENT_INTERACTIVE_PTY_SHELL_API_2026_09_10.md
    // §10.3) rather than forwarding it — Codex P1 on PR #3194: the
    // frontend's own gate (`AgentShellSubblock.tsx` dropping its
    // `sendDataHandler` calls while locked) is necessarily eventually-
    // consistent — it reacts to a WS-pushed meta value, not something
    // synchronous with the backend's own state — so a human keystroke
    // already in flight when the lock is set could otherwise still reach
    // `send_input` and interleave with the agent's own write, exactly the
    // collision the lease exists to prevent. This check, made at the
    // moment this handler actually runs against the backend's own current
    // state, is the real enforcement; the frontend gate is the UX layer
    // that keeps it from happening in the first place, not the
    // correctness boundary.
    //
    // Keep this check I/O-free. It reads `blockcontroller::agent_lock`, an
    // in-memory map, NOT the block's persisted meta: the original revision
    // called `mstore.get::<Block>()` here, putting a synchronous SQLite
    // query behind `Store`'s single process-wide `Mutex<Connection>` inside
    // an async handler, inline on a Tokio worker thread with no
    // `block_in_place` — the same failure class as the sysinfo incident
    // (#1782). This handler carries the Stop button's SIGINT and
    // `usePtyWidth`'s debounced resizes (which burst while a pane is being
    // dragged); it is NOT the keystroke path — those go through the
    // `blockinput` arm below, which has the same lock check via
    // `dispatch_blockinput` (closed 2026-09-16, previously `agent_lock`'s
    // documented "Known gap" — see its module doc history). See also
    // docs/analysis/ANALYSIS_CROSS_PANE_INPUT_DELAY_REGRESSION_2026_09_15.md.
    engine.register_typed(
        COMMAND_CONTROLLER_INPUT,
        move |cmd: CommandBlockInputData, _ctx| {
            async move {
                if blockcontroller::agent_lock::is_locked(&cmd.blockid) {
                    tracing::debug!(block_id = %cmd.blockid, "controllerinput: dropped — agent lock active");
                    return Ok(());
                }
                let input = parse_block_input(&cmd)?;
                blockcontroller::send_input(&cmd.blockid, input, cmd.seq)?;
                Ok(())
            }
        },
    );

    // createsubblock → create a headless sub-block (no tab/layout entry)
    // parented to an existing block, e.g. a `term`-view PTY embedded in an
    // agent pane's details drawer.
    let mstore_csb = state.mstore.clone();
    // `Value` request, parsed into its struct below (a BlockDef); the stub keeps its
    // hand-written request type (generated-manifest.json `dataType`).
    engine.register_typed(COMMAND_CREATE_SUB_BLOCK, move |data: serde_json::Value, _ctx| {
            let mstore = mstore_csb.clone();
        async move {
                let cmd: CommandCreateSubBlockData = serde_json::from_value(data)
                    .map_err(|e| format!("createsubblock: {e}"))?;

                let mut meta = cmd.blockdef.meta.clone();
                let raw_cwd = meta
                    .get(blockcontroller::META_KEY_CMD_CWD)
                    .and_then(|v| v.as_str())
                    .filter(|s| !s.is_empty())
                    .map(str::to_string);
                if let Some(raw_cwd) = raw_cwd {
                    // The PTY spawn path (blockcontroller/shell/lifecycle.rs)
                    // reads cmd:cwd raw with no MSYS→Windows conversion, unlike
                    // shellexec — normalize here so a Git-Bash-style path
                    // doesn't hit os error 267 on Windows. Mirrors
                    // server/mod.rs's shellexec cwd-fallback precedent: an
                    // invalid/non-absolute value degrades to no cwd rather
                    // than failing the whole call — this is a best-effort
                    // convenience derived from the parent's meta, not a
                    // caller-supplied value worth hard-rejecting.
                    match normalize_working_dir(&raw_cwd).filter(|p| std::path::Path::new(p).is_absolute()) {
                        Some(norm) => {
                            meta.insert(
                                blockcontroller::META_KEY_CMD_CWD.to_string(),
                                serde_json::Value::String(norm),
                            );
                        }
                        None => {
                            tracing::warn!(
                                raw_cwd = %raw_cwd,
                                "createsubblock: invalid or non-absolute cmd:cwd, dropping (no cwd)"
                            );
                            meta.remove(blockcontroller::META_KEY_CMD_CWD);
                        }
                    }
                }

                let child_id = uuid::Uuid::new_v4().to_string();
                let mut block = Block {
                    oid: child_id.clone(),
                    parentoref: format!("block:{}", cmd.parentblockid),
                    meta,
                    ..Default::default()
                };
                mstore
                    .insert(&mut block)
                    .map_err(|e| format!("createsubblock: insert: {e}"))?;

                // Best-effort link into the parent's subblockids. Not a CAS
                // (Store::update is a blind version-bump) — an acceptable
                // race for a single lazily-created shell per agent pane.
                if let Ok(mut parent) = mstore.must_get::<Block>(&cmd.parentblockid) {
                    parent
                        .subblockids
                        .get_or_insert_with(Vec::new)
                        .push(child_id.clone());
                    let _ = mstore.update(&mut parent);
                }

                tracing::info!(
                    child_id = %child_id,
                    parent_id = %cmd.parentblockid,
                    "CreateSubBlock"
                );
                // TS `CreateSubBlockCommand` resolves `Promise<ORef>`, and
                // `ORef` (srv-types.d.ts) is a plain string — return the
                // bare "block:<id>" string, not a wrapper object.
                Ok(format!("block:{child_id}"))
        }
    });

    // deletesubblock → tear down a sub-block created via createsubblock.
    // Kill-first ordering matches sagas/delete_block.rs; unlike that saga,
    // this does NOT touch tab bookkeeping — sub-blocks are never
    // tab-referenced, so delete_block.rs's precondition would reject them.
    let mstore_dsb = state.mstore.clone();
    engine.register_typed(
        COMMAND_DELETE_SUB_BLOCK,
        move |cmd: CommandDeleteSubBlockData, _ctx| {
            let mstore = mstore_dsb.clone();
            async move {

                // Kill process FIRST — a lingering PTY tree is worse than a
                // delayed row delete.
                crate::sagas::agent_teardown::discard(&cmd.blockid);

                let parent_id = mstore
                    .get::<Block>(&cmd.blockid)
                    .ok()
                    .flatten()
                    .and_then(|b| b.parentoref.strip_prefix("block:").map(str::to_string));

                mstore
                    .delete::<Block>(&cmd.blockid)
                    .map_err(|e| format!("deletesubblock: delete: {e}"))?;

                // Best-effort unlink from the parent's subblockids — the
                // controller is already dead and the row is already gone
                // either way, so don't fail the call over this step.
                if let Some(parent_id) = parent_id {
                    if let Ok(mut parent) = mstore.must_get::<Block>(&parent_id) {
                        if let Some(ids) = parent.subblockids.as_mut() {
                            ids.retain(|id| id != &cmd.blockid);
                            let _ = mstore.update(&mut parent);
                        }
                    }
                }

                tracing::info!(block_id = %cmd.blockid, "DeleteSubBlock");
                Ok(())
            }
        },
    );

    // tooldecision → reply to a per-tool-call permission gate via the Agent SDK
    // **control protocol** — the `AgentDecisionPanel` Allow/Deny buttons, once
    // `should_route_to_decision_panel` (persistent/mod.rs) is flipped on for real
    // (that gate is currently hardcoded false, so `decide_tool_permission`
    // below will fail every call today with "no pending tool-permission
    // request" — expected until the Phase 2 policy decision is made; see
    // that function's own doc comment).
    //
    // The original PR-3a draft tried to write `y\n` / `n\n` to the
    // subprocess's stdin via `blockcontroller::send_input`. Codex P1
    // on PR #557 caught that this would fail: `SubprocessController::
    // send_input` (and `PersistentSubprocessController::send_input`)
    // both reject raw `input_data` with an `Err`. The deeper truth is
    // that AgentMux runs the agent CLI in non-interactive `--print`
    // mode — the CLI shows no y/n prompt and stdin carries only the
    // structured message stream, so a y/n write would be meaningless even if
    // the controller accepted it. Real delivery is the same
    // control_response mechanism `agent.answer` uses below for
    // AskUserQuestion. See SPEC_DECISION_PROMPT_2026_04_24.md §9.1
    // and SPEC_AGENT_CONTROL_PROTOCOL_2026_06_15.md.
    //
    // `cmd.request_id` is used here as the `tool_use_id` key
    // `pending_permissions`/`decide_tool_permission` are keyed on — NOT
    // as a separate correlator against the speculative
    // `PermissionRequestEvent` §4.1 describes, which was never built.
    // The eventual mechanism (still unbuilt — see the Phase 2 gate)
    // correlates a `pending_approval` ToolNode by `tool_use_id`, exactly
    // like AskUserQuestion's `awaiting_answer` already does; whatever
    // populates that field on the frontend side should set it to the
    // ToolNode's `tool_use_id`.
    //
    // `scope` is still validated but not yet acted on: §6's rules-
    // persistence layer (`permissions.json`) does not exist anywhere in
    // this tree. That is separate, still-open work — this PR only makes
    // the *transport* (control_response delivery) real, not repeat-
    // decision memory.
    engine.register_typed(
        COMMAND_TOOL_DECISION,
        move |cmd: CommandToolDecisionData, _ctx| async move {
                match cmd.outcome.as_str() {
                    "allow" | "deny" => {}
                    other => {
                        return Err(format!(
                            "tooldecision: invalid outcome '{}' (expected 'allow' or 'deny')",
                            other
                        ));
                    }
                }
                // Validate scope so a future rules-persistence layer (§6)
                // can trust the value without re-checking. Reagent P1
                // round-3 on PR #557.
                match cmd.scope.as_str() {
                    "once" | "session" | "project" | "global" => {}
                    other => {
                        return Err(format!(
                            "tooldecision: invalid scope '{}' (expected 'once'/'session'/'project'/'global')",
                            other
                        ));
                    }
                }
                tracing::info!(
                    block_id = %cmd.blockid,
                    request_id = %cmd.request_id,
                    outcome = %cmd.outcome,
                    scope = %cmd.scope,
                    has_feedback = cmd.feedback.is_some(),
                    "[tooldecision] delivering via control protocol"
                );
                let ctrl = blockcontroller::get_controller(&cmd.blockid)
                    .ok_or_else(|| format!("tooldecision: no controller for block {}", cmd.blockid))?;
                if let Some(persistent_ctrl) = ctrl
                    .as_any()
                    .downcast_ref::<blockcontroller::persistent::PersistentSubprocessController>()
                {
                    persistent_ctrl.decide_tool_permission(cmd.request_id, &cmd.outcome, cmd.feedback)
                } else {
                    Err("tooldecision: UNSUPPORTED_CONTROLLER: per-tool permission decisions \
                         require a persistent (host) agent; container/one-shot agents are not \
                         yet supported (same limitation as agent.answer)".to_string())
                }
        },
    );

    // docknodestatus → fire-and-forget push of a ToolNode's latest status,
    // cached in-memory per block for `muxspect dock` to read. See
    // docs/specs/SPEC_MUXSPECT_DOCK_DIAGNOSIS_AND_REMEDIATION_2026_08_06.md §3.1.
    //
    // Also mirrors a declared-background task's liveness into the durable
    // `db_background_tasks` registry (see
    // docs/status/STATUS_ATTACHED_TASK_AXIS_AND_DEV_LOOP_2026_08_15.md) —
    // `DockSnapshotCache` above is intentionally ephemeral (never persisted,
    // 1-hour eviction on read); the registry is the durable source of truth
    // consumers that need to survive a cache eviction or a session
    // reconnect (Swarm, #2492's teardown-survival) read from instead.
    //
    // `run_in_background == Some(true)` is ONLY ever sent once
    // `isAcceptedBackgroundLaunch()` holds client-side (`tool-adapter.ts`),
    // which by its own definition requires `status == "success"` — checking
    // `status == "running"` here (an earlier version of this handler did)
    // made the observe call unreachable, since the flag and that status
    // value are never sent together (reagentx P0 / codex P1 on PR #2590).
    // Best-effort: a registry write failure never blocks the dock push it
    // rides alongside — `dock_snapshots.push_delta` below already gives the
    // user the live UI signal regardless. Completion is NOT detected here —
    // see `COMMAND_BACKGROUND_TASK_COMPLETION` below for why this handler
    // structurally can't see it (the originating ToolNode's status never
    // changes again after acceptance).
    let dock_snapshots_dns = state.dock_snapshots.clone();
    // Ambient narration — generate a short user-facing line about something
    // AgentMux just did on its own, and broadcast it to the pane.
    //
    // Fire-and-forget in both directions: the renderer does not await a reply,
    // and the work runs in a detached task so a slow or capped model call never
    // holds the RPC open. Every failure path is silence — the narrated action
    // has already happened and the UI already reflects it, so a missing line is
    // a missing nicety, not a missing state change.
    let mstore_narrate = state.mstore.clone();
    let broker_narrate = state.broker.clone();
    let narrated = state.narrated_events.clone();
    engine.register_typed(
        COMMAND_AMBIENT_NARRATE,
        move |cmd: CommandAmbientNarrateData, _ctx| {
            let mstore = mstore_narrate.clone();
            let broker = broker_narrate.clone();
            let narrated = narrated.clone();
            async move {

                // Dedupe per narrated event. A ToolNode can be re-observed, and
                // narrating one event twice is both noise and a wasted model
                // call. Checked before spawning so a repeat costs nothing.
                if !cmd.dedupe_key.is_empty() && !narrated.mark_new(&cmd.dedupe_key) {
                    return Ok(());
                }

                tokio::spawn(async move {
                    let generation = agentmux_common::time::now_ms_u64();
                    let Some(text) = crate::ambient::tasks::generate_ambient_narration(
                        &mstore,
                        &cmd.blockid,
                        &cmd.dedupe_key,
                        generation,
                        &cmd.kind,
                        &cmd.context,
                    )
                    .await
                    else {
                        return;
                    };
                    // A short task can finish while the model call is still
                    // in flight, so by now the dock may already show it done.
                    // Narrating it at that point is worse than silence: the line
                    // describes a launch the user has already seen conclude.
                    //
                    // Belt and braces with the prompt, which is phrased to stay
                    // true after completion — this drops the message entirely
                    // rather than relying on wording alone. (codex P2 on #3169.)
                    //
                    // Only skips on a DEFINITE completion: an unknown id (no
                    // registry row — the task was never observed, or this is a
                    // narration kind that has none) falls through and publishes,
                    // since absence of a row is not evidence of completion.
                    if !cmd.dedupe_key.is_empty() {
                        if let Ok(Some(task)) = mstore.background_task_get(&cmd.dedupe_key) {
                            if task.status != crate::backend::storage::background_tasks::BackgroundTaskStatus::Running {
                                return;
                            }
                        }
                    }
                    // Carries the text itself, unlike background-task-updated
                    // next to it: there is no list query to re-read, so an
                    // invalidation ping would have nothing to invalidate.
                    let narration = crate::backend::mps_events::AmbientNarrationEvent { block_id: cmd.blockid.clone(), kind: cmd.kind.clone(), text: text.to_string() };
                    broker.publish_event::<crate::backend::mps_events::AmbientNarration>(&narration, vec![format!("block:{}", cmd.blockid)], 0);
                });
                Ok(())
            }
        },
    );

    let mstore_dns = state.mstore.clone();
    let pending_pids_dns = state.pending_background_pids.clone();
    let broker_dns = state.broker.clone();
    engine.register_typed(
        COMMAND_DOCK_NODE_STATUS,
        move |cmd: CommandDockNodeStatusData, _ctx| {
            let dock_snapshots = dock_snapshots_dns.clone();
            let mstore = mstore_dns.clone();
            let pending_pids = pending_pids_dns.clone();
            let broker = broker_dns.clone();
            async move {
                let observed_at = agentmux_common::time::now_ms();

                if cmd.run_in_background == Some(true) {
                    match mstore.background_task_observe(
                        &cmd.node_id,
                        &cmd.blockid,
                        &cmd.tool_name,
                        observed_at,
                        observed_at,
                    ) {
                        Ok(()) => publish_background_task_updated(&broker, &cmd.blockid),
                        Err(e) => tracing::warn!(
                            target: "background_tasks",
                            node_id = %cmd.node_id,
                            error = %e,
                            "failed to observe declared-background task in the durable registry",
                        ),
                    }
                    // bashwrap's own pid publish (COMMAND_BACKGROUND_TASK_PID
                    // below) routinely races ahead of the observe call above —
                    // it fires at bashwrap's process start, before the frontend
                    // even sees an accepted-background tool result. Check for
                    // (and apply) a pid stashed ahead of this row's existence.
                    // Runs unconditionally on observe failure too — a stashed
                    // pid may still apply if the row already existed from an
                    // earlier retry. Atomic with the pid handler's own
                    // set_or_stash call for the same id — see
                    // pending_background_pids.rs's module doc for why that
                    // matters (Codex/reagentx findings on PR #2681).
                    let node_id = cmd.node_id.clone();
                    let mstore_apply = mstore.clone();
                    let result: Result<(), crate::backend::storage::StoreError> = pending_pids
                        .observe_and_apply(&cmd.node_id, observed_at, |pid| {
                            mstore_apply.background_task_set_pid(&node_id, pid)
                        });
                    if let Err(e) = result {
                        tracing::warn!(
                            target: "background_tasks",
                            node_id = %cmd.node_id,
                            error = %e,
                            "failed to apply a pid stashed ahead of this task's registry row",
                        );
                    }
                }

                dock_snapshots.push_delta(
                    &cmd.blockid,
                    crate::backend::dock_snapshot::DockNodeSnapshot {
                        node_id: cmd.node_id,
                        tool_name: cmd.tool_name,
                        status: cmd.status,
                        timestamp: cmd.timestamp,
                        observed_at,
                        run_in_background: cmd.run_in_background,
                    },
                );
                Ok(())
            }
        },
    );

    // backgroundtaskcompletion → fire-and-forget push of a declared-
    // background task's REAL terminal outcome, parsed client-side from its
    // `<task-notification>` message (`tool-adapter.ts`'s
    // `parseTaskNotification`). This is a separate command from
    // `docknodestatus` above on purpose: the originating ToolNode's own
    // `status` field goes "success" at acceptance and never changes again
    // (that's the raw tool_result's outcome, not the background task's) —
    // there is no later `docknodestatus` push this handler could key off of
    // to learn a background task actually finished (reagentx P0 / codex P1
    // on PR #2590, corrected here). Routing this through `docknodestatus`
    // instead of a dedicated command would also corrupt `DockSnapshotCache`:
    // `push_delta` fully overwrites a node's snapshot, and this event has no
    // `tool_name`/original `run_in_background` value to carry forward — the
    // exact bug class #2520 already fixed once for a different call site.
    let mstore_btc = state.mstore.clone();
    let broker_btc = state.broker.clone();
    engine.register_typed(
        COMMAND_BACKGROUND_TASK_COMPLETION,
        move |cmd: CommandBackgroundTaskCompletionData, _ctx| {
            let mstore = mstore_btc.clone();
            let broker = broker_btc.clone();
            async move {
                let ended_at = cmd.timestamp.unwrap_or_else(|| {
                    agentmux_common::time::now_ms()
                });
                let status = crate::backend::storage::background_tasks::BackgroundTaskStatus::from_str(&cmd.status);
                match mstore.background_task_complete(&cmd.node_id, status, ended_at) {
                    Ok(_) => publish_background_task_updated(&broker, &cmd.blockid),
                    Err(e) => tracing::warn!(
                        target: "background_tasks",
                        node_id = %cmd.node_id,
                        error = %e,
                        "failed to mark background task terminal in the durable registry",
                    ),
                }
                Ok(())
            }
        },
    );

    // backgroundtaskpid → fire-and-forget push of a declared-background
    // task's real OS pid, relayed from `agentmux-bashwrap`'s own MPS "pid"
    // chunk. Closes the gap where `db_background_tasks.pid` existed but
    // nothing in production ever wrote it (Phase A of
    // docs/specs/SPEC_BACKGROUND_TASK_PID_CAPTURE_2026_08_20.md). Best-effort,
    // same as the two handlers above: a write failure here doesn't block
    // anything else and is only ever a diagnostic/teardown-survival input,
    // never something the model-visible tool_result depends on.
    //
    // bashwrap publishes this essentially at process start — routinely
    // BEFORE `COMMAND_DOCK_NODE_STATUS` above has created this task's
    // `db_background_tasks` row (that requires a full round-trip through
    // the frontend recognizing an accepted background launch first).
    // `background_task_set_pid` silently no-ops on a missing row
    // (`Ok(false)`), which would lose the pid permanently with no way to
    // retry a write that will never succeed — stash it instead, and
    // `COMMAND_DOCK_NODE_STATUS`'s handler applies it the moment the row
    // exists. See Codex/reagentx findings on PR #2681.
    let mstore_btp = state.mstore.clone();
    let pending_pids_btp = state.pending_background_pids.clone();
    let broker_btp = state.broker.clone();
    engine.register_typed(
        COMMAND_BACKGROUND_TASK_PID,
        move |cmd: CommandBackgroundTaskPidData, _ctx| {
            let mstore = mstore_btp.clone();
            let pending_pids = pending_pids_btp.clone();
            let broker = broker_btp.clone();
            async move {
                let now_ms = agentmux_common::time::now_ms();
                let node_id = cmd.node_id.clone();
                let mstore_set = mstore.clone();
                let result = pending_pids.set_or_stash(&cmd.node_id, cmd.pid as i64, now_ms, |pid| {
                    mstore_set.background_task_set_pid(&node_id, pid)
                });
                match result {
                    Ok(()) => publish_background_task_updated(&broker, &cmd.blockid),
                    Err(e) => tracing::warn!(
                        target: "background_tasks",
                        node_id = %cmd.node_id,
                        error = %e,
                        "failed to record background task pid in the durable registry",
                    ),
                }
                Ok(())
            }
        },
    );

    // listbackgroundtasks → request/response: this block's current
    // db_background_tasks rows, so the frontend can seed its attachedTask
    // axis from the durable registry on mount/reconnect instead of only
    // ever re-deriving it from this tab's own live transcript replay
    // (which has no way to know about a task that survived a session
    // restart under a controller generation with no transcript history of
    // ever launching it — Phase B of
    // docs/specs/SPEC_BACKGROUND_TASK_TEARDOWN_SURVIVAL_2026_08_20.md).
    // See docs/specs/SPEC_BACKGROUND_TASK_DASHBOARD_INTELLIGENCE_2026_08_20.md §3.1.
    let mstore_lbt = state.mstore.clone();
    engine.register_typed(
        COMMAND_LIST_BACKGROUND_TASKS,
        move |cmd: CommandListBackgroundTasksData, _ctx| {
            let mstore = mstore_lbt.clone();
            async move {
                // An empty blockid asks for the whole fleet (Swarm): every
                // running task plus those that ended recently enough to still
                // show a finished row. SPEC_BACKGROUND_TASK_STRUCTURED_FEED_
                // AND_SWARM_OWNERSHIP_2026_09_27.md §3.
                let tasks = if cmd.blockid.is_empty() {
                    let now_ms = agentmux_common::time::now_ms();
                    mstore.background_task_list_fleet(now_ms - FLEET_ENDED_WINDOW_MS)
                } else {
                    mstore.background_task_list_for_block(&cmd.blockid)
                }
                .map_err(|e| format!("listbackgroundtasks: {e}"))?;
                let views: Vec<super::muxspect_handlers::BackgroundTaskView> =
                    tasks.into_iter().map(Into::into).collect();
                Ok(views)
            }
        },
    );

    // agent.answer → deliver an AskUserQuestion answer to the running agent CLI
    // via the Agent SDK **control protocol**: the persistent controller replies
    // to the CLI's parked `can_use_tool` control_request with a control_response
    // carrying `updatedInput.answers`. (Delivering a `tool_result` on stdin does
    // NOT work — the CLI auto-rejects AskUserQuestion within the turn; see spec
    // §2/§3.) Only the persistent controller speaks the control protocol;
    // container/one-shot subprocess agents return UNSUPPORTED_CONTROLLER (Phase 2).
    // Spec: docs/specs/SPEC_AGENT_CONTROL_PROTOCOL_2026_06_15.md.
    // `Value` request, parsed into its struct below (an answer to a question); the stub keeps its
    // hand-written request type (generated-manifest.json `dataType`).
    engine.register_typed(COMMAND_AGENT_ANSWER, move |data: serde_json::Value, _ctx| {
        async move {
                let cmd: CommandAgentAnswerData = serde_json::from_value(data)
                    .map_err(|e| format!("agent.answer: {e}"))?;
                with_question_controller("agent.answer", "answering", &cmd.blockid, &cmd.tool_use_id, |ctrl| {
                    ctrl.answer_question(cmd.tool_use_id.clone(), cmd.answers)
                })?;
                tracing::info!(
                    block_id = %cmd.blockid,
                    tool_use_id = %cmd.tool_use_id,
                    "[agent.answer] control_response delivered to persistent stdin"
                );
                Ok(())
        }
    });

    // agent.cancel → a REAL protocol-level decline of a pending AskUserQuestion
    // (Cancel button / Escape in AgentQuestionPanel.tsx), not a UI-only dismiss.
    // Sends `behavior: "deny"` over the same control protocol agent.answer uses
    // for `behavior: "allow"` — see `deny_question`'s doc comment for why this
    // is a general, documented Agent SDK mechanism rather than something
    // special-cased for AskUserQuestion. Its error strings come from
    // `with_question_controller`, shared with agent.answer (`no controller for
    // block`, `UNSUPPORTED_CONTROLLER`): the frontend's SAFE_TO_RETRY_VIA_FOLLOWUP
    // allowlist (useAgentQuestions.ts) matches on these substrings for both
    // commands. Spec: docs/specs/SPEC_AGENT_CONTROL_PROTOCOL_2026_06_15.md.
    engine.register_typed(
        COMMAND_AGENT_CANCEL,
        move |cmd: CommandAgentCancelData, _ctx| async move {
                with_question_controller("agent.cancel", "canceling", &cmd.blockid, &cmd.tool_use_id, |ctrl| {
                    ctrl.deny_question(
                        cmd.tool_use_id.clone(),
                        blockcontroller::persistent::ASK_USER_QUESTION_DENY_MESSAGE.to_string(),
                    )
                })?;
                tracing::info!(
                    block_id = %cmd.blockid,
                    tool_use_id = %cmd.tool_use_id,
                    "[agent.cancel] deny control_response delivered to persistent stdin"
                );
                Ok(())
        },
    );

    // The pending-shutdown banner's "Keep running"
    // (docs/specs/SPEC_AGENT_SELF_QUIT_2026_09_24.md §6.5).
    let state_keep = state.clone();
    engine.register_typed(
        COMMAND_AGENT_SHUTDOWN_KEEP,
        move |cmd: CommandAgentShutdownKeepData, _ctx| {
            let state = state_keep.clone();
            async move {
                let outcome = crate::sagas::pending_shutdown::keep(&state, &cmd.blockid, &cmd.request_id);
                Ok(AgentShutdownKeepResult { outcome: outcome.to_string() })
            }
        },
    );

    // Agent input/stop + subprocess spawn handlers
    super::agent_handlers::register_agent_input_handlers(engine, &state);

    // `/btw` slash command — one-shot, tool-less side question.
    super::agent_handlers::register_side_question_handlers(engine, &state);

    // Shell exec/stop handlers
    super::shell_handlers::register_shell_handlers(engine, &state);

    // Editor/file-ops + write_agent_config handlers
    super::editor_handlers::register_editor_handlers(engine, &state);

    // Files pane: fs.list/places/watch/unwatch + mutations
    // (SPEC_FILE_BROWSER_PANE_2026_10_01.md §6.3)
    super::fs_handlers::register_fs_handlers(engine, &state);

    // LSP handlers (lspstart, lspsend, lspstop)
    super::lsp_handlers::register_lsp_handlers(engine, &state);

    // CLI handlers (resolvecli, checkcliauth, runclilogin), then widget packages (widgets.*)
    super::cli_handlers::register_cli_handlers(engine, &state);
    super::widget_handlers::register_widget_handlers(engine, &state);

    // Tool store handlers (gettoolstatus, installtool)
    super::tool_handlers::register_tool_handlers(engine, &state);

    // Provider model catalog (providers.models → authoritative /v1/models list)
    super::providers_handlers::register_providers_handlers(engine, &state);

    // reactive.registrations → Stash "Registration" tab live status (#2696)
    super::reactive::register_reactive_ws_handlers(engine, &state);

    // eventreadhistory → read persisted event history from the MPS broker
    let broker_history = state.broker.clone();
    engine.register_typed(
        COMMAND_EVENT_READ_HISTORY,
        move |cmd: CommandEventReadHistoryData, _ctx| {
            let broker = broker_history.clone();
            async move {
                let max_items = if cmd.maxitems == 0 { 1024 } else { cmd.maxitems };
                Ok(broker.read_event_history(&cmd.event, &cmd.scope, max_items))
            }
        },
    );

    // setconfig → merge settings keys into settings.json AND update in-memory config immediately.
    // Writing to disk + broadcasting directly gives instant UI response without waiting for
    // the fs watcher (which has a ~300-800ms debounce + polling delay on Windows).
    // The fs watcher's subsequent reload is a no-op (settings already up to date).
    let config_watcher_setconfig = state.config_watcher.clone();
    let event_bus_setconfig = state.event_bus.clone();

    let lan_listeners_setconfig = state.lan_listeners.clone();
    // `Value` request, parsed into its struct below (a partial settings object); the stub keeps its
    // hand-written request type (generated-manifest.json `dataType`).
    engine.register_typed(COMMAND_SET_CONFIG, move |data: serde_json::Value, _ctx| {
            let cw = config_watcher_setconfig.clone();
            let eb = event_bus_setconfig.clone();

            let lan_listeners = lan_listeners_setconfig.clone();
        async move {
                let new_keys: serde_json::Map<String, serde_json::Value> =
                    serde_json::from_value(data).map_err(|e| format!("setconfig: {e}"))?;

                // 1. Write to disk (fs watcher will re-broadcast, harmlessly)
                crate::backend::config_watcher_fs::merge_settings_to_disk(new_keys.clone())
                    .map_err(|e| format!("setconfig write: {e}"))?;

                // 2. Update in-memory config immediately
                let merged_settings = crate::backend::config_watcher_fs::merge_settings_into_current(&cw, new_keys);
                let lan_enabled = merged_settings.network_lan_discovery;
                cw.update_settings(merged_settings);

                // 3. Live-toggle LAN discovery if the key changed. `apply` is
                //    idempotent so it's safe to call unconditionally — when the
                //    daemon is already in the requested state, this is a no-op.
                //    See docs/specs/lan-discovery-toggle.md.
                //
                //    Only the LISTENER supervisor is driven here. It owns the
                //    advertise/reachable pairing and re-gates `lan` itself on
                //    every reconcile, so mDNS never advertises an address
                //    nothing is listening on ("discovered but unreachable").
                //    Calling `lan.apply` here as well would fight that gate —
                //    it would advertise immediately, before any listener is
                //    bound, which is the exact state being designed out.
                lan_listeners.apply(lan_enabled);

                // 4. Broadcast updated config now — no waiting for fs watcher.
                //    Shared with the settings-watcher reload path and the
                //    browser-start-page watcher — see its own doc comment
                //    for why this used to be duplicated inline here.
                crate::backend::config_watcher_fs::broadcast_full_config(&cw, &eb);
                Ok(())
        }
    });

    // Agent handlers (definitions, content, skills, history, import, reseed)
    super::agent_handlers::register_agent_handlers(engine, &state);

    // Drone handlers (issue #753 — Drone pane DAG executor)
    super::drone_handlers::register_drone_handlers(engine, &state);

    // Pre-launch OAuth handlers (auth.start / poll / submitcallback /
    // cancel — see docs/specs/SPEC_PRE_LAUNCH_OAUTH_FLOW_2026_05_14.md)
    super::identity_handlers::register_identity_handlers(engine, &state);

    // Install handlers (install.start / install.cancel — see
    // docs/specs/SPEC_AGENT_INSTALL_STAGE_2026_05_17.md)
    super::install_handlers::register_install_handlers(engine, &state);

    // System-toolchain install handlers (toolchain.resolve_install_command /
    // toolchain.install_system_tool — see
    // docs/specs/SPEC_SYSTEM_TOOLCHAIN_INSTALLER_2026_08_24.md). Shares
    // state.install_sessions with the handlers just above (same
    // InstallSessionRegistry — install.cancel already works unchanged for
    // these sessions too, no new cancel command needed).
    super::system_install_handlers::register_system_install_handlers(engine, &state);

    // MuxAPI handlers (agent.open, agent.send, agent.stop, agent.status, agent.list, agent.output)
    super::app_api::register_app_api_handlers(engine, &state);

    // MuxBus cloud connectivity (muxbus.login / muxbus.login.cancel / muxbus.status / muxbus.disconnect / muxbus.cloudconfig)
    super::muxbus_handlers::register_muxbus_handlers(engine, &state);

    // Native memory file browser (agent:memory:list / read_file / write_file)
    super::native_memory_handlers::register_native_memory_handlers(engine, &state);

    // Memory delivery: the hidden reinjection's claim against the
    // SessionStart hook (memorydelivery:claim_fallback).
    super::memory_delivery_handlers::register_memory_delivery_handlers(engine, &state);

    // OS notification Router (notify.emit / focus / ack / test / takeactivation
    // — docs/specs/SPEC_OS_NOTIFICATIONS_SYSTEM_2026_09_24.md §3.3).
    super::notify_handlers::register_notify_handlers(engine, &state, conn_id.clone());
}


/// Route a decoded `blockinput` payload — the actual human-keystroke path
/// (`termViewModel.ts`'s `blockinput` WS command, used by both the
/// standalone Terminal widget and an agent pane's composer-drawer shell) —
/// to its block's controller.
///
/// Mirrors `controllerinput`'s own lock check above: drops the input
/// outright while `block_id` is under an active PtyShell agent lease,
/// rather than forwarding it. This was `agent_lock`'s documented "Known
/// gap" — `blockinput`, not `controllerinput`, is where a human's real
/// keystrokes actually travel, and until now nothing server-side stopped
/// one from interleaving with an agent's own `PtyShellInput` write into the
/// same PTY. The frontend's `agentLocked()` gate is still the primary UX
/// layer (stops it before it's even sent), but it is eventually consistent;
/// this is the synchronous backstop against this handler's own state.
fn dispatch_blockinput(block_id: &str, data: Vec<u8>) -> Result<(), String> {
    if blockcontroller::agent_lock::is_locked(block_id) {
        tracing::debug!(block_id = %block_id, "blockinput: dropped — agent lock active");
        return Ok(());
    }
    let input = blockcontroller::BlockInputUnion::data(data);
    blockcontroller::send_input(block_id, input, None)
}

/// Run `f` on `block_id`'s persistent controller for `agent.answer` /
/// `agent.cancel`. The error strings are matched by the frontend's
/// SAFE_TO_RETRY_VIA_FOLLOWUP allowlist (useAgentQuestions.ts).
fn with_question_controller<T>(
    command: &str,
    action: &str,
    block_id: &str,
    tool_use_id: &str,
    f: impl FnOnce(&blockcontroller::persistent::PersistentSubprocessController) -> Result<T, String>,
) -> Result<T, String> {
    if tool_use_id.is_empty() {
        return Err(format!("{command}: MISSING_ARG: tool_use_id"));
    }
    let ctrl = blockcontroller::get_controller(block_id)
        .ok_or_else(|| format!("{command}: no controller for block {block_id}"))?;
    match ctrl
        .as_any()
        .downcast_ref::<blockcontroller::persistent::PersistentSubprocessController>()
    {
        Some(persistent_ctrl) => f(persistent_ctrl),
        None => Err(format!(
            "{command}: UNSUPPORTED_CONTROLLER: {action} AskUserQuestion \
             requires a persistent (host) agent; container/one-shot agents are \
             not yet supported (Phase 2)"
        )),
    }
}

/// Parse a CommandBlockInputData into a BlockInputUnion.
fn parse_block_input(
    cmd: &CommandBlockInputData,
) -> Result<blockcontroller::BlockInputUnion, String> {
    if !cmd.inputdata64.is_empty() {
        let data = base64::engine::general_purpose::STANDARD
            .decode(&cmd.inputdata64)
            .map_err(|e| format!("controllerinput: base64 decode: {e}"))?;
        return Ok(blockcontroller::BlockInputUnion::data(data));
    }
    if !cmd.signame.is_empty() {
        return Ok(blockcontroller::BlockInputUnion::signal(&cmd.signame));
    }
    if let Some(ref ts_val) = cmd.termsize {
        let ts: TermSize =
            serde_json::from_value(ts_val.clone()).map_err(|e| format!("controllerinput: {e}"))?;
        return Ok(blockcontroller::BlockInputUnion::resize(ts));
    }
    Err("controllerinput: no input data, signal, or termsize".to_string())
}

// ====================================================================
// Tests
// ====================================================================

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn question_controller_errors_keep_the_strings_the_frontend_matches() {
        let missing = with_question_controller("agent.cancel", "canceling", "blk", "", |_| Ok(()));
        assert_eq!(missing.unwrap_err(), "agent.cancel: MISSING_ARG: tool_use_id");
        let absent = with_question_controller("agent.answer", "answering", "no-such-block-qc", "tu-1", |_| Ok(()));
        assert_eq!(absent.unwrap_err(), "agent.answer: no controller for block no-such-block-qc");
    }

    /// Every no-argument stub on this connection sends `null`, not `{}`.
    ///
    /// `waveinfo`, `getfullconfig`, `eventunsuball`, `routeannounce` and
    /// `routeunannounce` are all `client.rpcCall("...", null, opts)`. That is
    /// the opposite of the assumption the earlier slices were built on -- those
    /// stubs send `{}` -- and it is why these register `Option<NoArgsReq>`
    /// rather than the bare struct: serde deserializes a struct only from an
    /// object, so a bare `NoArgsReq` would reject every one of these calls
    /// while compiling and passing every gate.
    #[test]
    fn the_no_argument_commands_accept_the_null_their_stubs_send() {
        for payload in [serde_json::Value::Null, serde_json::json!({})] {
            serde_json::from_value::<Option<crate::backend::rpc_types::NoArgsReq>>(payload.clone())
                .unwrap_or_else(|e| panic!("{payload} should deserialize, got {e}"));
        }
        // ...and the bare struct really does reject null, so the Option is
        // load-bearing rather than decoration.
        assert!(
            serde_json::from_value::<crate::backend::rpc_types::NoArgsReq>(serde_json::Value::Null)
                .is_err(),
            "if this ever passes, the Option wrappers here are redundant",
        );
    }

    /// `eventunsub` takes a BARE STRING payload -- the only command in the app
    /// shaped that way. Its stub is `data: string`.
    ///
    /// Registered as `serde_json::Value`, NOT `Option<String>`. reagent caught
    /// that difference on #3348 and was right: the old handler read
    /// `data.as_str().unwrap_or("")`, so a number/object/array was a silent
    /// no-op, while `Option<String>` hard-errors on exactly those. Claiming it
    /// "keeps that tolerance" was false. `Value` keeps it for real.
    ///
    /// This test now covers "anything else" rather than only null, which is
    /// what its name always promised.
    #[test]
    fn eventunsub_takes_a_bare_string_and_tolerates_anything_else() {
        // Mirrors the handler exactly: deserialize as Value, then as_str().
        let name = |v: serde_json::Value| {
            serde_json::from_value::<serde_json::Value>(v)
                .expect("Value deserializes from anything")
                .as_str()
                .unwrap_or("")
                .to_string()
        };
        assert_eq!(name(serde_json::json!("blockfile:mux")), "blockfile:mux");
        // Each of these was a silent no-op before the migration and must stay one.
        for junk in [
            serde_json::Value::Null,
            serde_json::json!(42),
            serde_json::json!({ "event": "x" }),
            serde_json::json!(["x"]),
            serde_json::json!(true),
        ] {
            assert_eq!(name(junk.clone()), "", "{junk} must unsubscribe nothing, not error");
        }
    }

    /// Codex P1 on PR #3194: the frontend's own lock gate
    /// (`AgentShellSubblock.tsx` dropping `sendDataHandler` while
    /// `term:agentlockuntil` is in the future) is necessarily eventually-
    /// consistent — it reacts to a WS-pushed meta value, not something
    /// synchronous with the backend's own state — so a human keystroke
    /// already in flight could otherwise still reach `send_input` and
    /// interleave with an agent's own write. `controllerinput` must reject
    /// it itself, checked against the backend's own current state at the
    /// moment it actually runs.
    ///
    /// Distinguishes "silently dropped by the lock gate" from "attempted
    /// and failed for lack of a live controller" by the response's `error`
    /// field: a target block here has no real controller registered at
    /// all, so an UNLOCKED call reaches `blockcontroller::send_input` and
    /// gets a real `"no controller for block"` error back — proving the
    /// call path is live. A LOCKED call must short-circuit before that
    /// point, returning cleanly with no error.
    #[tokio::test]
    async fn controllerinput_is_dropped_while_an_agent_lock_is_active() {
        use crate::backend::rpc::engine::WshRpcEngine;
        use crate::backend::rpc_types::RpcMessage;
        use crate::server::tests::test_state;

        let state = test_state();
        let (engine, mut output_rx) = WshRpcEngine::new();
        register_handlers(&engine, state.clone(), "test-conn".to_string());

        async fn send_input(
            engine: &std::sync::Arc<WshRpcEngine>,
            output_rx: &mut tokio::sync::mpsc::UnboundedReceiver<RpcMessage>,
            block_id: &str,
        ) -> RpcMessage {
            engine.handle_message(RpcMessage {
                command: COMMAND_CONTROLLER_INPUT.to_string(),
                reqid: uuid::Uuid::new_v4().to_string(),
                data: Some(serde_json::json!({ "blockid": block_id, "inputdata64": "eQ==" })), // "y"
                ..Default::default()
            });
            tokio::time::timeout(std::time::Duration::from_secs(2), output_rx.recv())
                .await
                .expect("handler should respond within 2s")
                .expect("response channel should not close")
        }

        // Locked target: no live controller registered, but the check must
        // never reach that far. The lease is seeded through
        // `blockcontroller::agent_lock` — the in-memory registry this
        // handler actually enforces against — NOT through the block's
        // `term:agentlockuntil` meta, which is now the frontend-badge copy
        // only (see the handler's own comment and `agent_lock`'s module doc:
        // the meta read this replaced put a synchronous SQLite query on the
        // per-keystroke path).
        blockcontroller::agent_lock::lock_until(
            "ci-locked-shell",
            agentmux_common::time::now_ms() + 60_000,
        );
        let resp = send_input(&engine, &mut output_rx, "ci-locked-shell").await;
        assert!(
            resp.error.is_empty(),
            "a locked target must be silently dropped, not reach send_input at all: {}",
            resp.error
        );

        // Unlocked target (no lease at all) — proves the call path above is
        // genuinely live: it must reach `send_input` and get a real "no
        // controller" error, not silently succeed.
        let resp = send_input(&engine, &mut output_rx, "ci-unlocked-shell").await;
        assert!(
            resp.error.contains("no controller"),
            "an unlocked target must reach send_input for real: {}",
            resp.error
        );

        // A stale `term:agentlockuntil` in meta must NOT lock anything by
        // itself. This is the fail-open half of moving enforcement into
        // memory (`agent_lock`'s module doc): meta can outlive the process
        // that wrote it, memory cannot, and a lease that survives a restart
        // is a human who can't type `exit` in their own terminal.
        let mut stale_meta_block = crate::backend::obj::Block {
            oid: "ci-stale-meta-shell".to_string(),
            meta: {
                let mut m = crate::backend::obj::MetaMapType::new();
                m.insert(
                    "term:agentlockuntil".to_string(),
                    serde_json::json!(agentmux_common::time::now_ms() + 60_000),
                );
                m
            },
            ..Default::default()
        };
        state.mstore.insert(&mut stale_meta_block).unwrap();
        let resp = send_input(&engine, &mut output_rx, "ci-stale-meta-shell").await;
        assert!(
            resp.error.contains("no controller"),
            "a lease that exists only in stale meta must not lock the keyboard: {}",
            resp.error
        );

        blockcontroller::agent_lock::release("ci-locked-shell");
    }

    /// Regression test for the gap `agent_lock`'s own module doc calls out
    /// by name ("Known gap this does NOT close"): unlike `controllerinput`
    /// above (Stop/SIGINT/resize), `blockinput` is the ACTUAL human-keystroke
    /// path (`termViewModel.ts`, both the standalone Terminal widget and an
    /// agent pane's composer-drawer shell) and had no lease check at all —
    /// the frontend's own `agentLocked()` gate was the only thing standing
    /// between a human keystroke and an agent's own `PtyShellInput` write
    /// landing in the same PTY, and that gate is eventually consistent (it
    /// reacts to a WS-pushed meta value), not synchronous with this
    /// handler's own state. A keystroke slipping through that window
    /// interleaves at the byte level with whatever the agent is mid-write
    /// to — vim's cursor-addressed alternate-screen redraw does not degrade
    /// gracefully under that; it renders as a visibly garbled pane, not a
    /// hang. This is exactly what corrupted a live pane during a vim
    /// investigation on 2026-09-16.
    ///
    /// Same "no controller registered" trick as the `controllerinput` test
    /// above: a LOCKED call must short-circuit before ever reaching
    /// `blockcontroller::send_input`, returning `Ok(())` with no error even
    /// though no controller exists for this block id. An UNLOCKED call DOES
    /// reach `send_input` and gets back a real `"no controller for block"`
    /// error — proving the call path is live, not merely never exercised.
    #[test]
    fn blockinput_is_dropped_while_an_agent_lock_is_active() {
        let locked_id = "ws-blockinput-lock-test-locked";
        blockcontroller::agent_lock::lock_until(locked_id, agentmux_common::time::now_ms() + 60_000);
        let result = dispatch_blockinput(locked_id, b"y".to_vec());
        assert!(
            result.is_ok(),
            "a locked blockinput call must be dropped cleanly, not forwarded to send_input: {result:?}"
        );
        blockcontroller::agent_lock::release(locked_id);

        let unlocked_id = "ws-blockinput-lock-test-unlocked";
        let result = dispatch_blockinput(unlocked_id, b"y".to_vec());
        assert_eq!(
            result,
            Err(format!("no controller for block {unlocked_id}")),
            "an unlocked blockinput call must still reach send_input for real"
        );
    }

    /// The lease must stop dropping input the moment it lapses, with nobody
    /// calling `PtyShellStop` — an agent that crashed mid-write, or simply
    /// stopped writing, must not leave the human unable to type into their
    /// own shell (up to and including typing `exit` to close the pane,
    /// SPEC_TERM_EXIT_RESPAWN_LOOP_2026_09_15.md §10).
    #[tokio::test]
    async fn controllerinput_flows_again_once_the_agent_lease_lapses() {
        use crate::backend::rpc::engine::WshRpcEngine;
        use crate::backend::rpc_types::RpcMessage;
        use crate::server::tests::test_state;

        let state = test_state();
        let (engine, mut output_rx) = WshRpcEngine::new();
        register_handlers(&engine, state, "test-conn-lapse".to_string());

        let block_id = "ci-lapsing-shell";
        blockcontroller::agent_lock::lock_until(block_id, agentmux_common::time::now_ms() + 40);

        let send = |engine: std::sync::Arc<WshRpcEngine>| {
            engine.handle_message(RpcMessage {
                command: COMMAND_CONTROLLER_INPUT.to_string(),
                reqid: uuid::Uuid::new_v4().to_string(),
                data: Some(serde_json::json!({ "blockid": block_id, "inputdata64": "eQ==" })),
                ..Default::default()
            });
        };

        send(engine.clone());
        let resp = tokio::time::timeout(std::time::Duration::from_secs(2), output_rx.recv())
            .await
            .expect("handler should respond within 2s")
            .expect("response channel should not close");
        assert!(resp.error.is_empty(), "still leased: input must be dropped");

        tokio::time::sleep(std::time::Duration::from_millis(80)).await;

        send(engine.clone());
        let resp = tokio::time::timeout(std::time::Duration::from_secs(2), output_rx.recv())
            .await
            .expect("handler should respond within 2s")
            .expect("response channel should not close");
        assert!(
            resp.error.contains("no controller"),
            "a lapsed lease must let input through without any explicit release: {}",
            resp.error
        );
    }

    fn rpc_event(block_id: &str) -> serde_json::Value {
        json!({
            "eventtype": "rpc",
            "data": {
                "command": "eventrecv",
                "data": {
                    "event": "blockfile",
                    "scopes": [format!("block:{block_id}")],
                    "data": {"data64": "aGk="},
                },
            },
        })
    }

    fn raw_waveobj_update(oref: &str) -> serde_json::Value {
        json!({
            "eventtype": "waveobj:update",
            "oref": oref,
            "data": {},
        })
    }

    fn batched_updates_event() -> serde_json::Value {
        json!({
            "eventtype": "waveobj:batchedupdates",
            "data": [],
        })
    }

    #[test]
    fn test_priority_pane_key_from_rpc_blockfile_scopes() {
        let event = rpc_event("abc123");
        assert_eq!(priority_pane_key(&event), Some("block:abc123".to_string()));
    }

    /// Raw (non-rpc) object-model broadcasts are NEVER split by pane —
    /// splitting them would let round-robin reorder them relative to other
    /// object updates for the same object (see `priority_pane_key`'s doc
    /// comment for the concrete resurrected-object failure this prevents).
    #[test]
    fn test_priority_pane_key_raw_waveobj_update_is_non_pane() {
        let event = raw_waveobj_update("block:xyz789");
        assert_eq!(priority_pane_key(&event), Some(NON_PANE_EVENT_KEY.to_string()));
    }

    #[test]
    fn test_priority_pane_key_non_pane_for_batched_updates() {
        let event = batched_updates_event();
        assert_eq!(priority_pane_key(&event), Some(NON_PANE_EVENT_KEY.to_string()));
    }

    /// The core regression this exists to prevent: a noisy pane (A) producing
    /// many queued frames must not delay a quiet pane's (B) single frame by
    /// the full length of A's backlog. Round-robin interleaving means B's
    /// frame comes out within one "round" (bounded by the number of DISTINCT
    /// panes with queued output), not after every one of A's frames.
    #[tokio::test]
    async fn test_fair_drain_priority_interleaves_noisy_and_quiet_panes() {
        let (tx, mut rx) = tokio::sync::mpsc::channel(64);

        // Pane A floods 10 frames; pane B contributes exactly 1, queued
        // in the middle of A's backlog (mirrors real arrival: B's keystroke
        // echo lands while A's flood is still being drained into the lane).
        for i in 0..5 {
            tx.try_send(rpc_event("A")).unwrap();
            let _ = i;
        }
        tx.try_send(rpc_event("B")).unwrap();
        for _ in 0..5 {
            tx.try_send(rpc_event("A")).unwrap();
        }

        let first = rx.recv().await.unwrap();
        let drained = fair_drain_priority(first, &mut rx);

        assert_eq!(drained.len(), 11, "all 11 queued events must be returned, none dropped");

        // Find B's position in the fairly-drained output — it must appear
        // near the front (within the first "round" across 2 distinct panes),
        // not at position 6 where naive FIFO would place it.
        let b_pos = drained
            .iter()
            .position(|e| priority_pane_key(e).as_deref() == Some("block:B"))
            .expect("pane B's event must be present");
        assert!(
            b_pos <= 1,
            "pane B's frame should be at most 1 slot behind pane A in round-robin order, was at position {b_pos}"
        );
    }

    #[tokio::test]
    async fn test_fair_drain_priority_single_pane_preserves_order() {
        let (tx, mut rx) = tokio::sync::mpsc::channel(64);
        for _ in 0..4 {
            tx.try_send(rpc_event("A")).unwrap();
        }
        let first = rx.recv().await.unwrap();
        let drained = fair_drain_priority(first, &mut rx);
        assert_eq!(drained.len(), 4);
    }

    #[tokio::test]
    async fn test_fair_drain_priority_no_events_lost_across_many_panes() {
        let (tx, mut rx) = tokio::sync::mpsc::channel(256);
        let pane_ids = ["A", "B", "C", "D"];
        for (i, pane) in pane_ids.iter().cycle().take(40).enumerate() {
            let _ = i;
            tx.try_send(rpc_event(pane)).unwrap();
        }
        let first = rx.recv().await.unwrap();
        let drained = fair_drain_priority(first, &mut rx);
        assert_eq!(drained.len(), 40);
    }

    /// Regression test for the object-ordering bug caught in review: a
    /// `waveobj:update` for object X followed by a LATER
    /// `waveobj:batchedupdates` that also touches X must come out in the
    /// same relative order they arrived in. An earlier version of
    /// `priority_pane_key` keyed the individual update by its oref and left
    /// the unscoped batch in its own group, so round-robin turned
    /// `[update1, update2, batch]` into `[update1, batch, update2]` —
    /// resurrecting a deleted object on the frontend until reload (the
    /// frontend doesn't version-guard deletes). Object-model events must
    /// never be split by pane key, only `blockfile` (terminal output) events
    /// may be.
    #[tokio::test]
    async fn test_fair_drain_priority_preserves_object_update_order() {
        let (tx, mut rx) = tokio::sync::mpsc::channel(64);

        let update1 = raw_waveobj_update("tab:X");
        let update2 = raw_waveobj_update("tab:X");
        let batch_delete = batched_updates_event();

        tx.try_send(update1.clone()).unwrap();
        tx.try_send(update2.clone()).unwrap();
        tx.try_send(batch_delete.clone()).unwrap();

        let first = rx.recv().await.unwrap();
        let drained = fair_drain_priority(first, &mut rx);

        assert_eq!(drained, vec![update1, update2, batch_delete], "object-model events must stay in pure arrival order, never reordered by round-robin");
    }

    #[test]
    fn srv_info_event_carries_the_info_on_the_eventrecv_path() {
        let ev = super::srv_info_event(crate::srv_info::srv_info("1.2.3", "box"));
        assert_eq!(ev["eventtype"], "rpc");
        assert_eq!(ev["data"]["command"], "eventrecv");
        assert_eq!(ev["data"]["data"]["event"], "srvinfo");
        assert_eq!(ev["data"]["data"]["data"]["version"], "1.2.3");
    }
}