Struct SubprocessController 

pub struct SubprocessController {
    tab_id: String,
    block_id: String,
    run_lock: Arc<AtomicBool>,
    inner: Arc<Mutex<SubprocessControllerInner>>,
    broker: Option<Arc<Broker>>,
    event_bus: Option<Arc<EventBus>>,
    mstore: Option<Arc<Store>>,
    filestore: Option<Arc<FileStore>>,
    health_monitor: Arc<TurnActivityTracker>,
    self_ref: Mutex<Option<Weak<Self>>>,
    lease_store: Option<Arc<LeaseStore>>,
    boot_id: Arc<str>,
}

SubprocessController manages per-turn subprocess lifecycle for agent blocks.

Unlike ShellController which maintains a long-running PTY process, SubprocessController spawns a fresh process for each user turn. Multi-turn continuity comes from --resume <session-id>.

Fields

tab_id: String

block_id: String

run_lock: Arc<AtomicBool>

Prevents concurrent spawns.

inner: Arc<Mutex<SubprocessControllerInner>>

Protected inner state.

broker: Option<Arc<Broker>>

MPS broker for publishing events (blockfile, controllerstatus).

event_bus: Option<Arc<EventBus>>

Event bus for obj:update broadcasts.

mstore: Option<Arc<Store>>

AgentMux object store for block metadata persistence.

filestore: Option<Arc<FileStore>>

FileStore for write-through persistence of output lines (Phase 1.3).

health_monitor: Arc<TurnActivityTracker>

Per-block turn-activity tracker.

self_ref: Mutex<Option<Weak<Self>>>

Weak self-reference for queue drain. Set by set_self_ref after the controller is wrapped in Arc.

lease_store: Option<Arc<LeaseStore>>

Cross-process session-ownership lease store. None when the shared registry can’t be resolved (CI / unusual envs) — leasing degrades to a no-op in that case, same convention as shared_agent_registry() elsewhere. See docs/retro/RETRO_DEV_BUILD_SHARED_AGENT_SESSION_COLLISION_2026_07_29.md.

boot_id: Arc<str>

This process’s boot id — the lease owner id. See AppState::boot_id.

Implementations

impl SubprocessController

pub fn spawn_container_turn( &self, cm: ContainerManager, container_name: String, base_cmd: Vec<String>, config: SubprocessSpawnConfig, ) -> Result<(), String>

Spawn a container agent turn via Docker socket (P1a: no secrets in argv).

This is the secure alternative to spawn_turn for container agents. Instead of running docker exec -e KEY=VALUE ... as a CLI subprocess (which exposes secrets in process argv / /proc/<pid>/cmdline, CWE-214), this method calls ContainerManager::exec directly, passing env vars through CreateExecOptions.env (Docker socket). The exec I/O (file-backed prompt

• stdout NDJSON stream) drives the same state machine as spawn_turn: • appends --resume <sid> if a prior session_id is known • uploads the turn message into the container as a file and redirects the CLI’s stdin from it — see container_turn_exec for why the exec’s own stdin, argv, and env are all unusable for this • reads NDJSON from the output stream, publishing MPS blockfile events • captures session_id from the provider’s init event • transitions status running → done • drains the pending-message queue when the exec exits

base_cmd is [cli_command] + cli_args WITHOUT resume — this method appends --resume <sid> internally before starting the exec. The exec env is derived from THIS message’s config.env_vars (denylist applied here, per-turn) — not carried across queue drains — so a queued message runs with its own freshly-resolved auth/env, matching spawn_turn.

Takes cm and container_name by value (not reference) so the returned future is 'static — required for tokio::spawn in the queue-drain path.

fn publish_queued_spawn_error( block_id: &str, error: &str, broker: &Option<Arc<Broker>>, filestore: &Option<Arc<FileStore>>, mstore: &Option<Arc<Store>>, )

A queued message has already been removed from pending_messages when its next spawn is attempted. If provider argv validation rejects that spawn, persist a transcript error so the message does not disappear with only a server log. Mirrors the host queue-drain failure path.

fn publish_line( line: &str, block_id: &str, session_id_field: &str, inner: &Mutex<SubprocessControllerInner>, mstore: &Option<Arc<Store>>, event_bus: &Option<Arc<EventBus>>, broker: &Option<Arc<Broker>>, filestore: &Option<Arc<FileStore>>, stats: &mut SessionStatsAccumulator, global_output_zone: Option<&str>, last_result_frame: &mut Option<Value>, last_inband_error: &mut Option<Value>, )

Publish a single NDJSON line from container exec output: session-id capture, MPS blockfile event, and FileStore write-through. Used by spawn_container_turn’s output reader task.

impl SubprocessController

pub fn spawn_turn(&self, config: SubprocessSpawnConfig) -> Result<(), String>

Spawn a single turn of the agent CLI.

This is the core method — it spawns claude -p, writes the user message to stdin, reads NDJSON from stdout (publishing MPS events), and waits for exit.

If a session_id exists from a previous turn, --resume <sid> is appended to args.

impl SubprocessController

