agentmux_srv\backend\blockcontroller/
persistent.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! PersistentSubprocessController: manages agent CLI as a long-running process
5//! with bidirectional NDJSON streaming via stdin/stdout.
6//!
7//! Architecture:
8//!   A single CLI process is spawned on first message and kept alive for the
9//!   entire session. User messages are written as NDJSON lines to stdin without
10//!   closing it. This enables mid-turn input (redirecting the agent while it
11//!   is still processing).
12//!
13//! State machine:
14//!   INIT ─(first message)─> RUNNING ─(idle between turns)─> RUNNING
15//!   RUNNING ─(kill/stop)─> DONE
16//!   RUNNING ─(process crash)─> DONE (auto-restart possible via session_id)
17//!
18//! I/O model (3 async tasks per session):
19//! 1. stdin_writer: mpsc channel → process stdin (NDJSON lines)
20//! 2. stdout_reader: process stdout → .jsonl persistence + WPS blockfile events
21//! 3. process_waiter: wait for exit, update status
22
23use std::collections::HashMap;
24use std::sync::atomic::{AtomicU64, Ordering};
25use std::sync::{Arc, Mutex};
26
27use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
28use tokio::sync::mpsc;
29
30use super::{
31    BlockControllerRuntimeStatus, BlockInputUnion, Controller, STATUS_DONE, STATUS_INIT,
32    STATUS_RUNNING,
33};
34use super::core;
35use super::health::{classify_output_line, HealthMonitor};
36use crate::backend::eventbus::EventBus;
37use crate::backend::storage::filestore::FileStore;
38use crate::backend::storage::store::Store;
39use crate::backend::wps;
40
41/// WPS file subject name for persistent subprocess output.
42pub const PERSISTENT_OUTPUT_SUBJECT: &str = "output";
43
44pub const BLOCK_CONTROLLER_PERSISTENT: &str = "persistent";
45
46/// Resolve the muxbus address (the agent's display name) from a spawn env map.
47/// `AGENTMUX_AGENT_ID` (= `agent.name`, set at block creation) is canonical;
48/// `WAVEMUX_AGENT_ID` is the legacy fallback. Returns `None` — i.e. not
49/// muxbus-addressable — when neither is present (a non-agent persistent block).
50fn muxbus_agent_id_from_env(env: &HashMap<String, String>) -> Option<String> {
51    for key in ["AGENTMUX_AGENT_ID", "WAVEMUX_AGENT_ID"] {
52        if let Some(v) = env.get(key) {
53            let trimmed = v.trim();
54            if !trimmed.is_empty() {
55                return Some(trimmed.to_string());
56            }
57        }
58    }
59    None
60}
61
62#[cfg(test)]
63mod muxbus_registration_tests {
64    use super::muxbus_agent_id_from_env;
65    use std::collections::HashMap;
66
67    #[test]
68    fn resolves_agentmux_agent_id() {
69        let mut env = HashMap::new();
70        env.insert("AGENTMUX_AGENT_ID".to_string(), "Naki".to_string());
71        assert_eq!(muxbus_agent_id_from_env(&env), Some("Naki".to_string()));
72    }
73
74    #[test]
75    fn falls_back_to_legacy_wavemux_id() {
76        let mut env = HashMap::new();
77        env.insert("WAVEMUX_AGENT_ID".to_string(), "clamk".to_string());
78        assert_eq!(muxbus_agent_id_from_env(&env), Some("clamk".to_string()));
79    }
80
81    #[test]
82    fn prefers_agentmux_over_legacy() {
83        let mut env = HashMap::new();
84        env.insert("AGENTMUX_AGENT_ID".to_string(), "new".to_string());
85        env.insert("WAVEMUX_AGENT_ID".to_string(), "old".to_string());
86        assert_eq!(muxbus_agent_id_from_env(&env), Some("new".to_string()));
87    }
88
89    #[test]
90    fn none_when_absent_or_blank() {
91        let mut env: HashMap<String, String> = HashMap::new();
92        assert_eq!(muxbus_agent_id_from_env(&env), None);
93        env.insert("AGENTMUX_AGENT_ID".to_string(), "   ".to_string());
94        assert_eq!(muxbus_agent_id_from_env(&env), None);
95    }
96}
97
98/// Configuration for spawning the persistent process.
99#[derive(Debug, Clone)]
100pub struct PersistentSpawnConfig {
101    pub cli_command: String,
102    pub cli_args: Vec<String>,
103    pub working_dir: String,
104    pub env_vars: HashMap<String, String>,
105    pub session_id_field: String,
106    /// Resume flag for this provider (e.g. "--resume"), read from
107    /// `agent:resume_flag` meta. Empty = provider has no simple-flag resume.
108    /// Mirrors `SubprocessSpawnConfig::resume_flag` so a respawn (after a
109    /// runtime/model change or the picker reattach path) continues the same
110    /// conversation instead of starting fresh.
111    pub resume_flag: String,
112    /// Session id to hydrate `inner.session_id` with BEFORE spawning, when the
113    /// controller hasn't captured one yet (fresh controller after a forced
114    /// restart, or picker reattach). Read from `agent:sessionid` meta. With a
115    /// non-empty `resume_flag` this makes `--resume <sid>` land on the respawn.
116    pub session_id: String,
117    /// Echoed back as `agent-message-accepted` so the frontend can promote the
118    /// pending entry. Matches `CommandAgentInputData.message_id` on the AgentInput
119    /// command; absent for legacy callers.
120    pub message_id: Option<String>,
121}
122
123/// Inner state protected by mutex.
124struct PersistentInner {
125    proc_status: String,
126    proc_exit_code: i32,
127    status_version: i32,
128    session_id: Option<String>,
129    current_pid: Option<u32>,
130    /// Channel to send messages to the stdin writer task.
131    stdin_tx: Option<mpsc::Sender<String>>,
132    /// Handle to kill the process.
133    kill_tx: Option<tokio::sync::oneshot::Sender<bool>>,
134    /// AskUserQuestion `can_use_tool` control_requests awaiting a user answer:
135    /// `tool_use_id -> (request_id, questions JSON)`. Filled by the stdout
136    /// reader when the CLI sends a `can_use_tool` control_request for
137    /// AskUserQuestion; consumed by `answer_question` to build the matching
138    /// `control_response`. Spec: docs/specs/SPEC_AGENT_CONTROL_PROTOCOL_2026_06_15.md.
139    pending_questions: HashMap<String, (String, serde_json::Value)>,
140}
141
142/// PersistentSubprocessController keeps a long-running CLI process alive,
143/// sending user messages as NDJSON lines on stdin.
144pub struct PersistentSubprocessController {
145    #[allow(dead_code)]
146    tab_id: String,
147    block_id: String,
148    inner: Arc<Mutex<PersistentInner>>,
149    broker: Option<Arc<wps::Broker>>,
150    event_bus: Option<Arc<EventBus>>,
151    wstore: Option<Arc<Store>>,
152    /// FileStore for write-through persistence of output lines (Phase 1.3).
153    filestore: Option<Arc<FileStore>>,
154    health_monitor: Arc<HealthMonitor>,
155    /// Monotonic counter bumped for every stdout line (including control frames).
156    /// The AskUserQuestion dead-air fallback snapshots this *before* sending the
157    /// answer and re-checks after a short window; any increment means the CLI
158    /// produced output (assistant content OR a follow-up control_request), i.e.
159    /// the turn resumed. Counting *all* frames — not just `record_output`, which
160    /// the reader skips for control frames — avoids a spurious fallback when the
161    /// resumed turn's first activity is a tool-permission round-trip.
162    stdout_seq: Arc<AtomicU64>,
163}
164
165/// How long to wait after delivering an AskUserQuestion answer before assuming
166/// the turn did not resume and re-delivering the answer as a follow-up message.
167/// See `answer_question` and SPEC_ASK_USER_QUESTION_2026_06_15.md §10.1.
168const ANSWER_RESUME_FALLBACK_MS: u64 = 4000;
169
170/// Compose the directive follow-up message used by the AskUserQuestion dead-air
171/// fallback. `answers` maps each question's text to the selected label(s) or free
172/// text (the same object delivered in the control_response). The message is
173/// deliberately directive so the model resumes the task instead of treating it
174/// as a no-op (the "user sent an empty message" failure mode).
175fn build_answer_resume_message(answers: &serde_json::Value) -> String {
176    let mut out = String::from(
177        "[AgentMux] Your earlier question was answered, but the turn had already ended, \
178         so the answer is delivered here as a follow-up. Resume the task you were working \
179         on using this answer — do not wait for further input:\n",
180    );
181    match answers.as_object() {
182        Some(map) if !map.is_empty() => {
183            for (question, answer) in map {
184                let rendered = match answer {
185                    serde_json::Value::String(s) => s.clone(),
186                    serde_json::Value::Array(items) => items
187                        .iter()
188                        .filter_map(|v| v.as_str().map(str::to_string))
189                        .collect::<Vec<_>>()
190                        .join(", "),
191                    other => other.to_string(),
192                };
193                out.push_str(&format!("\n• {question}: {rendered}"));
194            }
195        }
196        _ => out.push_str(&format!("\nAnswer: {answers}")),
197    }
198    out
199}
200
201impl PersistentSubprocessController {
202    pub fn new(
203        tab_id: String,
204        block_id: String,
205        broker: Option<Arc<wps::Broker>>,
206        event_bus: Option<Arc<EventBus>>,
207        wstore: Option<Arc<Store>>,
208        filestore: Option<Arc<FileStore>>,
209    ) -> Self {
210        let health_monitor = Arc::new(HealthMonitor::new(
211            block_id.clone(),
212            broker.clone(),
213        ));
214        Self {
215            tab_id,
216            block_id,
217            inner: Arc::new(Mutex::new(PersistentInner {
218                proc_status: STATUS_INIT.to_string(),
219                proc_exit_code: 0,
220                status_version: 0,
221                session_id: None,
222                current_pid: None,
223                stdin_tx: None,
224                kill_tx: None,
225                pending_questions: HashMap::new(),
226            })),
227            broker,
228            event_bus,
229            wstore,
230            filestore,
231            health_monitor,
232            stdout_seq: Arc::new(AtomicU64::new(0)),
233        }
234    }
235
236    fn set_status(inner: &mut PersistentInner, status: &str) {
237        inner.proc_status = status.to_string();
238        inner.status_version += 1;
239    }
240
241    fn get_status_snapshot(&self) -> BlockControllerRuntimeStatus {
242        let inner = self.inner.lock().unwrap();
243        BlockControllerRuntimeStatus {
244            blockid: self.block_id.clone(),
245            version: inner.status_version,
246            shellprocstatus: inner.proc_status.clone(),
247            shellprocconnname: "local".to_string(),
248            shellprocexitcode: inner.proc_exit_code,
249            spawn_ts_ms: None,
250            is_agent_pane: true,
251        }
252    }
253
254    fn publish_status(&self) {
255        if let Some(ref broker) = self.broker {
256            let status = self.get_status_snapshot();
257            super::publish_controller_status(broker, &status);
258        }
259    }
260
261    fn is_running(&self) -> bool {
262        let inner = self.inner.lock().unwrap();
263        inner.stdin_tx.is_some()
264    }
265
266    /// Send a user message to the running CLI process.
267    /// If the process isn't spawned yet, spawns it first.
268    /// Emit `agent-message-accepted` for a given message_id, if set.
269    /// Mirrors the subprocess controller's `emit_message_accepted` — signals the
270    /// frontend to promote the pending entry from queued to in-document.
271    fn emit_message_accepted(&self, message_id: Option<&str>) {
272        let Some(id) = message_id else { return };
273        let Some(ref broker) = self.broker else { return };
274        let event = crate::backend::wps::WaveEvent {
275            event: crate::backend::wps::EVENT_AGENT_MESSAGE_ACCEPTED.to_string(),
276            scopes: vec![format!("block:{}", self.block_id)],
277            sender: String::new(),
278            persist: 0,
279            data: Some(serde_json::json!({
280                "block_id": self.block_id,
281                "message_id": id,
282            })),
283        };
284        broker.publish(event);
285        tracing::info!(
286            block_id = %self.block_id,
287            message_id = %id,
288            "emitted agent-message-accepted"
289        );
290    }
291
292    pub fn send_message(&self, message: String, config: PersistentSpawnConfig) -> Result<(), String> {
293        // Spawn process if not running
294        if !self.is_running() {
295            self.spawn_process(config.clone())?;
296        }
297
298        // Format as stream-json user message
299        let json_msg = serde_json::json!({
300            "type": "user",
301            "message": {
302                "role": "user",
303                "content": message
304            }
305        });
306        let json_str = json_msg.to_string();
307
308        // Silently persist the user message to the blockfile + global zone so
309        // `parseHistoryLines` can reconstruct `user_message` nodes on the next
310        // open. No WPS event is published here — the live-display is handled by
311        // the `agent-message-accepted` path (UUID node), avoiding a duplicate.
312        let global_zone = super::shell::resolve_global_output_zone(&self.wstore, &self.block_id);
313        let line_with_newline = format!("{json_str}\n");
314        super::shell::persist_to_blockfile_silent(
315            &self.block_id,
316            crate::backend::agent_session::OUTPUT_FILE,
317            line_with_newline.as_bytes(),
318            self.filestore.as_ref(),
319            global_zone.as_deref(),
320        );
321
322        let inner = self.inner.lock().unwrap();
323        let tx = inner.stdin_tx.as_ref()
324            .ok_or("persistent process not running after spawn")?;
325        tx.try_send(json_str)
326            .map_err(|e| format!("stdin send failed: {e}"))?;
327        drop(inner);
328        self.emit_message_accepted(config.message_id.as_deref());
329        Ok(())
330    }
331
332    /// Deliver a user message to the **already-running** persistent process,
333    /// without a spawn config. Unlike `send_message`, this never spawns — it errors
334    /// if the process is not running. Used for controller-aware muxbus/reactive
335    /// delivery (`deliver_agent_message`), where the agent is live (busy or idle)
336    /// and we have no `PersistentSpawnConfig` to hand. Writing on the live stdin lets
337    /// the message land mid-turn (steering) instead of waiting for idle.
338    /// Spec: docs/specs/SPEC_AGENT_CONTROL_PROTOCOL_2026_06_15.md §6 (Phase 3).
339    pub fn send_user_message(&self, message: String) -> Result<(), String> {
340        let json_msg = serde_json::json!({
341            "type": "user",
342            "message": {
343                "role": "user",
344                "content": message
345            }
346        });
347
348        let inner = self.inner.lock().unwrap();
349        let tx = inner
350            .stdin_tx
351            .as_ref()
352            .ok_or("persistent process not running")?;
353        tx.try_send(json_msg.to_string())
354            .map_err(|e| format!("stdin send failed: {e}"))
355    }
356
357    /// Answer a parked AskUserQuestion via the Agent SDK **control protocol**.
358    ///
359    /// The CLI asked us with a `can_use_tool` control_request (parked in
360    /// `pending_questions` by the stdout reader); we reply with a
361    /// `control_response` carrying `updatedInput.answers`. This is the ONLY
362    /// mechanism the CLI accepts — delivering a `tool_result` on stdin does NOT
363    /// work (the CLI auto-rejects AskUserQuestion within the turn). `answers` is
364    /// the JSON object mapping each question's text to the selected label(s) or
365    /// free-text. Process must already be running (agent is mid-turn, blocked on
366    /// this answer). Spec: docs/specs/SPEC_AGENT_CONTROL_PROTOCOL_2026_06_15.md §2.3.
367    pub fn answer_question(&self, tool_use_id: String, answers: serde_json::Value) -> Result<(), String> {
368        let (request_id, questions, tx) = {
369            let mut inner = self.inner.lock().unwrap();
370            let (rid, qs) = inner
371                .pending_questions
372                .remove(&tool_use_id)
373                .ok_or_else(|| format!("no pending AskUserQuestion for tool_use_id {tool_use_id}"))?;
374            let tx = inner
375                .stdin_tx
376                .as_ref()
377                .ok_or("persistent process not running (cannot deliver answer)")?
378                .clone();
379            (rid, qs, tx)
380        };
381
382        let control_response = serde_json::json!({
383            "type": "control_response",
384            "response": {
385                "subtype": "success",
386                "request_id": request_id,
387                "response": {
388                    "behavior": "allow",
389                    "updatedInput": { "questions": questions, "answers": answers.clone() },
390                    "toolUseID": tool_use_id,
391                }
392            }
393        });
394        // Snapshot stdout activity BEFORE sending the answer, so a fast resume
395        // that emits between the send and the snapshot can't be mistaken for
396        // "no activity" (codex review on #1536).
397        let stdout_seq = Arc::clone(&self.stdout_seq);
398        let before_seq = stdout_seq.load(Ordering::Relaxed);
399
400        tx.try_send(control_response.to_string())
401            .map_err(|e| format!("control_response send failed: {e}"))?;
402
403        // Dead-air safety net. The CLI *abandons* a pending AskUserQuestion
404        // tool_use if its turn already ended, silently dropping the
405        // control_response above — the model then sees an empty message and
406        // stalls (SPEC_ASK_USER_QUESTION_2026_06_15.md §9/§10.1; the dead-air
407        // report). If no stdout activity appears shortly after the answer, the
408        // turn did not resume, so re-deliver the answer as a normal follow-up
409        // user turn — the same resilience the one-shot controllers already use.
410        // Gated on stdout activity (every frame, incl. control frames), so it is
411        // mutually exclusive with a real resume and never double-delivers.
412        let inner = Arc::clone(&self.inner);
413        let block_id = self.block_id.clone();
414        let resume_msg = build_answer_resume_message(&answers);
415        tokio::spawn(async move {
416            tokio::time::sleep(std::time::Duration::from_millis(ANSWER_RESUME_FALLBACK_MS)).await;
417            // Any stdout frame since the snapshot means the turn resumed — nothing to do.
418            if stdout_seq.load(Ordering::Relaxed) != before_seq {
419                return;
420            }
421            let line = serde_json::json!({
422                "type": "user",
423                "message": { "role": "user", "content": resume_msg }
424            })
425            .to_string();
426            let stdin_tx = { inner.lock().unwrap().stdin_tx.clone() };
427            match stdin_tx {
428                Some(stdin_tx) if stdin_tx.try_send(line).is_ok() => {
429                    tracing::warn!(
430                        block_id = %block_id,
431                        tool_use_id = %tool_use_id,
432                        fallback_ms = ANSWER_RESUME_FALLBACK_MS,
433                        "AskUserQuestion answer did not resume the turn — re-delivered as a follow-up message (dead-air fallback)"
434                    );
435                }
436                Some(_) => tracing::warn!(
437                    block_id = %block_id,
438                    "AskUserQuestion dead-air fallback: stdin send failed"
439                ),
440                None => tracing::warn!(
441                    block_id = %block_id,
442                    "AskUserQuestion dead-air fallback skipped: process not running"
443                ),
444            }
445        });
446        Ok(())
447    }
448
449    /// Push a raw NDJSON line to the live stdin (used to emit control_responses
450    /// from the stdout-reader task, which only holds an `Arc<Mutex<Inner>>`).
451    fn push_stdin(inner: &Arc<Mutex<PersistentInner>>, line: String) {
452        let guard = inner.lock().unwrap();
453        if let Some(tx) = guard.stdin_tx.as_ref() {
454            let _ = tx.try_send(line);
455        }
456    }
457
458    /// Handle a control-protocol frame from the CLI's stdout. `control_request`
459    /// of subtype `can_use_tool`: AskUserQuestion is **parked** (the frontend
460    /// panel — rendered from the assistant stream — answers it via
461    /// `answer_question`); every other tool is **auto-allowed** to preserve the
462    /// current bypass/yolo UX (Phase 1; Phase 2 routes these to the decision
463    /// prompt, #551). `control_response` frames (replies to requests we initiate,
464    /// none today) are logged and dropped. These frames are NOT conversation
465    /// output and never reach the blockfile.
466    /// Spec: docs/specs/SPEC_AGENT_CONTROL_PROTOCOL_2026_06_15.md §4.2.
467    fn handle_control_frame(
468        kind: &str,
469        parsed: &serde_json::Value,
470        block_id: &str,
471        inner: &Arc<Mutex<PersistentInner>>,
472    ) {
473        if kind == "control_response" {
474            return;
475        }
476        // control_request
477        let req = match parsed.get("request") {
478            Some(r) => r,
479            None => return,
480        };
481        let subtype = req.get("subtype").and_then(|v| v.as_str()).unwrap_or("");
482        let request_id = parsed
483            .get("request_id")
484            .and_then(|v| v.as_str())
485            .unwrap_or("")
486            .to_string();
487
488        if subtype != "can_use_tool" {
489            tracing::info!(block_id = %block_id, subtype = %subtype, "persistent control_request: unhandled subtype, ignoring");
490            return;
491        }
492
493        let tool_name = req.get("tool_name").and_then(|v| v.as_str()).unwrap_or("");
494        let tool_use_id = req
495            .get("tool_use_id")
496            .and_then(|v| v.as_str())
497            .unwrap_or("")
498            .to_string();
499        let input = req.get("input").cloned().unwrap_or_else(|| serde_json::json!({}));
500
501        if tool_name == "AskUserQuestion" {
502            // Park; the frontend question panel will answer via answer_question().
503            let questions = input
504                .get("questions")
505                .cloned()
506                .unwrap_or_else(|| serde_json::json!([]));
507            {
508                let mut guard = inner.lock().unwrap();
509                guard
510                    .pending_questions
511                    .insert(tool_use_id.clone(), (request_id, questions));
512            }
513            tracing::info!(block_id = %block_id, tool_use_id = %tool_use_id, "AskUserQuestion parked; awaiting user answer");
514        } else {
515            // Auto-allow every other tool (preserve today's bypass UX).
516            let resp = serde_json::json!({
517                "type": "control_response",
518                "response": {
519                    "subtype": "success",
520                    "request_id": request_id,
521                    "response": { "behavior": "allow", "updatedInput": input }
522                }
523            });
524            Self::push_stdin(inner, resp.to_string());
525        }
526    }
527
528    /// Spawn the persistent CLI process.
529    fn spawn_process(&self, config: PersistentSpawnConfig) -> Result<(), String> {
530        // Build command — use make_cli_cmd to resolve .cmd wrappers to node on Windows
531        let mut cmd = crate::server::cli_handlers::make_cli_cmd(&config.cli_command);
532
533        // Hydrate the captured session id from the config when we don't have one
534        // yet (fresh controller after a forced restart — e.g. a /model change —
535        // or the picker reattach path). Mirrors SubprocessController::
536        // hydrate_session_id_from_config so the respawn resumes the same
537        // conversation instead of starting blank.
538        if !config.session_id.is_empty() {
539            let mut inner = self.inner.lock().unwrap();
540            if inner.session_id.is_none() {
541                inner.session_id = Some(config.session_id.clone());
542            }
543        }
544
545        // Append `--resume <sid>` when we have a session id and the provider
546        // supports simple-flag resume — same construction as
547        // SubprocessController::spawn_turn. This is what makes a model/effort
548        // change (which respawns the persistent CLI with new flags) preserve the
549        // conversation. cli_args carries the runtime flags (model/effort/perm)
550        // already rebuilt by the frontend (useAgentCommands buildRuntimeArgs).
551        let mut spawn_args = config.cli_args.clone();
552        {
553            let inner = self.inner.lock().unwrap();
554            if let Some(ref sid) = inner.session_id {
555                if !config.resume_flag.is_empty() {
556                    spawn_args.push(config.resume_flag.clone());
557                    spawn_args.push(sid.clone());
558                }
559            }
560        }
561        cmd.args(&spawn_args);
562
563        core::apply_working_dir(&mut cmd, &self.block_id, &config.working_dir, &config.env_vars);
564
565        // On Windows: suppress console-window allocation. The srv runs without a
566        // console of its own, so spawning the agent CLI without CREATE_NO_WINDOW
567        // makes Windows allocate a fresh console — which Windows 11's default-
568        // terminal handler renders as a NEW Windows Terminal window. One leaks per
569        // agent start / resume / respawn; a flapping or restart-heavy session
570        // accumulates dozens. stdio is piped here, so the console is never needed.
571        // See docs/retro/retro-windows-terminal-window-leak-2026-06-21.md.
572        // Matches acp.rs / subprocess.rs; sibling of shell.rs's PTY path.
573        #[cfg(windows)]
574        {
575            const CREATE_NO_WINDOW: u32 = 0x0800_0000;
576            cmd.creation_flags(CREATE_NO_WINDOW);
577        }
578
579        cmd.stdin(std::process::Stdio::piped());
580        cmd.stdout(std::process::Stdio::piped());
581        cmd.stderr(std::process::Stdio::piped());
582
583        let mut child = cmd.spawn().map_err(|e| {
584            tracing::error!(block_id = %self.block_id, error = %e, "persistent process spawn failed");
585            format!("failed to spawn persistent process: {e}")
586        })?;
587
588        let pid = child.id().unwrap_or(0);
589
590        // Notify health monitor that a turn is starting. This arms the Stalled
591        // (30 s) and Dead (120 s) thresholds so the frontend learns the agent
592        // is not responding rather than silently waiting forever.
593        self.health_monitor.set_active_turn(true);
594
595        tracing::info!(
596            block_id = %self.block_id,
597            pid = pid,
598            cmd = %config.cli_command,
599            args = ?spawn_args,
600            working_dir = %config.working_dir,
601            "persistent process spawned"
602        );
603
604        // Assign the persistent CLI to this block's process tracker.
605        // Matches `SubprocessController`'s identical path — both controller
606        // types share the same swarm-pane visibility story.
607        if pid != 0 {
608            if let Some(registry) = crate::backend::process_tracker::registry::global() {
609                let tracker = registry.ensure_tracker(&self.block_id);
610                if let Err(e) = tracker.assign_process(pid) {
611                    tracing::warn!(
612                        block_id = %self.block_id,
613                        pid = pid,
614                        err = %e,
615                        "[process-tracker] assign_process failed"
616                    );
617                }
618            }
619        }
620
621        let (kill_tx, kill_rx) = tokio::sync::oneshot::channel::<bool>();
622        let stdin = child.stdin.take()
623            .ok_or_else(|| format!("[persistent] stdin not captured for block {}", self.block_id))?;
624        let stdout = child.stdout.take()
625            .ok_or_else(|| format!("[persistent] stdout not captured for block {}", self.block_id))?;
626        let stderr = child.stderr.take();
627
628        // Drain stderr in background — log lines for debugging
629        if let Some(stderr_pipe) = stderr {
630            let block_id_stderr = self.block_id.clone();
631            tokio::spawn(async move {
632                let mut reader = BufReader::new(stderr_pipe).lines();
633                while let Ok(Some(line)) = reader.next_line().await {
634                    tracing::warn!(
635                        block_id = %block_id_stderr,
636                        line = %line,
637                        "persistent stderr"
638                    );
639                }
640            });
641        }
642
643        // Create stdin writer channel
644        let (msg_tx, mut msg_rx) = mpsc::channel::<String>(32);
645
646        {
647            let mut inner = self.inner.lock().unwrap();
648            inner.current_pid = Some(pid);
649            inner.kill_tx = Some(kill_tx);
650            inner.stdin_tx = Some(msg_tx);
651            Self::set_status(&mut inner, STATUS_RUNNING);
652        }
653        self.publish_status();
654
655        // Auto-register with the muxbus reactive handler so inter-agent
656        // messages reach this persistent (no-PTY) agent. The PTY shell
657        // controller (shell.rs) was the only prior auto-register path, so
658        // stream-json agents were in the directory but absent from the
659        // delivery registry — `inject_message` returned "agent not found"
660        // (issue #1470). Tier-1 delivery is routed through the controller-
661        // aware MessageSender (→ send_user_message), not PTY keystrokes.
662        // See SPEC_MUXBUS_AGENT_DISCOVERY_AND_PERSISTENT_DELIVERY_2026_06_16.
663        let agent_id_for_muxbus = muxbus_agent_id_from_env(&config.env_vars);
664        if let Some(ref agent_id) = agent_id_for_muxbus {
665            match crate::backend::reactive::get_global_handler()
666                .register_agent(agent_id, &self.block_id, Some(&self.tab_id))
667            {
668                Ok(()) => {
669                    tracing::info!(
670                        block_id = %self.block_id,
671                        agent_id = %agent_id,
672                        "muxbus: auto-registered persistent agent"
673                    );
674                    // Also write the cross-instance (Tier-2) file registry.
675                    if let Ok(local_url) = std::env::var("AGENTMUX_LOCAL_URL") {
676                        let data_dir = crate::backend::base::get_wave_data_dir();
677                        crate::backend::reactive::registry::write(
678                            &data_dir,
679                            agent_id,
680                            &local_url,
681                            &self.block_id,
682                        );
683                    }
684                }
685                Err(e) => tracing::warn!(
686                    block_id = %self.block_id,
687                    agent_id = %agent_id,
688                    error = %e,
689                    "muxbus: persistent auto-register failed"
690                ),
691            }
692        }
693
694        // Record active pid for crash recovery (Phase 4.2). If the server
695        // dies while this subprocess is running, scan_orphans() will find
696        // the stale pid on next boot and flag the session as interrupted.
697        if let Some(ref wstore) = self.wstore {
698            super::session_recovery::mark_active_pid(wstore, &self.block_id, pid);
699        }
700
701        // Spawn stdin writer task
702        tokio::spawn(async move {
703            let mut stdin = stdin;
704            while let Some(msg) = msg_rx.recv().await {
705                if let Err(e) = stdin.write_all(msg.as_bytes()).await {
706                    tracing::warn!("persistent stdin write error: {}", e);
707                    break;
708                }
709                if let Err(e) = stdin.write_all(b"\n").await {
710                    tracing::warn!("persistent stdin newline error: {}", e);
711                    break;
712                }
713                if let Err(e) = stdin.flush().await {
714                    tracing::warn!("persistent stdin flush error: {}", e);
715                    break;
716                }
717            }
718            // Channel closed or write error → stdin drops → process gets EOF
719            drop(stdin);
720        });
721
722        // Spawn stdout reader task
723        let block_id_read = self.block_id.clone();
724        let broker_read = self.broker.clone();
725        let inner_read = Arc::clone(&self.inner);
726        let wstore_read = self.wstore.clone();
727        let event_bus_read = self.event_bus.clone();
728        let filestore_read = self.filestore.clone();
729        let health_read = Arc::clone(&self.health_monitor);
730        let stdout_seq_read = Arc::clone(&self.stdout_seq);
731        let session_id_field = config.session_id_field.clone();
732        // Resolve the agent's GLOBAL transcript zone (`agent:<defId>:current`)
733        // once, from the block's `agentId` meta, so every `output` line is also
734        // mirrored to the cross-channel store. `None` for non-agent blocks.
735        let global_output_zone =
736            super::shell::resolve_global_output_zone(&self.wstore, &self.block_id);
737
738        tokio::spawn(async move {
739            let reader = BufReader::new(stdout);
740            let mut lines = reader.lines();
741            let mut stats = super::session_stats::SessionStatsAccumulator::new(block_id_read.clone());
742
743            // NOTE: OSC window-title extraction is NOT done here.
744            // PersistentSubprocessController uses piped stdout with stream-json
745            // NDJSON protocol. Claude Code sets window titles via process.title
746            // (SetConsoleTitle on Windows; argv[0] on Unix), which does NOT
747            // produce OSC escape sequences in the piped stdout stream. Inserting
748            // OSC bytes into stream-json stdout would corrupt the JSON protocol.
749            // block:activity events for agent panes are instead published by
750            // the terminalSequence hooks path — see spec §2.5 and the future
751            // SPEC_AGENT_HOOKS_TERMINAL_SEQUENCE spec.
752
753            while let Ok(Some(line)) = lines.next_line().await {
754                if line.trim().is_empty() {
755                    continue;
756                }
757                // Bump the activity counter for EVERY non-empty stdout line —
758                // including control frames (which `continue` below before
759                // `record_output`) — so the AskUserQuestion dead-air fallback can
760                // tell whether the turn resumed. See `answer_question`.
761                stdout_seq_read.fetch_add(1, Ordering::Relaxed);
762
763                // Track session metadata (debounced 1 s)
764                stats.record_line(line.len(), &wstore_read);
765
766                // Parse JSON for health monitoring and session ID capture
767                if let Ok(parsed) = serde_json::from_str::<serde_json::Value>(&line) {
768                    // Control-protocol frames (can_use_tool / AskUserQuestion) are
769                    // NOT conversation output — handle them and skip the blockfile
770                    // so the frontend stream never sees them.
771                    // Spec: docs/specs/SPEC_AGENT_CONTROL_PROTOCOL_2026_06_15.md.
772                    if let Some(kind) = parsed.get("type").and_then(|v| v.as_str()) {
773                        if kind == "control_request" || kind == "control_response" {
774                            Self::handle_control_frame(kind, &parsed, &block_id_read, &inner_read);
775                            continue;
776                        }
777                    }
778                    let (meaningful, _error) = classify_output_line(&parsed);
779                    health_read.record_output(meaningful);
780                    if let Some(sid) = parsed.get(&session_id_field).and_then(|v| v.as_str()) {
781                        let sid_string = sid.to_string();
782                        let already_captured = inner_read.lock().unwrap().session_id.is_some();
783                        if !already_captured {
784                            tracing::info!(
785                                block_id = %block_id_read,
786                                session_id = %sid_string,
787                                "persistent session ID captured"
788                            );
789                            inner_read.lock().unwrap().session_id = Some(sid_string.clone());
790                            core::persist_session_id(&block_id_read, &sid_string, &wstore_read, &event_bus_read);
791                        }
792                    }
793                }
794
795                // Publish line as WPS blockfile event and write-through to FileStore
796                // for persistent history (Phase 1.3).
797                tracing::info!(
798                    block_id = %block_id_read,
799                    line_len = line.len(),
800                    "persistent stdout → blockfile"
801                );
802                let line_with_newline = format!("{}\n", line);
803                if let Some(ref broker) = broker_read {
804                    super::shell::handle_append_block_file(
805                        broker,
806                        &block_id_read,
807                        PERSISTENT_OUTPUT_SUBJECT,
808                        line_with_newline.as_bytes(),
809                        filestore_read.as_ref(),
810                        global_output_zone.as_deref(),
811                    );
812                } else {
813                    tracing::warn!(block_id = %block_id_read, "persistent stdout: no broker available");
814                }
815            }
816
817            tracing::info!(block_id = %block_id_read, "persistent stdout reader finished");
818        });
819
820        // Spawn health watchdog — checks every 5 s while turn is active.
821        // Emits `agenthealth` WPS events when the process stalls (30 s) or
822        // dies (120 s) without producing meaningful output, giving the
823        // frontend enough signal to show a "not responding" warning.
824        core::spawn_health_watchdog(&self.health_monitor);
825
826        // Spawn process waiter task
827        let block_id_wait = self.block_id.clone();
828        let inner_wait = Arc::clone(&self.inner);
829        let broker_wait = self.broker.clone();
830        let wstore_wait = self.wstore.clone();
831        let health_wait = Arc::clone(&self.health_monitor);
832        // Captured so the waiter can deregister this agent from muxbus on exit.
833        let agent_id_wait = agent_id_for_muxbus.clone();
834
835        tokio::spawn(async move {
836            tokio::select! {
837                status = child.wait() => {
838                    let exit_code = status.map(|s| s.code().unwrap_or(-1)).unwrap_or(-1);
839                    tracing::info!(
840                        block_id = %block_id_wait,
841                        exit_code = exit_code,
842                        "persistent process exited"
843                    );
844
845                    // Notify health monitor so Stalled/Dead watchdog stops.
846                    health_wait.set_exited(exit_code);
847
848                    let mut inner = inner_wait.lock().unwrap();
849                    inner.proc_exit_code = exit_code;
850                    inner.current_pid = None;
851                    inner.stdin_tx = None;
852                    inner.kill_tx = None;
853                    Self::set_status(&mut inner, STATUS_DONE);
854                    drop(inner);
855
856                    // Deregister from muxbus so later sends fall through to the
857                    // lower tiers instead of resolving to a dead block. Mirrors
858                    // the shell controller's exit path.
859                    crate::backend::reactive::get_global_handler()
860                        .unregister_block(&block_id_wait);
861                    if let Some(ref agent_id) = agent_id_wait {
862                        let data_dir = crate::backend::base::get_wave_data_dir();
863                        crate::backend::reactive::registry::remove(&data_dir, agent_id);
864                        if let Some(sub) = crate::muxbus::cloud_subscriber::get_global_subscriber() {
865                            sub.remove_agent(agent_id);
866                        }
867                    }
868
869                    // Clear active pid — clean exit, no recovery needed.
870                    if let Some(ref wstore) = wstore_wait {
871                        super::session_recovery::clear_active_pid(wstore, &block_id_wait);
872                    }
873
874                    // Publish status
875                    if let Some(ref broker) = broker_wait {
876                        let status = BlockControllerRuntimeStatus {
877                            blockid: block_id_wait.clone(),
878                            version: 0,
879                            shellprocstatus: STATUS_DONE.to_string(),
880                            shellprocconnname: "local".to_string(),
881                            shellprocexitcode: exit_code,
882                            spawn_ts_ms: None,
883                            is_agent_pane: true,
884                        };
885                        super::publish_controller_status(broker, &status);
886                    }
887                }
888                Ok(force) = kill_rx => {
889                    tracing::info!(
890                        block_id = %block_id_wait,
891                        force = force,
892                        "persistent process kill requested"
893                    );
894                    if force {
895                        let _ = child.kill().await;
896                    } else {
897                        // Graceful: drop stdin to send EOF, then wait briefly
898                        {
899                            let mut inner = inner_wait.lock().unwrap();
900                            inner.stdin_tx = None; // drops the sender → stdin writer exits → stdin closes
901                        }
902                        tokio::select! {
903                            _ = child.wait() => {}
904                            _ = tokio::time::sleep(std::time::Duration::from_secs(5)) => {
905                                let _ = child.kill().await;
906                            }
907                        }
908                    }
909
910                    health_wait.set_exited(-1);
911
912                    let mut inner = inner_wait.lock().unwrap();
913                    inner.proc_exit_code = -1;
914                    inner.current_pid = None;
915                    inner.stdin_tx = None;
916                    inner.kill_tx = None;
917                    Self::set_status(&mut inner, STATUS_DONE);
918                    drop(inner);
919
920                    // Deregister from muxbus (see the clean-exit arm above).
921                    crate::backend::reactive::get_global_handler()
922                        .unregister_block(&block_id_wait);
923                    if let Some(ref agent_id) = agent_id_wait {
924                        let data_dir = crate::backend::base::get_wave_data_dir();
925                        crate::backend::reactive::registry::remove(&data_dir, agent_id);
926                        if let Some(sub) = crate::muxbus::cloud_subscriber::get_global_subscriber() {
927                            sub.remove_agent(agent_id);
928                        }
929                    }
930
931                    // Clear active pid — user-initiated stop, no recovery needed.
932                    if let Some(ref wstore) = wstore_wait {
933                        super::session_recovery::clear_active_pid(wstore, &block_id_wait);
934                    }
935                }
936            }
937        });
938
939        Ok(())
940    }
941
942    pub fn stop_process(&self, force: bool) -> Result<(), String> {
943        let kill_tx = {
944            let mut inner = self.inner.lock().unwrap();
945            inner.kill_tx.take()
946        };
947        match kill_tx {
948            Some(tx) => {
949                let _ = tx.send(force);
950                Ok(())
951            }
952            None => Ok(()),
953        }
954    }
955
956    pub fn session_id(&self) -> Option<String> {
957        self.inner.lock().unwrap().session_id.clone()
958    }
959}
960
961impl Controller for PersistentSubprocessController {
962    fn start(
963        &self,
964        _block_meta: super::super::obj::MetaMapType,
965        _rt_opts: Option<serde_json::Value>,
966        _force: bool,
967    ) -> Result<(), String> {
968        tracing::info!(
969            block_id = %self.block_id,
970            "persistent controller registered (spawns on first message)"
971        );
972        Ok(())
973    }
974
975    fn stop(&self, _graceful: bool, new_status: &str) -> Result<(), String> {
976        self.stop_process(true)?;
977        let mut inner = self.inner.lock().unwrap();
978        if inner.proc_status != new_status {
979            Self::set_status(&mut inner, new_status);
980        }
981        Ok(())
982    }
983
984    fn get_runtime_status(&self) -> BlockControllerRuntimeStatus {
985        self.get_status_snapshot()
986    }
987
988    fn send_input(&self, input: BlockInputUnion, _seq: Option<u64>) -> Result<(), String> {
989        // Persistent controllers have no PTY and don't take raw keystrokes —
990        // user messages go through send_message(). But the agent-pane Stop
991        // button / Esc delivers an *interrupt* as a signal via
992        // `ControllerInputCommand({signame:"SIGINT"})` (see useAgentCommands
993        // `stopAgent`). Without handling it here, stopping a persistent (e.g.
994        // Claude stream-json) agent failed with "does not accept raw input".
995        // Route the interrupt to the same kill path `stop()` uses, mirroring
996        // SubprocessController. The session_id is retained, so the next message
997        // resumes the conversation.
998        if let Some(sig) = input.sig_name.as_deref() {
999            if sig == "SIGINT" || sig == "SIGTERM" {
1000                tracing::info!(
1001                    block_id = %self.block_id,
1002                    sig = %sig,
1003                    "persistent controller: received signal, stopping current process"
1004                );
1005                return self.stop_process(true);
1006            }
1007            return Err(format!(
1008                "persistent controller: unsupported signal {sig} (only SIGINT/SIGTERM)"
1009            ));
1010        }
1011        // Raw keystrokes are genuinely unsupported — user messages go through
1012        // send_message(), not the PTY input channel.
1013        if input.input_data.is_some() {
1014            return Err(
1015                "persistent controller does not accept raw input; use send_message()".to_string(),
1016            );
1017        }
1018        // Term resize / other benign input types: accepted no-op. A persistent
1019        // controller has no PTY, so there is nothing to resize — but the agent
1020        // pane's `usePtyWidth` hook sends a `termsize` on every running turn
1021        // (it can't tell a PTY-backed controller from a PTY-less one). Returning
1022        // an error here surfaced a spurious "resize to N cols failed" warning in
1023        // the agent pane's activity log. Mirror SubprocessController, which
1024        // already no-ops termsize. See AGENT_PANE_PTY_RESIZE_RACE_2026_06_16.md.
1025        Ok(())
1026    }
1027
1028    fn controller_type(&self) -> &str {
1029        BLOCK_CONTROLLER_PERSISTENT
1030    }
1031
1032    fn block_id(&self) -> &str {
1033        &self.block_id
1034    }
1035
1036    fn as_any(&self) -> &dyn std::any::Any {
1037        self
1038    }
1039}
1040
1041#[cfg(test)]
1042mod send_input_tests {
1043    use super::*;
1044    use crate::backend::obj::TermSize;
1045
1046    fn controller() -> PersistentSubprocessController {
1047        PersistentSubprocessController::new(
1048            "tab".to_string(),
1049            "block".to_string(),
1050            None,
1051            None,
1052            None,
1053            None,
1054        )
1055    }
1056
1057    // A persistent controller has no PTY, but the agent pane's usePtyWidth hook
1058    // sends a termsize resize on every running turn. It must be accepted as a
1059    // no-op, not rejected — otherwise the pane logs a spurious "resize to N cols
1060    // failed" warning. See AGENT_PANE_PTY_RESIZE_RACE_2026_06_16.md.
1061    #[test]
1062    fn termsize_resize_is_accepted_noop() {
1063        let c = controller();
1064        let res = c.send_input(BlockInputUnion::resize(TermSize { rows: 25, cols: 117 }), None);
1065        assert!(res.is_ok(), "termsize resize should be a no-op Ok, got {res:?}");
1066    }
1067
1068    // The AskUserQuestion dead-air fallback re-delivers the answer as a directive
1069    // follow-up message; the rendering must surface each Q/A so the model can
1070    // resume with the decision in context. See SPEC_ASK_USER_QUESTION §10.1.
1071    #[test]
1072    fn answer_resume_message_renders_qa_pairs() {
1073        let answers = serde_json::json!({
1074            "Pick a color": "blue",
1075            "Pick toppings": ["cheese", "olives"],
1076        });
1077        let msg = build_answer_resume_message(&answers);
1078        assert!(msg.contains("Resume the task"), "must be directive: {msg}");
1079        assert!(msg.contains("Pick a color: blue"), "string answer: {msg}");
1080        assert!(
1081            msg.contains("Pick toppings: cheese, olives"),
1082            "multi-select joins labels: {msg}"
1083        );
1084    }
1085
1086    #[test]
1087    fn answer_resume_message_handles_non_object() {
1088        let msg = build_answer_resume_message(&serde_json::json!("just text"));
1089        assert!(msg.contains("Resume the task"), "still directive: {msg}");
1090        assert!(msg.contains("Answer: "), "non-object falls back: {msg}");
1091    }
1092
1093    // Raw keystrokes are genuinely unsupported on a persistent controller —
1094    // user messages go through send_message(), so they must still be rejected.
1095    #[test]
1096    fn raw_input_is_still_rejected() {
1097        let c = controller();
1098        let err = c
1099            .send_input(BlockInputUnion::data(b"ls\n".to_vec()), None)
1100            .unwrap_err();
1101        assert!(
1102            err.contains("does not accept raw input"),
1103            "raw input should be rejected, got {err:?}"
1104        );
1105    }
1106}
agentmux_srv\backend\blockcontroller/persistent.rs

agentmux_srv\backend\blockcontroller/
persistent.rs
