agentmux_srv/bootstrap/

delivery.rs

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

//! Split out of the single-file bootstrap module unchanged (SPEC_LARGE_FILE_MODULE_ANALYSIS_2026_09_30.md §4.3).

use super::*;

/// Deliver cron fires in process, through the server's shared inject path
/// (`server::reactive::deliver`) on the instance-key tier — identity M4c-3
/// (SPEC_AGENT_IDENTITY_CARRIED_NOT_DERIVED_2026_09_23.md §6.5.9). The fire
/// keeps every tier the route has (this instance, cross-instance,
/// cross-channel, LAN, cloud) and audits the job's creator UID here only.
///
/// Must run AFTER `build_app_state` and BEFORE `cron_scheduler.start()`, so
/// no fire takes the HTTP fallback.
pub fn install_cron_delivery(state: &AppState) {
    let st = state.clone();
    state.cron_scheduler.install_delivery(Arc::new(move |req, creator_uid| {
        let st = st.clone();
        Box::pin(async move {
            crate::server::reactive::deliver(
                &st,
                crate::server::ReactiveAuthVia::FullAuthKey,
                &creator_uid,
                req,
            )
            .await
        })
    }));
}

/// Where a message for an agent block goes.
pub(crate) enum AgentRoute {
    /// Taken on the controller's own channel — persistent stdin, ACP
    /// `session/prompt`, App Server — either now (`Delivered`) or queued for when
    /// the process can take it (`Deferred`). `Pty`: the controller is PTY-based,
    /// and the caller decides what that means for it.
    Sent(reactive::SenderDelivery),
    /// No structured channel takes it now: the caller starts a turn
    /// (`run_agent_turn`). `subprocess` is true for a `SubprocessController`,
    /// false for a persistent controller that is registered but not yet spawned.
    StartTurn { subprocess: bool },
}

/// The routing half of the reactive sender below, shared with the Swarm broadcast
/// (`server::app_api::fleet`) so a broadcast reaches ACP, App Server, running
/// persistent and not-yet-spawned agents exactly as an inter-agent message does.
///
/// Persistent + ACP + App Server keep their structured delivery; genuine PTY
/// controllers report `Pty`; a `SubprocessController` has no PTY and rejects raw
/// input, so it gets a turn instead (docs/reports/
/// REPORT_JEKT_DELIVERY_DROPS_SUBPROCESS_AGENTS_2026_09_02.md). An `Err` is a
/// structured controller that failed to take the message; it must not be retried
/// as keystrokes.
pub(crate) fn route_agent_message(block_id: &str, message: &str) -> Result<AgentRoute, String> {
    route_agent_message_from(block_id, message, None)
}