pub(super) fn record_captured_session_id_inner( inner: &Mutex<SubprocessControllerInner>, sid: &str, ) -> bool

Record an authoritative session id captured from the CLI’s stdout init/thread.started event. The CLI is the source of truth for which session is live, so this ALWAYS overwrites any prior value of inner.session_id — including values previously hydrated from config on a picker reattach (which may be stale by the time the CLI speaks).

Free-function form (taking &Arc<Mutex<…Inner>> instead of &self) so the spawn_turn stdout-reader tokio task can call it without holding an Arc<Self> reference. The &SubprocessController method below just delegates.

Returns true when the value changed (caller should broadcast the meta update + persist to block meta). Returns false when the new id matches the current one — common when the CLI emits the same session_id on every NDJSON frame within a single turn.

pub(super) (not pub(crate)): only called from the subprocess module tree (mod.rs/host_spawn/container_spawn/ tests) — matching the module-private SubprocessControllerInner parameter’s own reachability avoids a private_interfaces lint.

pub(crate) fn hydrate_session_id_from_config(&self, config_sid: Option<&str>)

Hydrate inner.session_id from a config-supplied id when the controller hasn’t seen a value yet.

Picker reattach path: a fresh SubprocessController is registered for the new block, so its inner.session_id is None. The frontend persisted the prior block’s session id into agent:sessionid meta, the websocket / app_api caller read it into SubprocessSpawnConfig::session_id, and this method copies it to inner so the spawn_turn args-builder appends --resume <sid> on the FIRST turn.

Hydration is best-effort, not authoritative. If inner.session_id is already Some we no-op (don’t overwrite a value already in place — could be a captured-from-stdout id from an earlier turn, or a prior hydration on the same reattach). Critically, the CLI’s stdout-emitted session id is authoritative and overwrites any prior value at capture time (see the stdout-reader block in spawn_turn). So if the hydrated value is stale, the FIRST turn passes the stale id via --resume (likely accepted as a no-op or rejected with a “no such session” error from the CLI), the CLI then emits its own session id in the init event, and inner.session_id is overwritten with that authoritative value for subsequent turns. Without the capture overwrite, a stale hydrated id would be re-used forever — that was the bug codex flagged on PR #1018 first cut.

Empty &str is treated as “no value” so the caller can use it unconditionally without filtering.

impl SubprocessController

pub fn new( tab_id: String, block_id: String, broker: Option<Arc<Broker>>, event_bus: Option<Arc<EventBus>>, mstore: Option<Arc<Store>>, filestore: Option<Arc<FileStore>>, registry: Option<Arc<Registry>>, boot_id: Arc<str>, ) -> Self

Create a new SubprocessController.

registry/boot_id back the cross-process session-ownership lease (see lease_store field doc). Passing registry: None (or a registry whose root can’t be opened as a LeaseStore, which is logged and degrades the same way) disables leasing for every turn this controller spawns — used by test call sites that don’t need a real registry.

pub fn set_self_ref(self: &Arc<Self>)

Store a weak self-reference so the process_waiter can drain queued messages by calling spawn_turn after the current turn exits. Must be called after wrapping in Arc.

fn try_lock_run(&self) -> bool

Try to acquire the run lock. Returns false if a turn is already in progress.

fn unlock_run(&self)

Release the run lock.

fn set_status(inner: &mut SubprocessControllerInner, status: &str)

Update process status and increment version (must hold inner lock).

fn build_status_snapshot( inner: &SubprocessControllerInner, block_id: &str, turn_active: bool, ) -> BlockControllerRuntimeStatus

Build a BlockControllerRuntimeStatus snapshot from inner state.

Factored out of 4+ verbatim-duplicated struct literals (one here, one in host_spawn’s process_waiter, two in container_spawn’s running/done/failure transitions) — an audit finding from the modularization pass. turn_active is passed explicitly because call sites disagree on its value: get_status_snapshot reads the live health-monitor flag, while every turn-completion / exec-failure site always publishes false (the turn is over by construction at that point).

fn get_status_snapshot(&self) -> BlockControllerRuntimeStatus

Get the runtime status (snapshot).

fn publish_status(&self)

Publish current controller status via the MPS broker.

fn emit_message_accepted(&self, config: &SubprocessSpawnConfig)

Emit agent-message-accepted for the given config, if it carries a message_id. Called from spawn_turn / spawn_container_turn; a queued message gets here too, when the process-waiter task’s drain calls that same spawn function for it. No-op if the config has no id, or the broker isn’t configured.

pub(super) fn user_record_sink(&self) -> UserRecordSink

A handle that writes the user’s message to the transcript — see UserRecordSink. Cloned out so the delivery paths (the stdin writer thread, the container exec task) can write it once the prompt has actually reached the CLI.

pub fn session_id(&self) -> Option<String>

Get the stored session ID (if any).

pub fn stop_subprocess(&self, force: bool) -> Result<(), String>

Stop the currently running subprocess.

Trait Implementations

impl Controller for SubprocessController

fn start( &self, _block_meta: HashMap<String, Value>, _rt_opts: Option<Value>, _force: bool, ) -> Result<(), String>

