agentmux_srv/backend/blockcontroller/persistent/stdout_reader.rs
1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! The persistent CLI's stdout reader task, started by `spawn_process`
5//! (`spawn.rs`): control frames, session-id capture, resume-state events,
6//! failure classification, the task feed and the output append.
7
8use super::*;
9
10/// What the stdout reader task captures from `spawn_process`.
11pub(super) struct StdoutReaderCtx {
12 pub(super) stdout: tokio::process::ChildStdout,
13 pub(super) block_id_read: String,
14 pub(super) broker_read: Option<Arc<mps::Broker>>,
15 pub(super) inner_read: Arc<Mutex<PersistentInner>>,
16 pub(super) mstore_read: Option<Arc<Store>>,
17 pub(super) event_bus_read: Option<Arc<EventBus>>,
18 pub(super) filestore_read: Option<Arc<FileStore>>,
19 pub(super) health_read: Arc<TurnActivityTracker>,
20 pub(super) self_ref_read: Option<std::sync::Weak<PersistentSubprocessController>>,
21 pub(super) stdout_seq_read: Arc<AtomicU64>,
22 pub(super) session_id_field: String,
23 pub(super) my_generation_read: u64,
24 /// This spawn's process-wide registration nonce: whose MCP report this
25 /// reader records (`backend::mcp_observed`).
26 pub(super) mcp_owner_read: u64,
27 pub(super) global_output_zone: Option<String>,
28 pub(super) record_fence: Option<crate::backend::agent_admission::RecordFence>,
29 pub(super) segment_read: super::segments::SegmentRef,
30 pub(super) relocated_copy_read: Option<crate::backend::continuity_relocate::Relocated>,
31 pub(super) forked_from_read: Option<String>,
32}
33
34impl PersistentSubprocessController {
35 /// Body of the stdout reader task; `spawn_process` keeps its JoinHandle
36 /// for the process waiter.
37 pub(super) async fn run_stdout_reader(ctx: StdoutReaderCtx) {
38 let StdoutReaderCtx {
39 stdout,
40 block_id_read,
41 broker_read,
42 inner_read,
43 mstore_read,
44 event_bus_read,
45 filestore_read,
46 health_read,
47 self_ref_read,
48 stdout_seq_read,
49 session_id_field,
50 my_generation_read,
51 mcp_owner_read,
52 global_output_zone,
53 record_fence,
54 segment_read,
55 mut relocated_copy_read,
56 forked_from_read,
57 } = ctx;
58 let reader = BufReader::new(stdout);
59 let mut lines = reader.lines();
60 let mut stats = super::super::session_stats::SessionStatsAccumulator::new(block_id_read.clone());
61 // The CLI's own `task_started`/`task_updated`/`task_notification`
62 // lines drive the background-task registry, including tasks a
63 // subagent owns (see `background_task_feed.rs`).
64 let mut task_feed = crate::backend::background_task_feed::TaskFeed::default();
65
66 // NOTE: OSC window-title extraction is NOT done here.
67 // PersistentSubprocessController uses piped stdout with stream-json
68 // NDJSON protocol. Claude Code sets window titles via process.title
69 // (SetConsoleTitle on Windows; argv[0] on Unix), which does NOT
70 // produce OSC escape sequences in the piped stdout stream. Inserting
71 // OSC bytes into stream-json stdout would corrupt the JSON protocol.
72 // block:activity events for agent panes are instead published by
73 // the terminalSequence hooks path — see spec §2.5 and the future
74 // SPEC_AGENT_HOOKS_TERMINAL_SEQUENCE spec.
75
76 while let Ok(Some(line)) = lines.next_line().await {
77 if line.trim().is_empty() {
78 continue;
79 }
80 // Bump the activity counter for EVERY non-empty stdout line —
81 // including control frames handled via `continue` below — so
82 // the AskUserQuestion dead-air fallback can tell whether the
83 // turn resumed. See `answer_question`.
84 stdout_seq_read.fetch_add(1, Ordering::Relaxed);
85
86 // Track session metadata (debounced 1 s)
87 stats.record_line(line.len(), &mstore_read);
88
89 // Set (instead of persisted immediately) when this line turns
90 // out to be a terminal `result`/`is_error:true` event arriving
91 // while a stale-`--resume` retry could still be confirmed for
92 // this exact attempt — see `persistent_resume::ResumeState`'s
93 // `held_error_line`. `false` for every other line, matching
94 // today's behavior exactly.
95 let mut hold_back_for_resume_retry = false;
96 // Parse JSON for control-frame handling, turn-active tracking,
97 // and session ID capture
98 if let Ok(parsed) = serde_json::from_str::<serde_json::Value>(&line) {
99 // Control-protocol frames (can_use_tool / AskUserQuestion) are
100 // NOT conversation output — handle them and skip the blockfile
101 // so the frontend stream never sees them.
102 // Spec: docs/specs/SPEC_AGENT_CONTROL_PROTOCOL_2026_06_15.md.
103 if let Some(kind) = parsed.get("type").and_then(|v| v.as_str()) {
104 if kind == "control_request" || kind == "control_response" {
105 // The answer to our `get_context_usage`: where this
106 // process auto-compacts. Ignored from a replaced
107 // process; re-published only when it changed.
108 if let Some(usage) =
109 crate::backend::agent_context_usage::context_usage_from_control_response(&parsed)
110 {
111 let changed = {
112 let mut g = inner_read.lock().unwrap();
113 if g.spawn_generation == my_generation_read && g.context_usage.as_ref() != Some(&usage) {
114 g.context_usage = Some(usage.clone());
115 true
116 } else {
117 false
118 }
119 };
120 if let (true, Some(broker)) = (changed, broker_read.as_ref()) {
121 crate::backend::agent_context_usage::publish_agent_context_usage(
122 broker,
123 &block_id_read,
124 &usage,
125 );
126 }
127 }
128 // The answer to our `get_settings`: what the CLI
129 // really runs. Ignored from a replaced process.
130 if let Some(eff) = crate::backend::agent_runtime::effective_from_control_response(&parsed) {
131 let changed = {
132 let mut g = inner_read.lock().unwrap();
133 if g.spawn_generation == my_generation_read && g.spawn_runtime.is_some() {
134 let changed = g.effective_runtime.as_ref() != Some(&eff);
135 g.effective_runtime = Some(eff);
136 changed
137 } else {
138 false
139 }
140 };
141 if let (true, Some(broker)) = (changed, broker_read.as_ref()) {
142 super::status::publish_runtime_event(&inner_read, broker, &block_id_read);
143 }
144 }
145 // The answer to a self-heal `mcp_status`/`mcp_reconnect`.
146 if let Some(ctrl) = self_ref_read.as_ref().and_then(|w| w.upgrade()) {
147 ctrl.observe_mcp_frame(&parsed, my_generation_read);
148 }
149 Self::handle_control_frame(kind, &parsed, &block_id_read, &inner_read);
150 // OS notification: the agent is now blocked on the
151 // user — known here even with no pane mounted
152 // (SPEC_OS_NOTIFICATIONS_SYSTEM_2026_09_24 Phase 5).
153 if let (Some(broker), Some(asked)) = (
154 broker_read.as_ref(),
155 crate::backend::notify::sources::ask_user_question(&parsed),
156 ) {
157 if let Some(r) = crate::backend::notify::router::get(broker) {
158 r.input_waiting_nonblocking(&block_id_read, asked.text, asked.count);
159 }
160 }
161 continue;
162 }
163 }
164 if let Some(store) = mstore_read.as_deref() {
165 let now_ms = agentmux_common::time::now_ms();
166 task_feed.apply(
167 store,
168 broker_read.as_deref(),
169 &block_id_read,
170 &parsed,
171 now_ms,
172 );
173 }
174 // A pass the CLI starts by itself: a finished background
175 // task's `task_notification`, then a fresh `system/init` with
176 // no input from srv (turn-model spec §4.2). Recorded so the
177 // agent reads busy for that pass, everywhere `turn_active`
178 // is read. Ignored from a replaced process.
179 if parsed.get("type").and_then(|v| v.as_str()) == Some("system") {
180 let subtype = parsed.get("subtype").and_then(|v| v.as_str());
181 let current = || inner_read.lock().unwrap().spawn_generation == my_generation_read;
182 if subtype == Some("task_notification") && current() {
183 health_read.note_cli_task_notification(parsed.get("summary").and_then(|v| v.as_str()));
184 } else if subtype == Some("init") && current() && health_read.mark_turn_active_from_cli() {
185 if let Some(ctrl) = self_ref_read.as_ref().and_then(|w| w.upgrade()) {
186 ctrl.cli_started_pass();
187 }
188 }
189 }
190 // Whether its `agentmux` MCP server connected, from the same
191 // frame: one that didn't is repaired (`backend::mcp_health`).
192 if parsed.get("subtype").and_then(|v| v.as_str()) == Some("init") {
193 if let Some(ctrl) = self_ref_read.as_ref().and_then(|w| w.upgrade()) {
194 ctrl.observe_mcp_frame(&parsed, my_generation_read);
195 }
196 // Every server it reports, for Connectors (`backend::mcp_observed`);
197 // a replaced process's late init never writes over a newer one's.
198 if crate::backend::mcp_observed::record_init(&block_id_read, mcp_owner_read, &parsed) {
199 crate::backend::mcp_observed::publish(broker_read.as_deref(), &block_id_read);
200 }
201 }
202 // The CLI version this agent now runs, from Claude's own
203 // `system/init` frame: every spawn path passes here, and
204 // it is the version that actually runs, so a self-update
205 // shows too. A change from the agent's last run puts a
206 // notice in the pane (`backend::cli_notice`).
207 if parsed.get("subtype").and_then(|v| v.as_str()) == Some("init") {
208 if let (Some(version), Some(store), Some(broker), Some(filestore)) = (
209 parsed.get("claude_code_version").and_then(|v| v.as_str()),
210 mstore_read.clone(),
211 broker_read.clone(),
212 filestore_read.clone(),
213 ) {
214 let (block_id, version) = (block_id_read.clone(), version.to_string());
215 tokio::task::spawn_blocking(move || {
216 crate::backend::cli_notice::observe_and_notify(
217 &broker, &filestore, &store, &block_id, None, "claude", &version,
218 );
219 });
220 }
221 }
222 // This block's CLI compacted: only its pane may send the
223 // hidden memory reinjection for that boundary.
224 if parsed.get("subtype").and_then(|v| v.as_str()) == Some("compact_boundary") {
225 if let Some(uuid) = parsed.get("uuid").and_then(|v| v.as_str()) {
226 crate::server::memory_delivery_handlers::record_compaction_boundary(&block_id_read, uuid);
227 }
228 }
229 let is_result_frame =
230 parsed.get("type").and_then(|v| v.as_str()) == Some("result");
231 // Claude's turn-ending marker. Persistent mode never exits
232 // between turns, so this is the only place `turn_active`
233 // can go back to false without waiting for process exit —
234 // see `send_message`'s matching `set_active_turn(true)`.
235 if is_result_frame {
236 // A runtime-config change (model/effort/permission)
237 // arrived mid-turn and was deferred rather than
238 // killing this turn — see
239 // `request_restart_when_idle`. The turn is over now,
240 // so stop the process. Its kill arm eager-resumes the
241 // replacement at once (`respawn_after_config_restart`),
242 // which rebuilds `cli_args` from block meta, so the
243 // new flags take effect then. It used to wait for "the
244 // next message", which a jekt never is.
245 // Not `poison_resume` and not a user Stop — the
246 // session id is retained, so the respawn `--resume`s
247 // the same conversation.
248 // `set_active_turn(false)` happens under `inner` so it
249 // serializes with `request_restart_when_idle`'s own
250 // check — see that method for the interleaving this
251 // closes. `restart_pending` is committed in the SAME
252 // acquisition so no send can slip into `DeliverDirect`
253 // between the decision and the kill.
254 // Whether the deferred restart may apply: see
255 // `turn_boundary_locked`.
256 let boundary = PersistentSubprocessController::turn_boundary_locked(
257 &mut inner_read.lock().unwrap(),
258 &health_read,
259 &block_id_read,
260 my_generation_read,
261 Some(crate::backend::blockcontroller::health::PassStats::from_result_frame(&parsed)),
262 );
263 // `None`: this reader's process has been replaced. Its
264 // `result` ends nothing the current process is doing,
265 // so it must not idle or restart that process (codex
266 // P1 on #3562).
267 let boundary_is_current = boundary.is_some();
268 let deferred_restart = boundary.unwrap_or(false);
269 // The model can change under a running process (the CLI
270 // falls back to another model when one is overloaded), so
271 // ask again at every turn boundary. Cheap, and not for a
272 // process that is about to be restarted anyway.
273 if boundary_is_current && !deferred_restart {
274 let ask = inner_read.lock().unwrap().settings_readback;
275 if ask {
276 Self::push_stdin(&inner_read, crate::backend::agent_runtime::settings_request_line());
277 // The threshold moves with the model (a fallback
278 // model, a /model switch applied in place).
279 Self::push_stdin(
280 &inner_read,
281 crate::backend::agent_context_usage::context_usage_request_line(),
282 );
283 }
284 }
285 if deferred_restart {
286 tracing::info!(
287 block_id = %block_id_read,
288 "turn ended — applying the deferred runtime-config restart"
289 );
290 // Marked as a restart, not a plain stop, so the
291 // kill arm brings the replacement up itself: the
292 // next message is often a jekt, and automated
293 // delivery never spawns.
294 if let Some(ctrl) = self_ref_read.as_ref().and_then(|w| w.upgrade()) {
295 ctrl.stop_for_config_restart();
296 }
297 }
298 // Publish the flip so the Swarm view's live
299 // ControllerStatus subscription reflects "turn
300 // ended" immediately instead of only on the next
301 // unrelated status change (or process exit) — see
302 // send_message's matching publish_status() call for
303 // the turn-start side of this pair.
304 // The turn ended: any "needs your input" for it is
305 // moot (answered, or abandoned by a new message / the
306 // turn finishing). srv never drains
307 // `pending_questions` on its own, so this boundary is
308 // the reliable "resolved" signal.
309 if let (Some(broker), true) = (broker_read.as_ref(), boundary_is_current) {
310 if let Some(r) = crate::backend::notify::router::get(broker) {
311 r.resolve_nonblocking(&block_id_read, crate::backend::notify::policy::Family::Input);
312 }
313 }
314 if let (Some(broker), true) = (broker_read.as_ref(), boundary_is_current) {
315 let status = {
316 let locked = inner_read.lock().unwrap();
317 super::super::agent_runtime_status(
318 &block_id_read,
319 locked.status_version,
320 &locked.proc_status,
321 locked.proc_exit_code,
322 false,
323 )
324 };
325 super::super::publish_controller_status(broker, &status);
326 }
327 // SPEC_SUBAGENT_LIVE_RECONCILIATION_AND_RETIRE_2026_07_20
328 // Phase A: reconcile any subagent still Active for this
329 // block the instant its turn ends, not just at the next
330 // pane reopen — closes SPEC_SUBAGENT_LIFECYCLE_
331 // RECONCILIATION_2026_07_12.md's Open Question 1. A
332 // subagent runs inside the parent's own CLI process (a
333 // Task-tool call is synchronous within the parent's
334 // turn), so this is the same "turn ended" signal
335 // scan_session_subagents already reconciles against at
336 // reopen — just fired live instead of waiting.
337 // `global()` is `None` in tests that don't call
338 // `subagent_watcher::set_global` — a safe no-op, same
339 // pattern `process_tracker::registry` already uses.
340 //
341 // On a BLOCKING thread, not this one. Reconciliation
342 // used to be a pure in-memory status flip under a
343 // mutex (O(1)), but as of the completion-detection
344 // fix (#3007) it reads the parent transcript from
345 // disk — routinely tens of MB, plus per-line JSON
346 // parsing — to decide whether each member actually
347 // returned. Doing that inline would block a tokio
348 // worker thread for the whole read, and this fires on
349 // EVERY turn-end, so several blocks finishing at once
350 // could stall the runtime (reagent P1 on #3007).
351 //
352 // Fire-and-forget: the result is a status correction
353 // broadcast to whoever's listening, not something this
354 // stdout-reader task consumes, so there is nothing to
355 // await. The backfill call site (`scan_session_
356 // subagents`) needs no equivalent — it already runs in
357 // a blocking context that walks up to 200 files.
358 let session_id_snapshot = inner_read.lock().unwrap().session_id.clone();
359 if let Some(sid) = session_id_snapshot {
360 if let Some(watcher) = subagent_watcher::global() {
361 let block_id = block_id_read.clone();
362 tokio::task::spawn_blocking(move || {
363 watcher.reconcile_stale_subagents(&block_id, &sid);
364 });
365 }
366 }
367 }
368 // A turn WE interrupted to close the pane ends in an
369 // `is_error: true` result (`error_during_execution`).
370 // That is the requested stop, not a failure: treat it
371 // as an ordinary end of turn — no failure banner, no
372 // stale-`--resume` retry tracking (pane-close spec §9.4).
373 let closing_this_generation = is_result_frame
374 && inner_read.lock().unwrap().shutdown_generation == Some(my_generation_read);
375 let is_error_result = is_result_frame
376 && !closing_this_generation
377 && parsed.get("is_error").and_then(|v| v.as_bool()) == Some(true);
378 // reagentx P0 on PR #2371: the real CLI's stream-json
379 // protocol embeds `session_id_field` on EVERY event,
380 // including the terminal `result` — so the doomed
381 // attempt's own `is_error:true` line ALSO carries the
382 // (stale) sid it was given. Calling
383 // `try_capture_session_id` for THIS exact line would
384 // resolve this generation's resume tracking (a
385 // non-poisoned confirmation does — codex P1's fix,
386 // needed for the genuinely-successful-resume case)
387 // BEFORE the `ErrorResultLine` event below ever runs,
388 // reproducing the exact bubble this PR exists to
389 // suppress AND preventing PR #2360's own retry from
390 // ever being confirmed. An error frame's echoed sid is
391 // never genuine progress (the turn failed) — skip
392 // session-id capture entirely for this exact frame;
393 // every other frame type (system/init, a successful
394 // result) still captures normally.
395 // Set when the capture_effects loop below classifies and
396 // persists a flushed OLDER turn's failure this tick — the
397 // clear-on-success step further down must not immediately
398 // wipe out state it just recorded (SPEC_PERSISTENT_
399 // CONTROLLER_FAILURE_CLASSIFICATION_2026_08_04.md).
400 let mut flushed_failure_this_tick = false;
401 if !is_error_result {
402 if let Some(sid) = parsed.get(&session_id_field).and_then(|v| v.as_str()) {
403 let sid_string = sid.to_string();
404 // reagentx P0 on PR #2371: a genuinely
405 // successful terminal result (is_result_frame
406 // && !is_error, which is guaranteed true here
407 // since is_error_result already excluded the
408 // failing case) is the ONLY unambiguous proof
409 // that resuming THIS sid actually worked —
410 // any earlier frame (e.g. a "system"/init
411 // frame) echoes the same attempted sid
412 // regardless of whether the resume goes on to
413 // fail, per persistent_resume::update's own
414 // handling of `SessionCaptured`.
415 let is_confirmed_success = is_result_frame;
416 // See PersistentInner::try_capture_session_id — refuses
417 // to (re-)adopt an id the stderr reader (`stderr_reader.rs`) already
418 // confirmed unreachable, whichever task wins the race.
419 let (should_capture, capture_effects, holds_current) = {
420 let mut inner = inner_read.lock().unwrap();
421 let (adopted, effects) = inner.try_capture_session_id(&sid_string, my_generation_read, is_confirmed_success);
422 (adopted, effects, inner.holds_current_session(&sid_string, my_generation_read))
423 };
424 if should_capture {
425 tracing::info!(
426 block_id = %block_id_read,
427 session_id = %sid_string,
428 "persistent session ID captured"
429 );
430 core::persist_session_id(&block_id_read, &sid_string, &mstore_read, &event_bus_read);
431 }
432 let kept_relocated_id = super::segments::kept_relocated_id(
433 should_capture,
434 is_confirmed_success,
435 holds_current,
436 relocated_copy_read.is_some(),
437 forked_from_read.as_deref(),
438 &sid_string,
439 );
440 if should_capture {
441 super::segments::record_segment_session(&segment_read, &sid_string);
442 // The fork has its own session now; the copy it
443 // read from goes (spec §4.3 (3), I4).
444 if let Some(copy) = relocated_copy_read.take() {
445 super::segments::settle_relocated_copy(&block_id_read, ©, forked_from_read.as_deref(), &sid_string);
446 }
447 } else if kept_relocated_id {
448 // Kept under the attempted id, the copy is the live
449 // session: recorded only while it is still this
450 // spawn's, never a replacement's at the same path.
451 if let Some(copy) = relocated_copy_read.take() {
452 if super::segments::settle_relocated_copy(&block_id_read, ©, forked_from_read.as_deref(), &sid_string) {
453 super::segments::record_segment_session(&segment_read, &sid_string);
454 }
455 }
456 }
457 // reagentx P0 on PR #2373: resolving tracking
458 // here can legitimately flush a held-back
459 // error line from an earlier turn on this
460 // same still-alive generation — execute it,
461 // same as every other `ResumeEffect` call
462 // site in this module. `SessionCaptured` can
463 // also now resolve tracking outright and
464 // produce an `EmitSessionOutcome` (see
465 // `persistent_resume::update`'s own handling,
466 // SPEC_AGENT_PANE_HISTORY_ALIGNMENT_2026_08_05.md
467 // §2.1) — handled explicitly below; anything
468 // else falls to the catch-all, kept exhaustive
469 // rather than assuming the effect set never
470 // grows again.
471 for effect in capture_effects {
472 match effect {
473 persistent_resume::ResumeEffect::EmitSessionOutcome {
474 outcome,
475 attempted_sid,
476 actual_sid,
477 } => {
478 // A fork reports a new id by design: that is
479 // the resume succeeding, not a fresh start.
480 let outcome = super::segments::forked_outcome(
481 outcome,
482 forked_from_read.as_deref(),
483 &attempted_sid,
484 actual_sid.as_deref(),
485 );
486 // The retry (if any led here) is now resolved one
487 // way or the other — clear "Reconnecting…". A no-op
488 // publish (still fine) when this outcome came from a
489 // plain first-time resume that was never retried.
490 publish_resume_retry_status(&broker_read, &block_id_read, "resolved");
491 // A recovery scan can turn an
492 // already-disclosed resume failure
493 // into a genuine resume — retract the
494 // banner so it can't contradict the
495 // `resumed` divider appended just
496 // below. See
497 // `session_recovery::clear_resume_failed`.
498 if matches!(outcome, persistent_resume::SessionOutcome::Resumed) {
499 if let Some(ref store) = mstore_read {
500 super::super::session_recovery::clear_resume_failed(
501 store,
502 &event_bus_read,
503 &block_id_read,
504 );
505 }
506 }
507 if let Some(ref broker) = broker_read {
508 let line = session_outcome_line(outcome, attempted_sid, actual_sid);
509 super::super::shell::handle_append_block_file(
510 broker,
511 &block_id_read,
512 PERSISTENT_OUTPUT_SUBJECT,
513 line.as_bytes(),
514 filestore_read.as_ref(),
515 crate::backend::agent_admission::fenced_zone(global_output_zone.as_deref(), &record_fence),
516 );
517 }
518 }
519 persistent_resume::ResumeEffect::FlushErrorLine(line) => {
520 if let Some(ref broker) = broker_read {
521 super::super::shell::handle_append_block_file(
522 broker,
523 &block_id_read,
524 PERSISTENT_OUTPUT_SUBJECT,
525 line.as_bytes(),
526 filestore_read.as_ref(),
527 crate::backend::agent_admission::fenced_zone(global_output_zone.as_deref(), &record_fence),
528 );
529 }
530 // reagentx P2 on PR #2421: this flushes
531 // an earlier held-back turn's error,
532 // finally confirmed final now that
533 // session-id tracking resolved — must
534 // get the same classify/persist/publish
535 // treatment as the identical
536 // FlushErrorLine handled at the
537 // process-exit arm, or this turn's
538 // failure silently loses its recovery
539 // banner. No exit code exists for this
540 // now-superseded turn.
541 if surface_error_line(
542 &block_id_read,
543 None,
544 &line,
545 broker_read.as_deref(),
546 &mstore_read,
547 &event_bus_read,
548 ) {
549 flushed_failure_this_tick = true;
550 }
551 }
552 other => {
553 tracing::warn!(
554 block_id = %block_id_read,
555 effect = ?other,
556 "unexpected ResumeEffect from a SessionCaptured event"
557 );
558 }
559 }
560 }
561 }
562 }
563 // Issue #2368: this generation's resume tracking
564 // (`persistent_resume::ResumeState`) decides whether
565 // this line is still a retry candidate — if it has
566 // already resolved (a session id was captured above or
567 // on an earlier line, or this generation never
568 // attempted an untrusted `--resume`), `update()`
569 // returns a `PersistImmediately` effect and today's
570 // immediate-persist behavior is unchanged; otherwise
571 // the line is held back pending the retry decision at
572 // process exit.
573 //
574 // reagentx P2 on PR #2373: hold-back is now decided
575 // from the RESULTING state, not `effects.is_empty()`
576 // — a still-tracking result can now ALSO carry a
577 // `PersistImmediately` effect for a SUPERSEDED
578 // held-back line from an earlier turn on this same
579 // generation (a second `is_error:true` while tracking
580 // is undecided means the first was a separate,
581 // already-settled turn's error). That effect must be
582 // flushed here explicitly; when NOT still tracking,
583 // any effect returned is THIS exact line's own
584 // `PersistImmediately`, already handled by the
585 // unchanged fallthrough below — executing it here too
586 // would double-persist it.
587 if is_error_result {
588 let (effects, still_tracking) = {
589 let mut inner = inner_read.lock().unwrap();
590 let effects = inner.apply_resume_event(persistent_resume::ResumeEvent::ErrorResultLine {
591 generation: my_generation_read,
592 line: format!("{}\n", line),
593 });
594 let still_tracking = matches!(
595 &inner.resume,
596 persistent_resume::ResumeState::AwaitingOutcome { generation, .. }
597 if *generation == my_generation_read
598 ) || matches!(
599 &inner.resume,
600 persistent_resume::ResumeState::ConfirmedRetry { generation, .. }
601 if *generation == my_generation_read
602 );
603 (effects, still_tracking)
604 };
605 hold_back_for_resume_retry = still_tracking;
606 if still_tracking {
607 for effect in effects {
608 match effect {
609 persistent_resume::ResumeEffect::PersistImmediately(old_line)
610 | persistent_resume::ResumeEffect::FlushErrorLine(old_line) => {
611 if let Some(ref broker) = broker_read {
612 super::super::shell::handle_append_block_file(
613 broker,
614 &block_id_read,
615 PERSISTENT_OUTPUT_SUBJECT,
616 old_line.as_bytes(),
617 filestore_read.as_ref(),
618 crate::backend::agent_admission::fenced_zone(global_output_zone.as_deref(), &record_fence),
619 );
620 }
621 // reagentx P1 on PR #2421 (round 2):
622 // this is a SEPARATE, already-settled
623 // older turn's error, superseded by
624 // the current still-tracking line —
625 // now confirmed final, same as the
626 // other three FlushErrorLine/
627 // PersistImmediately call sites this
628 // PR wired up. `is_error_result` is
629 // true for the rest of this tick, so
630 // this can never collide with the
631 // clear-on-success step below.
632 if surface_error_line(
633 &block_id_read,
634 None,
635 &old_line,
636 broker_read.as_deref(),
637 &mstore_read,
638 &event_bus_read,
639 ) {
640 flushed_failure_this_tick = true;
641 }
642 }
643 other => {
644 tracing::warn!(
645 block_id = %block_id_read,
646 effect = ?other,
647 "unexpected ResumeEffect from a still-tracking ErrorResultLine event"
648 );
649 }
650 }
651 }
652 }
653 }
654 // Classify + surface a genuine, non-retried error result
655 // (429/overloaded/auth/etc.) to the pane's failure-recovery
656 // UI — SPEC_PERSISTENT_CONTROLLER_FAILURE_CLASSIFICATION.
657 // Gated on `!hold_back_for_resume_retry`: when the stale-
658 // `--resume` machinery above is still tracking this exact
659 // error as a live retry candidate, it must stay invisible
660 // to the user (per its own "must never reach the user"
661 // invariant at `process_waiter.rs`'s ProcessExited/FireRetry arm) —
662 // classify() only runs once this error is confirmed final.
663 if is_error_result && !hold_back_for_resume_retry {
664 let failure = crate::agents::failure::classify(None, None, "", Some(&parsed));
665 surface_failure(&block_id_read, &failure, broker_read.as_deref(), &mstore_read, &event_bus_read);
666 } else if is_result_frame && !is_error_result && !flushed_failure_this_tick {
667 // reagentx P1 on PR #2421: unlike host_spawn.rs, this
668 // controller never exits between turns, so nothing
669 // else ever clears a previously recorded failure —
670 // once one rate-limit/overloaded error was persisted,
671 // the pane's onMount seed logic kept re-showing that
672 // stale banner on every future reload, even after
673 // many later successful turns. A genuinely successful
674 // terminal result on this still-alive process is the
675 // signal that it's stale. persist_last_failure is a
676 // no-op when there's nothing to clear, so this is
677 // cheap on the (overwhelmingly common) already-clear
678 // path. Skipped when the capture_effects loop above
679 // just persisted a freshly-flushed OLDER failure this
680 // same tick — that state must survive, not be
681 // immediately wiped by this frame's own success.
682 core::persist_last_failure(&block_id_read, None, &mstore_read, &event_bus_read);
683 // A completed turn is when the agent's running
684 // summary may be due for an update. Background, and
685 // a no-op unless enough happened since the last one
686 // (SPEC_DURABLE_CONVERSATION_MEMORY_2026_09_23.md §4.4).
687 crate::backend::continuity_state::after_successful_turn(
688 mstore_read.clone(),
689 block_id_read.clone(),
690 );
691 }
692 }
693
694 if hold_back_for_resume_retry {
695 continue;
696 }
697
698 // Publish line as MPS blockfile event and write-through to FileStore
699 // for persistent history (Phase 1.3).
700 //
701 // debug, not info: fires on EVERY output line a streaming agent
702 // produces — the single largest contributor (~27%) to an
703 // unrotated 406 MB launcher-log mirror on a real machine
704 // (SPEC_WIN10_PAGEFILE_OOM_CRASH_2026_06_29 P1). Default
705 // production filter is info, so this is now suppressed unless
706 // RUST_LOG=debug is set.
707 tracing::debug!(
708 block_id = %block_id_read,
709 line_len = line.len(),
710 "persistent stdout → blockfile"
711 );
712 if let Some(ref broker) = broker_read {
713 super::super::shell::append_output_line(
714 broker,
715 &block_id_read,
716 &line,
717 filestore_read.as_ref(),
718 crate::backend::agent_admission::fenced_zone(global_output_zone.as_deref(), &record_fence),
719 );
720 } else {
721 tracing::warn!(block_id = %block_id_read, "persistent stdout: no broker available");
722 }
723 }
724
725 tracing::info!(block_id = %block_id_read, "persistent stdout reader finished");
726 }
727}