/// [`route_agent_message`], saying who the input is from: the Swarm broadcast
/// labels its delivery to a running persistent agent as the user's
/// (`deliver_agent_message_from`). `None` is exactly `route_agent_message`.
pub(crate) fn route_agent_message_from(
    block_id: &str,
    message: &str,
    origin: Option<backend::blockcontroller::health::TurnInput>,
) -> Result<AgentRoute, String> {
    let Some(ctrl) = backend::blockcontroller::get_controller(block_id) else {
        return Err(format!("no controller for block {block_id}"));
    };
    let is_subprocess = ctrl
        .as_any()
        .downcast_ref::<backend::blockcontroller::subprocess::SubprocessController>()
        .is_some();
    if is_subprocess {
        return Ok(AgentRoute::StartTurn { subprocess: true });
    }
    match backend::blockcontroller::deliver_agent_message_from(block_id, message, origin) {
        Ok(backend::blockcontroller::AgentDelivery::Structured) => {
            Ok(AgentRoute::Sent(reactive::SenderDelivery::Delivered))
        }
        Ok(backend::blockcontroller::AgentDelivery::StructuredDeferred) => {
            Ok(AgentRoute::Sent(reactive::SenderDelivery::Deferred))
        }
        Ok(backend::blockcontroller::AgentDelivery::Pty) => Ok(AgentRoute::Sent(reactive::SenderDelivery::Pty)),
        Err(e) => {
            // A persistent controller that is REGISTERED BUT NOT YET SPAWNED can't
            // be steered — `deliver_agent_message` writes to a live stdin and there
            // isn't one — but it can be STARTED. Controllers register lazily, so
            // after any srv restart every persistent agent sits in this state until
            // something is sent to it; without this, first contact fails with
            // "persistent process not running"
            // (docs/reports/REPORT_JEKT_DELIVERY_DROPS_UNSPAWNED_PERSISTENT_AGENTS_2026_09_03.md).
            //
            // Narrow on purpose: `needs_spawn()` is false while a spawn is already
            // in flight (that surfaces as the retryable "still starting up"), and
            // false for a live process whose delivery failed for another reason;
            // both keep returning the original error rather than start a second
            // turn.
            // An ACP agent whose process exited likewise: the turn path
            // starts it again (`AcpController::ensure_started`).
            let recoverable = ctrl
                .as_any()
                .downcast_ref::<backend::blockcontroller::persistent::PersistentSubprocessController>()
                .is_some_and(|p| p.needs_spawn())
                || ctrl
                    .as_any()
                    .downcast_ref::<backend::blockcontroller::acp::AcpController>()
                    .is_some_and(|a| !a.is_running());
            if !recoverable {
                return Err(e);
            }
            // Nothing was persisted or written, so falling through cannot
            // double-deliver.
            tracing::info!(
                block_id = %block_id,
                error = %e,
                "reactive delivery: persistent controller not yet spawned — starting a turn instead"
            );
            Ok(AgentRoute::StartTurn { subprocess: false })
        }
    }
}