Start the controller. May spawn background tasks. force restarts even if already running.

fn stop(&self, _graceful: bool, new_status: &str) -> Result<(), String>

Stop the controller. graceful waits for process to exit; new_status is the target state.

fn get_runtime_status(&self) -> BlockControllerRuntimeStatus

Get the current runtime status.

fn send_input( &self, input: BlockInputUnion, _seq: Option<u64>, ) -> Result<(), String>

Send input (terminal data, signal, or resize) to the controller. seq is the per-TermViewModel monotonic counter; None means fire-and-forget (no ordering).

fn turn_tracker(&self) -> Option<Arc<TurnActivityTracker>>

This controller’s turn tracker, for the turn ledger’s bookkeeping from outside the controller (a held message’s join hint). None for controllers that track no turns.

fn controller_type(&self) -> &str

Get the controller type (e.g., “shell”, “cmd”).

fn block_id(&self) -> &str

Get the block ID.

fn as_any(&self) -> &dyn Any

Downcast support for concrete controller types.

fn stop_for_replace(&self, new_status: &str) -> Result<(), String>

Stop this controller because it’s being REPLACED by a new one for the same block (session restart / resync’s needs_replace path), NOT because the block itself is being closed. Must terminate this controller’s own CLI process so it doesn’t linger, but — unlike stop() — must NOT reach any declared-background descendant the process may have spawned (e.g. a task dev instance launched via run_in_background: true); those must survive the replace. See docs/specs/SPEC_BACKGROUND_TASK_TEARDOWN_SURVIVAL_2026_08_20.md.

fn shutdown( &self, _deadline: Instant, ) -> Pin<Box<dyn Future<Output = StopOutcome> + Send>>

Stop gracefully because the block is being CLOSED: end any turn in progress, ask the process to exit, force-kill at deadline. The returned future resolves once the process has actually exited (or there was none). Does NOT drop the block’s process tracker — the caller does that after this resolves, so nothing the agent started outlives it. See docs/specs/SPEC_AGENT_PANE_CLOSE_GRACEFUL_SHUTDOWN_2026_09_18.md §9.3.

fn agent_id(&self) -> Option<String>

This block’s own live, spawn-time-captured jekt/muxbus identity, if it has one — an independent source of truth for the recipient- identity check in ReactiveHandler::inject_message_inner, deliberately NOT derived from ReactiveHandler’s own name_to_blocks/agent_info maps (checking a registry against itself would be a tautology and catch nothing). Default None for controller types that aren’t jekt-addressable at all (e.g. plain terminals) — only ShellController and PersistentSubprocessController currently override this, mirroring the two paths that already resolve a jekt-registration identity at spawn time (resolve_agent_id_for_jekt, muxbus_agent_id_from_env).

fn turn_provenance(&self) -> Option<TurnProvenance>

What started this agent’s current turn (SPEC_AGENT_SELF_QUIT §6.3), if the controller tracks it. None = unknown, which never counts as the user — the safe answer for controllers that can’t tell (ACP, App Server, subprocess: both origins reach them through one method).

fn set_agent_id(&self, _id: Option<String>)

Refresh this block’s own captured jekt/muxbus identity (see agent_id’s doc comment). Called whenever ReactiveHandler::register_agent/register_agent_with_nonce (re-)registers THIS block’s block_id under a (possibly different) agent_id, so the two independently-written copies never drift apart (#2697: agent_id() was captured once at spawn and never refreshed, while the registry’s name bindings get re-keyed on every register_agent call — e.g. handle_reactive_register’s frontend-initiated HTTP path — causing a legitimately renamed or reconfigured agent’s own messages to be falsely rejected as an identity mismatch). Default no-op for controller types that don’t override agent_id either.

fn stable_agent_id(&self) -> Option<String>

This block’s STABLE jekt identity — AGENTMUX_AGENT_ID as captured once at spawn, never refreshed. Distinct from agent_id, which tracks the LIVE, renameable display name and is deliberately refreshed every turn (see that method’s doc comment on #2697). AGENTMUX_AGENT_ID is embedded verbatim in PR-body tags specifically because it does NOT change if the agent is later renamed — but ReactiveHandler’s registry only tracks one key per block, and every turn’s agent_id() refresh evicts whatever this block was previously registered under. ReactiveHandler::inject_message_inner’s recipient-identity check (#2695) uses this alongside agent_id() so a jekt addressed to the stable ID is validated correctly even after the primary registry key has moved on to the live display name. See INCIDENT_2026_09_09_JEKT_STABLE_ID_ALIAS.md. Default None, mirroring agent_id’s default — only PersistentSubprocessController currently overrides this.

fn stable_agent_uid(&self) -> Option<String>

Identity M2: this block’s UID (db_agents.id) as carried in its spawn env (AGENTMUX_AGENT_UID, set by build_persistent_spawn_env since M1a), captured once at spawn like [stable_agent_id]. Backs the registry’s UID confirmer: consulted only for targets that resolved by UID, and None — a registered-but-not-yet-spawned controller, or a block with no row — is “unverifiable”, not a mismatch. Default None; only PersistentSubprocessController overrides this.