/// Give the reactive handler a delivery route to `SubprocessController` agents.
///
/// `spawn_background_subsystems` installs a message sender that can only reach
/// the two controller kinds `deliver_agent_message` knows about — persistent
/// (stream-json stdin) and ACP (`session/prompt`). Everything else it reports as
/// [`AgentDelivery::Pty`], so the caller falls back to keystroke injection.
///
/// That fallback is wrong for a `SubprocessController`, which has no PTY and
/// rejects raw input outright ("subprocess controller does not accept raw input;
/// use AgentInputCommand"). The result was that EVERY inter-agent message to
/// such an agent was dropped: 6 of 10 providers as host agents (codex, gemini,
/// qwen, kimi, muxcode, antigravity) and every container agent of any provider,
/// since `agent_open.rs` forces `controller_type = "subprocess"` for container
/// mode. The swarm pane still listed them as present, because registration and
/// delivery are separate paths. See
/// `docs/reports/REPORT_JEKT_DELIVERY_DROPS_SUBPROCESS_AGENTS_2026_09_02.md`.
///
/// This re-installs the sender with the missing branch: a subprocess target gets
/// a real turn via `run_agent_turn` — the same thing `AgentInputCommand` does
/// when the operator types into the pane — instead of keystrokes it will refuse.
///
/// Must run AFTER `build_app_state`: unlike the early sender, the turn needs
/// `AppState` (stores, identity, broker, container manager) and is async.
pub fn install_agent_turn_delivery(state: &AppState) {
    let deps = crate::server::agent_handlers::AgentTurnDeps::from_state(state);
    state
        .reactive_handler
        .set_message_sender(Arc::new(move |block_id: &str, message: &str| {
            // Which controller kinds take a structured message and which need a turn
            // started is `route_agent_message`, shared with the Swarm broadcast so
            // the two cannot drift.
            let is_subprocess = match route_agent_message(block_id, message)? {
                AgentRoute::Sent(outcome) => return Ok(outcome),
                AgentRoute::StartTurn { subprocess } => subprocess,
            };

            // `MessageSender` is synchronous but starting a turn is not, so this
            // waits for the turn to START and reports what actually happened.
            //
            // It must NOT spawn-and-return-`Ok(true)` optimistically (codex P1 on
            // PR #2930). `success` is not advisory: `cloud_subscriber` treats a
            // successful delivery as final and releases the claim back to pending
            // for retry ONLY when `!delivery.success`
            // (`muxbus/cloud_subscriber.rs`). Reporting success before the
            // fallible work ran would therefore turn every transient failure —
            // stale block meta, an oauth spawn-gate refusal, Docker down,
            // `ensure_running` failing — into a WAN message that is marked
            // delivered, never retried, and permanently lost. An honest `Err` here
            // is what lets that message come back.
            //
            // `block_in_place` yields this worker thread back to the runtime for
            // the duration, so other tasks keep running. The cost is real and
            // worth naming: `inject_message_inner` calls this while holding the
            // reactive `Handler` mutex, so a slow start (first-time container
            // image pull is the pathological case — everything else is a block
            // read plus an already-running `docker exec`) serializes other
            // injections behind it. Bounded stalling beats silent loss.
            //
            // `run_agent_turn` returns once the turn has been STARTED, not once
            // the agent has answered, so this is not waiting on model latency.
            //
            // `TurnRegistration::Skip` is LOAD-BEARING, not a tidy-up. We are
            // running under that same non-reentrant `Mutex<Handler>`, so the
            // registration tail's `get_global_handler().register_agent(...)`
            // would try to re-lock a mutex this very thread already holds and
            // wedge the reactive handler process-wide (reagent P0 on PR #2930).
            // It is also simply redundant here: this block was resolved BY a
            // `resolve_target` lookup in that handler's registry, so the agent
            // is registered by construction. Do not "simplify" this to Register.
            let Ok(handle) = tokio::runtime::Handle::try_current() else {
                return Err(
                    "no tokio runtime available to start an agent turn".to_string(),
                );
            };
            // `block_in_place` panics on a current-thread runtime. Production is
            // multi-thread (`#[tokio::main]`); refusing loudly on the other flavor
            // is correct, since the alternative is the optimistic lie above.
            if handle.runtime_flavor() == tokio::runtime::RuntimeFlavor::CurrentThread {
                return Err(
                    "cannot start an agent turn from a current-thread runtime"
                        .to_string(),
                );
            }
            let deps = deps.clone();
            let block_id_owned = block_id.to_string();
            let message = message.to_string();
            tracing::info!(
                block_id = %block_id_owned,
                kind = if is_subprocess { "subprocess" } else { "persistent (not yet spawned)" },
                "reactive delivery: starting agent turn (no PTY fallback)"
            );
            let started = tokio::task::block_in_place(|| {
                handle.block_on(crate::server::agent_handlers::run_agent_turn(
                    &deps,
                    block_id_owned.clone(),
                    message,
                    None,
                    crate::server::agent_handlers::TurnRegistration::Skip,
                    // Reactive delivery: a jekt, cron, nudge, broadcast, loop.
                    crate::backend::blockcontroller::health::TurnOrigin::Automated,
                    Vec::new(),
                ))
            });
            match started {
                Ok(()) => Ok(reactive::SenderDelivery::Delivered),
                Err(e) => {
                    tracing::error!(
                        block_id = %block_id_owned,
                        error = %e,
                        "reactive delivery: agent turn failed to start"
                    );
                    Err(e)
                }
            }
        }));
}

/// Wires `blockcontroller::close_on_exit` (see that function's own doc
/// comment) up to the real close action, now that `AppState` exists.
/// Mirrors `install_agent_turn_delivery` just above — same "backend code
/// needs a capability only available once `AppState` is built" shape,
/// solved the same way (a process-wide callback set once at boot).
///
/// The callback itself is deliberately synchronous (`Fn(String, String)`,
/// not `async fn`) — `shell/lifecycle.rs`'s wait/cleanup task calls it from
/// a plain (non-async) decision point after its own cleanup, and trait
/// objects can't hold an `async fn` without extra boxing machinery this
/// doesn't need. `tokio::spawn` here is what actually runs the (necessarily
/// async) `sagas::delete_block::run` call; the callback returns
/// immediately, fire-and-forget, matching `close_on_exit`'s own contract.
///
/// See `docs/specs/SPEC_TERM_EXIT_RESPAWN_LOOP_2026_09_15.md` §10 for why
/// this exists: a shell pane's controller has no code path today that
/// closes it when the shell process exits, which (combined with §7-9's
/// fix for the OTHER half of that bug) leaves an exited pane sitting inert
/// — technically correct, but indistinguishable from "hung" to a user.
pub fn install_close_on_exit_handler(state: &AppState) {
    let state = state.clone();
    backend::blockcontroller::set_close_on_exit_handler(Arc::new(move |tab_id: String, block_id: String| {
        let state = state.clone();
        tokio::spawn(async move {
            if let Err(e) = crate::sagas::delete_block::run(&state, tab_id.clone(), block_id.clone()).await {
                // Best-effort, same posture as every other close-on-exit
                // failure mode: the shell already exited and its PTY is
                // already gone either way, so there is nothing left to roll
                // back — a failed close just leaves the pane sitting inert,
                // the same as if close-on-exit weren't enabled for it at
                // all, not a lost or corrupted state.
                tracing::warn!(
                    tab_id = %tab_id,
                    block_id = %block_id,
                    error = %e,
                    "close-on-exit: delete_block saga failed"
                );
            }
        });
    }));
}

#[cfg(test)]
mod route_agent_message_tests {
    use super::*;
    use backend::blockcontroller::{delete_controller, register_controller};

    #[test]
    fn a_block_with_no_controller_is_an_error() {
        let Err(e) = route_agent_message("route-test-no-such-block", "hi") else {
            panic!("an unregistered block has nowhere to go");
        };
        assert!(e.contains("no controller for block"), "{e}");
    }

    #[test]
    fn a_subprocess_agent_gets_a_turn_not_keystrokes() {
        let block_id = "route-test-subprocess";
        register_controller(
            block_id,
            Arc::new(backend::blockcontroller::subprocess::SubprocessController::new(
                "tab-route".to_string(),
                block_id.to_string(),
                None,
                None,
                None,
                None,
                None,
                Arc::from("test-boot"),
            )),
        );
        assert!(matches!(
            route_agent_message(block_id, "hi"),
            Ok(AgentRoute::StartTurn { subprocess: true })
        ));
        delete_controller(block_id);
    }

    #[test]
    fn an_app_server_agent_is_routed_to_its_own_channel_never_to_a_turn_or_keystrokes() {
        let block_id = "route-test-app-server";
        register_controller(
            block_id,
            Arc::new(backend::blockcontroller::app_server_controller::AppServerController::new(
                block_id.to_string(),
                None,
                None,
                None,
                None,
            )),
        );
        // No process yet, so its own `send_message` refuses: reaching that refusal
        // is the proof the structured route was taken (not StartTurn, not Pty).
        let Err(e) = route_agent_message(block_id, "hi") else {
            panic!("an App Server controller with no process cannot take a message");
        };
        assert!(e.contains("not initialized"), "{e}");
        delete_controller(block_id);
    }

    /// A message for an ACP agent whose process has exited is a turn to
    /// start (which starts the agent again), not a refusal that leaves the
    /// message stashed with nothing to run it.
    #[test]
    fn a_stopped_acp_agent_is_routed_to_a_turn_that_restarts_it() {
        let block_id = "route-test-acp-stopped";
        register_controller(
            block_id,
            Arc::new(backend::blockcontroller::acp::AcpController::new(
                "tab-route".to_string(),
                block_id.to_string(),
                None,
                None,
                None,
                None,
            )),
        );
        assert!(matches!(
            route_agent_message(block_id, "hi"),
            Ok(AgentRoute::StartTurn { subprocess: false })
        ));
        delete_controller(block_id);
    }
}