agentmux_srv/backend/blockcontroller/persistent/

resume_retry.rs

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

//! Stale-`--resume` recovery: executing the effects `persistent_resume` decides,
//! locating a recoverable session id, and the retry-batch rewrite.

use super::*;

impl PersistentSubprocessController {
    /// Appends a terminal error-result line to this block's output now and
    /// surfaces its classified failure: a held-back line confirmed final
    /// outside the process-waiter task (its retry failed to launch, or its
    /// generation was superseded), or an error frame synthesized for
    /// prompts that could not be delivered. See
    /// `retry_after_resume_failure`'s doc comment for when a held line is
    /// flushed rather than dropped.
    pub(super) fn flush_error_line_now(&self, line: String) {
        let Some(ref broker) = self.broker else { return };
        let global_output_zone = super::super::shell::resolve_global_output_zone(&self.mstore, &self.block_id);
        super::super::shell::handle_append_block_file(
            broker,
            &self.block_id,
            PERSISTENT_OUTPUT_SUBJECT,
            line.as_bytes(),
            self.filestore.as_ref(),
            global_output_zone.as_deref(),
        );
        // Same gap reagentx flagged (PR #2421 P2) at the other FlushErrorLine
        // call sites: this is also a previously held-back turn's error only
        // now confirmed final (a fresh spawn superseding a still-tracking
        // generation, or a retry batch exhausted with nothing left to
        // deliver) — give it the same classify/persist/publish treatment so
        // it isn't silently dropped from the pane's failure-recovery UI. No
        // exit code exists for this now-superseded turn.
        surface_error_line(&self.block_id, None, &line, Some(broker.as_ref()), &self.mstore, &self.event_bus);
    }

    /// A `Fresh` outcome whose new session will carry AgentMux's record of
    /// the conversation, so the pane says "continued" rather than "new".
    pub(super) fn emit_fresh_outcome_now(&self, attempted_sid: String, continued: bool) {
        self.emit_session_outcome_now_with(persistent_resume::SessionOutcome::Fresh, attempted_sid, None, continued);
    }

    /// Whether a fresh spawn right now would carry a continuation packet
    /// (what `spawn` gives any no-`--resume` spawn onto prior history).
    /// Reads the transcript tail: call it on rare paths, never under `inner`.
    pub(super) fn fresh_spawn_would_continue(&self) -> bool {
        self.has_prior_transcript() && self.continuation_packet().is_some()
    }

    fn emit_session_outcome_now_with(
        &self,
        outcome: persistent_resume::SessionOutcome,
        attempted_sid: String,
        actual_sid: Option<String>,
        continued: bool,
    ) {
        let Some(ref broker) = self.broker else { return };
        let line = session_outcome_line_with(outcome, attempted_sid, actual_sid, continued);
        let global_output_zone = super::super::shell::resolve_global_output_zone(&self.mstore, &self.block_id);
        super::super::shell::handle_append_block_file(
            broker,
            &self.block_id,
            PERSISTENT_OUTPUT_SUBJECT,
            line.as_bytes(),
            self.filestore.as_ref(),
            global_output_zone.as_deref(),
        );
    }

    /// Does a transcript already exist for this pane — either in this
    /// channel's own blockfile, or in the agent's GLOBAL transcript zone?
    ///
    /// The second half is the one that matters for
    /// [`fresh_start_needs_disclosure`]: on a cross-channel or cross-version
    /// open this channel's blockfile is empty, but the pane still renders the
    /// full prior conversation through `app_api::global_output_source`'s read
    /// fallback. Mirrors that function's checks (agent-anchored, not archived,
    /// non-empty `output`) so "the pane will show history" and "we disclose a
    /// fresh start" can't disagree.
    ///
    /// Two `stat` calls at most, on the spawn path only — cheap enough to run
    /// unconditionally behind the caller's own generation gate.
    pub(super) fn has_prior_transcript(&self) -> bool {
        if let Some(ref fs) = self.filestore {
            if matches!(fs.stat(&self.block_id, PERSISTENT_OUTPUT_SUBJECT), Ok(Some(ref f)) if f.size > 0)
            {
                return true;
            }
        }
        let Some(zone) = global_prior_zone(self.mstore.as_deref(), &self.block_id) else {
            return false;
        };
        let Some(gfs) = crate::backend::agent_session::global_transcript_store() else {
            return false;
        };
        matches!(
            gfs.stat(&zone, crate::backend::agent_session::OUTPUT_FILE),
            Ok(Some(ref f)) if f.size > 0
        )
    }

    /// The session to continue on a first spawn that holds no id of its own
    /// (SPEC_DURABLE_CONVERSATION_MEMORY_2026_09_23.md §5 P0a): the session
    /// of the history this pane renders ([`pane_history_session_id`]), when
    /// `--resume` would find it under this spawn's config dir and it isn't
    /// already poisoned.
    pub(super) fn find_continuation_session_id(&self, config: &PersistentSpawnConfig) -> Option<String> {
        if config.resume_flag.is_empty() {
            return None;
        }
        let config_dir = config.env_vars.get("CLAUDE_CONFIG_DIR")?;
        let sid = pane_history_session_id(
            self.filestore.as_deref(),
            self.mstore.as_deref(),
            &self.block_id,
            &config.session_id_field,
        )?;
        if self.inner.lock().unwrap().resume_poisoned.as_deref() == Some(sid.as_str()) {
            return None;
        }
        crate::backend::session_backfill::session_is_reachable(config_dir, &config.working_dir, &sid)
            .then_some(sid)
    }

    /// AgentMux's record of the conversation this pane renders, as a
    /// continuation packet for a process that starts without it
    /// (SPEC_DURABLE_CONVERSATION_MEMORY_2026_09_23.md §4.4).
    pub(super) fn continuation_packet(&self) -> Option<String> {
        let zone = global_prior_zone(self.mstore.as_deref(), &self.block_id);
        let gfs = crate::backend::agent_session::global_transcript_store();
        let global = zone.as_deref().zip(gfs.map(|g| &**g)).map(|(z, g)| (g, z));
        let (tail, starts_mid_line) = packet_history_tail(
            self.filestore.as_deref(),
            &self.block_id,
            global,
            crate::backend::continuity::PACKET_TAIL_BYTES,
        )?;
        // The agent's running summary, kept beside its global transcript
        // (`continuity_state.rs`). Missing for panes that aren't anchored to
        // an agent, or until the first update has run.
        let state = global.and_then(|(g, z)| crate::backend::continuity_state::latest_state(g, z));
        crate::backend::continuity::build_continuation_packet(
            &tail,
            starts_mid_line,
            state.as_ref().map(|s| crate::backend::continuity::RunningState {
                text: &s.text,
                created_at_ms: s.created_at_ms,
            }),
        )
    }

    /// The continuation packet a fresh spawn onto prior history carries, or
    /// `None` when there is no record to give it.
    ///
    /// Carrying one also retracts the "Couldn't resume the previous
    /// conversation — started a new one" banner a rejected `--resume` raised
    /// on the previous generation: the conversation wasn't lost, this session
    /// continues it from AgentMux's record. Same reasoning as the `Resumed`
    /// retract in the stdout reader — the banner must not contradict the
    /// "continued" outcome the pane shows.
    pub(super) fn carry_continuation(&self) -> Option<String> {
        let packet = self.continuation_packet()?;
        tracing::info!(
            block_id = %self.block_id,
            packet_chars = packet.len(),
            "continuity: carrying AgentMux's record of the conversation into the fresh session"
        );
        if let Some(ref store) = self.mstore {
            super::super::session_recovery::clear_resume_failed(store, &self.event_bus, &self.block_id);
        }
        Some(packet)
    }

    /// After a confirmed-stale `--resume` failure, try to recover a REAL
    /// session instead of giving up and starting blank
    /// (`docs/status/STATUS_CROSS_CHANNEL_RESUME_STALE_SESSION_ID_2026_08_20.md`).
    /// Looks for the largest on-disk session under this spawn's own
    /// `CLAUDE_CONFIG_DIR`/working dir — the same "largest session wins"
    /// recovery `session_backfill::backfill_session_ids` already trusts
    /// for the analogous cross-channel-open case.
    ///
    /// Returns `None` (caller falls back to starting blank, same as
    /// before this existed) when: `CLAUDE_CONFIG_DIR` isn't set on this
    /// config, no session exists there, or the only candidate found is
    /// the SAME id already confirmed poisoned — guards against looping
    /// forever "recovering" an id that is itself dead (nothing on disk
    /// changes between attempts, so an unguarded retry would rediscover
    /// the identical bad id every time).
    pub(super) fn find_recovery_session_id(&self, config: &PersistentSpawnConfig) -> Option<String> {
        let config_dir = config.env_vars.get("CLAUDE_CONFIG_DIR")?;
        let candidate = crate::backend::session_backfill::find_largest_session_for_working_dir(
            config_dir,
            &config.working_dir,
        )?;
        let already_poisoned = self.inner.lock().unwrap().resume_poisoned.as_deref() == Some(candidate.as_str());
        if already_poisoned {
            return None;
        }
        // The largest session under this config dir and cwd is only this
        // conversation if it's the head of the agent's segment chain
        // (SPEC_DURABLE_CONVERSATION_MEMORY_2026_09_23.md §4.2). Otherwise
        // (after an account switch, typically) it's older history. Returning
        // `None` makes the retry a fresh session, which now carries the
        // continuation packet.
        if let (Some(uid), Some(gfs)) = (self.stable_agent_uid(), crate::backend::agent_session::global_transcript_store()) {
            let chain = crate::backend::continuity_segments::segments(gfs, &uid);
            let current = self.current_segment.lock().unwrap().clone();
            let head = crate::backend::continuity_segments::head_session(&chain, current.as_deref());
            if !crate::backend::continuity_segments::recovery_allowed(&candidate, head.as_deref()) {
                tracing::info!(
                    block_id = %self.block_id,
                    candidate = %candidate,
                    head = ?head,
                    "continuity: not recovering a session that isn't the head of the agent's chain"
                );
                return None;
            }
        }
        Some(candidate)
    }

    /// Retries the message that triggered a `--resume <sid>` attempt this
    /// controller's own stderr reader just confirmed is unreachable ("No
    /// conversation found with session ID" — see `poison_resume`). Called
    /// from the process-waiter task once the doomed process has actually
    /// exited, via the weak self-reference (mirrors `SubprocessController`'s
    /// queued-message drain — see `set_self_ref`), and ONLY for a
    /// `persistent_resume::ResumeEffect::FireRetry` (a `ConfirmedRetry`
    /// generation's exit) — never for an unrelated exit.
    ///
    /// Respawns with the stale `session_id` replaced: by a recovered
    /// on-disk session if `find_recovery_session_id` finds one, otherwise
    /// cleared so no `--resume` is attempted. Redelivers EVERY message the
    /// doomed process's stdin channel had accepted (the drain appends each
    /// later delivery to the batch —
    /// `persistent_resume::ResumeEvent::MessageAppendedToRetryBatch` — so
    /// this is a batch, not just the one that triggered the spawn) — does
    /// NOT re-persist to the blockfile or re-emit `agent-message-accepted`
    /// for any of them, since both already happened correctly on each
    /// message's original (failed) attempt; only the underlying CLI
    /// process needed a fresh start.
    ///
    /// This is itself a spawn attempt, and must not race a genuinely
    /// concurrent `send_message` call the same way the ORIGINAL doomed
    /// spawn could — see `PersistentInner::spawning_in_progress`'s doc
    /// comment. By the time this runs, the original `send_message` call
    /// that triggered the doomed process has long since returned (its own
    /// spawn-claim-and-deliver sequence completed synchronously, well
    /// before this process even exited), so the WHOLE batch can safely go
    /// through `decide_retry_batch_action` — the SAME decision
    /// `send_message` uses, but enqueueing everything atomically in one
    /// lock acquisition (see its own doc comment for why per-message
    /// decisions aren't safe for a batch).
    ///
    /// A held-back error line must reach the user if a confirmed retry
    /// turns out NOT to actually launch (#2371) — otherwise an
    /// already-accepted prompt ends in total silence: neither the original
    /// error nor a replacement one. It is flushed only when a
    /// `BecomeSpawner` respawn fails; the `FlushClaimed` and `Queued` arms
    /// hand off to a background drain whose own `stalled_with_leftovers`
    /// branch already publishes a status update on genuine total failure,
    /// so those arms drop the line instead — see the `Queued` arm below
    /// for why flushing eagerly there would reproduce the very bubble
    /// #2371 removed.
    ///
    /// `retry_generation` is the spawn generation whose `ProcessExited`
    /// fired this retry (the process-waiter's own `my_generation_wait`) —
    /// `decide_retry_batch_action` needs it to tell a live NEWER spawn
    /// apart from the impossible "our own process is somehow still
    /// running" case (issue #2367). `attempted_sid` is the id that was
    /// just confirmed unreachable — threaded through from
    /// `persistent_resume::ResumeEffect::FireRetry` so this function can
    /// emit the eventual `EmitSessionOutcome` itself once recovery is
    /// known (reagentx + Codex on PR #2693 — `persistent_resume.rs` can
    /// no longer decide Fresh-vs-Resumed at the point it fires this
    /// retry, since recovery might succeed).
    pub(super) fn retry_after_resume_failure(
        &self,
        retry_generation: u64,
        mut config: PersistentSpawnConfig,
        mut entries: Vec<persistent_resume::QueuedRetryEntry>,
        held_error_line: Option<String>,
        attempted_sid: String,
    ) {
        // An empty batch is not a retry at all — see
        // `settle_empty_resume_retry`. Decided BEFORE the "retrying" publish
        // and the recovery search below (codex P1, second round on PR
        // #3551): both are user-visible side effects, and for a superseded
        // generation they must not happen either.
        if entries.is_empty() {
            self.settle_empty_resume_retry(retry_generation, config, held_error_line, attempted_sid);
            return;
        }
        // "Reconnecting…" starts here regardless of which branch below is
        // taken — the user-visible gap begins the moment a retry is known
        // to be needed, not once recovery search finishes. See §6.2.
        publish_resume_retry_status(&self.broker, &self.block_id, "retrying");
        let recovered = self.find_recovery_session_id(&config);
        config.session_id = recovered.clone().unwrap_or_default();
        match &recovered {
            // A real on-disk session was found — don't claim "Resumed"
            // yet (codex P1 on PR #2693: the CLI can still reject this
            // recovered id too, e.g. a corrupt file or a non-top-level
            // session). Left unemitted here on purpose: the spawn below
            // now threads this attempt through the SAME resume-tracking
            // machinery a genuine first-time `--resume` uses (`Some(first
            // .clone())`, not `None`), so `persistent_resume.rs`'s
            // already-correct, already-tested `SessionCaptured` handling
            // emits `Resumed` once the CLI actually confirms it — or, if
            // this recovered id is ALSO stale, cascades into another
            // `ConfirmedRetry` → this same function again, where
            // `find_recovery_session_id`'s poison guard refuses to
            // re-recover the identical dead id and this arm's `None`
            // branch below correctly emits `Fresh` instead. "Reconnecting…"
            // is deliberately NOT resolved here — the eventual
            // `EmitSessionOutcome` handling (stdout-reader / process-exit
            // match arms) is what clears it, once the CLI actually confirms
            // this recovered id one way or the other.
            Some(_) => {}
            // No recovery candidate — this IS genuinely, unambiguously a
            // fresh conversation, decided right now. Safe to emit
            // immediately: nothing downstream can turn this back into a
            // resume. Also resolves "Reconnecting…" immediately, in step
            // with the outcome itself.
            None => {
                publish_resume_retry_status(&self.broker, &self.block_id, "resolved");
                // The respawn below carries the continuation packet whenever
                // there's history (§4.2.1), so say so.
                let continued = self.fresh_spawn_would_continue();
                self.emit_fresh_outcome_now(attempted_sid, continued)
            }
        }
        // Non-empty by the guard at the top.
        let first = entries.remove(0);
        let rest = entries;

        match self.decide_retry_batch_action(retry_generation, &first, &rest) {
            RetryBatchAction::FlushClaimed => {
                // A live, NEWER-generation process was already running by
                // the time this retry got scheduled (issue #2367 — the
                // retry's own process exited; a live `stdin_tx` is by
                // definition an unrelated spawn that raced ahead). The
                // batch was prepended to the queue and `drain_claim` was
                // taken in the SAME lock acquisition that decided this
                // arm, so every concurrent `decide_send_action` caller
                // already routes to `Queued` behind it — the queue is the
                // single ordering authority. Flush through the same drain
                // loop a successful spawn uses (`Sender::send`
                // backpressure, seed-aware retry-batch appends,
                // delivery-order persistence); it releases `drain_claim`
                // once the queue runs dry. This subsumes the previous
                // `try_send`-then-fall-back-to-queue design: everything
                // goes through the queue up front, so the batch can never
                // reorder relative to itself or to racing sends, and the
                // spawn flag is no longer borrowed for a non-spawn (the
                // round-7/round-8 fallback machinery this replaces).
                self.mark_turn_active_and_publish();
                // `allow_fallback_respawn: false` — the flush targets an
                // already-running process; a stall means that process
                // died, and its own exit handling (or the next send's
                // spawn) picks the leftovers up.
                self.drain_queue_with_claim(config.clone(), false, QueueDrainClaim::RetryFlush);
                // reagentx P1 (round 2 on PR #2371): eventual delivery is
                // the overwhelmingly common outcome — flushing the held
                // line eagerly here would show a stale, wrong error
                // bubble immediately followed by the real (successful)
                // response. On the rare total-failure path the drain's
                // own `stalled_with_leftovers` branch (with
                // `allow_fallback_respawn: false`) already calls
                // `publish_status()` — never silent forever, even without
                // the specific original error text.
                drop(held_error_line);
            }
            RetryBatchAction::Queued => {
                // Someone else is already spawning — their own
                // `release_spawn_claim_and_drain_queue` will deliver this
                // (or, if their process turns out to already be dead, its
                // own bounded fallback respawn will).
                //
                // reagentx P1 on PR #2371 (round 2): flushing eagerly HERE
                // (an earlier cut of this fix did, reasoning that losing
                // it silently on eventual failure was worse) contradicts
                // this codebase's own established pattern for a `Queued`
                // outcome — `decide_send_action`'s doc comment: side
                // effects for a queued item happen "later, inside the
                // drain, at the exact moment this message is actually
                // delivered," not eagerly at enqueue time. Eventual
                // success is the OVERWHELMINGLY common outcome for a
                // queued message (that's the whole point of the
                // queue/drain/fallback-respawn infrastructure below), so
                // flushing eagerly would show a stale, wrong error bubble
                // in the common case, immediately followed by the real
                // (successful) response — reproducing the exact bug this
                // PR exists to fix, just via a different path. Dropped
                // instead: on the rare total-failure path (the fallback
                // respawn ALSO fails), `release_spawn_claim_and_drain_queue`'s
                // own stalled-fallback branch already publishes a status
                // update and keeps the messages queued for a future spawn
                // attempt — never silent forever, even without the
                // specific original error text.
                drop(held_error_line);
            }
            RetryBatchAction::BecomeSpawner { own_seq } => {
                // Only clear inner.session_id now that THIS retry is
                // actually about to spawn — codex P2 on PR #2360 (sixth
                // review pass, round 3): clearing it unconditionally up
                // front could erase a session id a DIFFERENT,
                // concurrently-installed process had already legitimately
                // captured, if this retry instead resolved via
                // `DeliverDirect` or `Queued` above — breaking in-memory
                // session tracking and turn-end subagent reconciliation
                // for that process's remaining lifetime.
                // Clear + reserve a fresh generation in one lock
                // acquisition (same race as the leftover-queue fallback —
                // see `clear_session_id_for_fresh_spawn`).
                self.clear_session_id_for_fresh_spawn();
                let retry_config = config.clone();
                // `Some(first.clone())`, not `None` (codex P1 on PR
                // #2693): when `config.session_id` holds a recovered
                // on-disk session (not empty), this MUST thread through
                // as a real resume-tracking seed — same as any first-time
                // `--resume` attempt (mirrors `SendAction::BecomeSpawner`'s
                // own `spawn_process` call above) — or a recovered id
                // that turns out to be ALSO stale/rejected has no
                // detection/retry-cascade at all, stranding the batch on
                // that attempt's raw error instead of eventually falling
                // through to the promised blank-conversation fallback.
                // Harmless when there's nothing to resume: `spawn_process`
                // only constructs `SpawnedWithResume` tracking when
                // `inner.session_id` actually ends up populated, which it
                // won't if `config.session_id` is empty — this always
                // correctly falls back to `SpawnedFresh` in that case,
                // same as passing `None` used to.
                let spawn_result = self.spawn_process(config, Some(first.clone()));
                match &spawn_result {
                    Ok(_) => self.mark_turn_active_and_publish(),
                    Err(e) => tracing::error!(
                        block_id = %self.block_id,
                        error = %e,
                        "failed to respawn after a stale --resume session id"
                    ),
                }
                self.release_spawn_claim_and_drain_queue(spawn_result.is_ok(), retry_config, own_seq);
                if spawn_result.is_err() {
                    // Surface this, or the pane hangs forever with NO
                    // signal at all — codex P2 on PR #2360 (fifth review
                    // pass): the outer process-waiter already suppressed
                    // its own terminal-status publish for the ORIGINAL
                    // exit specifically because a retry was in flight, and
                    // send_message already returned success (possibly
                    // emitting agent-message-accepted) for the message
                    // this retry was supposed to deliver. If this respawn
                    // attempt ALSO fails, nothing else will ever tell the
                    // frontend this turn is over. `inner.proc_status`/
                    // `turn_active` are already `STATUS_DONE`/`false` (set
                    // by the original exit's own cleanup before this
                    // function was ever called) — this just actually
                    // broadcasts that state, which the original exit
                    // deliberately withheld pending this retry's outcome.
                    self.publish_status();
                    // codex P2 on PR #2371: the retry never actually
                    // launched — flush any held error line now instead
                    // of silently dropping it, so the user gets at least
                    // one explanation (the original error) instead of
                    // total silence.
                    if let Some(line) = held_error_line {
                        self.flush_error_line_now(line);
                    }
                }
            }
        }
    }

    /// Same shape of decision as `decide_send_action`, but atomically
    /// enqueues the ENTIRE batch (not just one message) in every arm —
    /// used only by `retry_after_resume_failure`. codex P2 on PR #2360
    /// (sixth review pass, round 6): deciding and enqueueing a
    /// multi-message retry batch one call at a time (each through its own
    /// `decide_send_action` call) left a window between them where a
    /// genuinely new, unrelated message could interleave into the MIDDLE
    /// of the same original batch, reordering it relative to how the
    /// doomed process actually received it. Batch prepend/dedup rules
    /// live in [`Self::prepend_retry_batch`].
    ///
    /// Issue #2367 (spec §4, option 2): the live-process outcome is no
    /// longer a caller-side `try_send` (`DeliverDirect`) — that let a
    /// concurrent send race ahead of, or into the middle of, the batch.
    /// It is now [`RetryBatchAction::FlushClaimed`]: batch prepended and
    /// `drain_claim` taken under this one lock, then flushed through the
    /// shared queue drain.
    pub(super) fn decide_retry_batch_action(
        &self,
        retry_generation: u64,
        first: &persistent_resume::QueuedRetryEntry,
        rest: &[persistent_resume::QueuedRetryEntry],
    ) -> RetryBatchAction {
        let mut inner = self.inner.lock().unwrap();
        if inner.stdin_tx.is_some() && !inner.spawning_in_progress && !inner.drain_claim {
            // The retry's own process exited — that exit is what fired
            // this retry, and the exit-handler clears `stdin_tx` under
            // the same `is_current_generation` gate — so a live
            // `stdin_tx` here is by definition a NEWER, unrelated spawn
            // that raced ahead during the exit-handler's cleanup window
            // (issue #2367).
            debug_assert!(
                inner.spawn_generation != retry_generation,
                "a retry's own generation cannot still be current while stdin_tx is live"
            );
            // Prepend the batch and take the drain claim in THIS SAME
            // lock acquisition (spec §4 option 2): from this instant
            // every `decide_send_action` caller routes to `Queued`
            // behind the batch, so the queue — not a caller's own
            // `try_send` — is the single ordering authority. The only
            // residue is a `DeliverDirect` *decided* before this lock
            // was taken that lands mid-batch: that send was already
            // racing the process exit itself, and no claim scheme can
            // sequence it.
            inner.drain_claim = true;
            Self::prepend_retry_batch(&mut inner, first, rest);
            RetryBatchAction::FlushClaimed
        } else if inner.spawning_in_progress || inner.drain_claim {
            Self::prepend_retry_batch(&mut inner, first, rest);
            RetryBatchAction::Queued
        } else {
            inner.spawning_in_progress = true;
            inner
                .pending_send_messages
                .push_back(QueuedMessage::already_persisted(first.seq, first.json.clone()));
            for entry in rest {
                inner
                    .pending_send_messages
                    .push_back(QueuedMessage::already_persisted(entry.seq, entry.json.clone()));
            }
            RetryBatchAction::BecomeSpawner { own_seq: first.seq }
        }
    }

    /// Prepends a retry batch to `pending_send_messages`, deduping only
    /// `first` by seq. codex P2 on PR #2360 (sixth review pass, round
    /// 11): prepend rather than append — this batch represents messages
    /// accepted by the doomed process BEFORE whatever's currently
    /// queued (anything queued arrived after the original spawn claimed
    /// `spawning_in_progress`, so it's chronologically later); appending
    /// would let the fresh process receive later-arriving input first.
    ///
    /// `first` may itself STILL be sitting in the queue (see
    /// `decide_send_action`'s doc comment on `skip_if_seq_queued`) —
    /// always at index 0 if so, since it's the ONLY thing ever present
    /// when a claim starts and nothing but `push_back` ever touches
    /// this queue elsewhere. In that case `rest` belongs immediately
    /// after it (same batch, preserving order), not ahead of it. `rest`
    /// is always pushed as-is: dedup must never apply WITHIN the same
    /// batch (two entries can legitimately carry identical text and
    /// both need redelivering).
    pub(super) fn prepend_retry_batch(
        inner: &mut PersistentInner,
        first: &persistent_resume::QueuedRetryEntry,
        rest: &[persistent_resume::QueuedRetryEntry],
    ) {
        let first_already_queued = inner.pending_send_messages.iter().any(|m| m.seq == first.seq);
        if first_already_queued {
            for (i, entry) in rest.iter().enumerate() {
                inner
                    .pending_send_messages
                    .insert(i + 1, QueuedMessage::already_persisted(entry.seq, entry.json.clone()));
            }
        } else {
            let mut front: VecDeque<QueuedMessage> = VecDeque::new();
            front.push_back(QueuedMessage::already_persisted(first.seq, first.json.clone()));
            for entry in rest {
                front.push_back(QueuedMessage::already_persisted(entry.seq, entry.json.clone()));
            }
            front.append(&mut inner.pending_send_messages);
            inner.pending_send_messages = front;
        }
    }
}

impl PersistentSubprocessController {
    /// `retry_after_resume_failure`'s handling of an EMPTY retry batch:
    /// `--resume <sid>` was attempted with nothing queued — the eager-resume
    /// path (issue #3463) — and the CLI rejected the id before any prompt
    /// arrived. There is no message to re-send, so this is a terminal
    /// idle-resume failure, not a retry. Codex P1 on PR #3538: returning
    /// early as the old empty-batch branch did left the pane with no
    /// controller status ever published, because the process-waiter hands
    /// `FireRetry` (not `PublishDone`) to the retry path on the assumption
    /// that a launch follows. Complete the recovery *decision* without a
    /// launch:
    ///
    /// - adopt whatever `find_recovery_session_id` found (or `None`) so the
    ///   NEXT message resumes it through the same gated, tracked `--resume`
    ///   a first-time resume gets — its outcome (`Resumed`) is emitted then,
    ///   by the CLI's actual confirmation, never claimed here (codex P1 on
    ///   PR #2693); with no candidate, `Fresh` IS decidable now and is
    ///   disclosed now;
    /// - publish the same terminal status a `PublishDone` exit would.
    ///
    /// No "Reconnecting…" is shown or resolved here, unlike the non-empty
    /// path: no user turn is in flight, so there is no gap to narrate.
    ///
    /// EVERYTHING here is conditioned on this retry's generation still
    /// owning the controller state (codex P1 on PR #3551, both rounds): a
    /// message that arrived after the doomed process cleared `stdin_tx` but
    /// before this ran may already have spawned a newer generation with its
    /// own live session id — possibly resuming a DIFFERENT conversation.
    /// Overwriting that id, stamping `done` over it, or appending a `Fresh`
    /// disclosure to its history would each be wrong. The check runs twice:
    /// once before the recovery search (which does disk I/O outside the
    /// lock, and must not even start for a superseded generation), and
    /// again under the same lock acquisition as the mutation, so nothing
    /// that raced in during the search can be overwritten either.
    /// `spawn_generation` is bumped in the same lock acquisition that
    /// installs a new spawn, and `spawning_in_progress` covers the claim
    /// window before it — together they say whether anyone has moved on.
    ///
    /// The side effects (outcome frame, status publish, error flush) run
    /// UNDER that same lock acquisition, not after it (reagent P1 and codex
    /// P1 on PR #3551, rounds three and four). Every scheme that released
    /// the lock first had a window: a check-then-act with nothing holding
    /// the state still, or a spawn claim that either this settlement took
    /// (fine) or someone else already held (the eager path's own unwinding
    /// claim — whose owner could release it mid-effect and let a send
    /// become a newer generation while the `Fresh` line was still being
    /// written). Holding `inner` across the effects makes the window not
    /// exist: a concurrent `decide_send_action` blocks on the lock and,
    /// once it gets it, becomes a spawner on the already-adopted session.
    /// This is safe because nothing below re-enters the controller — the
    /// broker and event bus are `mpsc` channel sends, the block-file append
    /// and failure persistence touch the file store and SQLite only — and
    /// cheap enough for a path taken once per failed eager resume. It is
    /// the one deliberate exception to the file's convention of dropping
    /// `inner` before effects, which exists because most effect handlers
    /// call back into methods that lock it; these three do not.
    pub(super) fn settle_empty_resume_retry(
        &self,
        retry_generation: u64,
        config: PersistentSpawnConfig,
        held_error_line: Option<String>,
        attempted_sid: String,
    ) {
        // Ownership is generation + no live process — deliberately NOT
        // `!spawning_in_progress` (codex P1 on PR #3551, third round). That
        // flag can still be THIS generation's own eager claim: the eagerly
        // resumed CLI can reject a stale id fast enough for its waiter to
        // fire before `try_eager_resume` reaches the arm that releases it.
        // Treating that as "superseded" dropped the whole settlement while
        // `FireRetry` had already suppressed the waiter's `PublishDone` —
        // status stuck on `running`, recovery candidate and `Fresh` lost.
        // A genuinely newer generation is identified by `spawn_generation`,
        // which `spawn_process` bumps under the lock before anything of the
        // new process exists; a newer spawner that has claimed but not yet
        // bumped has no session of its own yet, and adopting the recovered
        // id underneath it is exactly what it would then `--resume`.
        let owns = |inner: &PersistentInner| inner.spawn_generation == retry_generation && inner.stdin_tx.is_none();
        if !owns(&self.inner.lock().unwrap()) {
            tracing::debug!(
                block_id = %self.block_id,
                retry_generation,
                "empty resume-retry settlement superseded by a newer generation — nothing to do"
            );
            return;
        }
        let recovered = self.find_recovery_session_id(&config);
        // Decided before the lock below: it reads the transcript tail.
        let continued = recovered.is_none() && self.fresh_spawn_would_continue();
        // Second check, mutation, and every side effect under ONE lock
        // acquisition — see the doc comment for why nothing may sit
        // between them.
        let mut inner = self.inner.lock().unwrap();
        if !owns(&inner) {
            return;
        }
        inner.session_id = recovered.clone();
        // Accepted prompts queued behind the eager spawn claim (codex P1 on
        // PR #3551, fifth round): they never reached the doomed process, so
        // they are not in the (empty) retry batch — but they are real work,
        // and this is NOT a message-free settlement. The eager path's own
        // drain will find the dead `stdin_tx`, stall, and hand them to
        // `respawn_once_for_leftover_queue`; leave it the candidate so that
        // respawn resumes it instead of clearing to fresh, and leave the
        // terminal status and the held error line alone — a spawn follows
        // immediately, and the CLI's stale-resume error would otherwise
        // show as a stale bubble right before a successful reply (reagentx
        // P1 on PR #2371's reasoning). `Fresh` is still disclosed when
        // there is nothing to recover: that decision is final either way.
        let queued_behind_claim = !inner.pending_send_messages.is_empty();
        if queued_behind_claim {
            inner.leftover_resume_candidate = recovered.clone();
        } else {
            Self::set_status(&mut inner, STATUS_DONE);
        }
        if recovered.is_none() {
            self.emit_fresh_outcome_now(attempted_sid, continued);
        }
        if queued_behind_claim {
            drop(inner);
            return;
        }
        // `publish_status` would re-lock `inner`; build the same snapshot
        // from the guard we already hold.
        if let Some(ref broker) = self.broker {
            let status = super::super::agent_runtime_status(
                &self.block_id,
                inner.status_version,
                &inner.proc_status,
                inner.proc_exit_code,
                self.health_monitor.is_active_turn(),
            );
            super::super::publish_controller_status(broker, &status);
            // `set_status(STATUS_DONE)` above cleared what the process was
            // spawned with; announce it from the guard already held.
            super::status::announce_runtime(broker, &self.block_id, &inner);
        }
        // Nothing was accepted from the user, but the CLI's own account of
        // why the resume failed still reaches them (codex P2 on PR #2371).
        if let Some(line) = held_error_line {
            self.flush_error_line_now(line);
        }
        drop(inner);
    }
}

/// The session id of the conversation a pane renders: the last provider
/// session id in its own transcript, else in the agent's global transcript
/// zone. Same sources, same order, as `has_prior_transcript`, so "the pane
/// shows history" and "the spawn continues it" can't disagree.
///
/// Taken from AgentMux's own record rather than a scan of the provider's
/// dir, so an archived conversation (every explicit "start fresh" action
/// archives) is never picked up, and a subagent's transcript can't pass for
/// the agent's own. No reachability check; callers decide what an
/// unreachable id means.
pub(crate) fn pane_history_session_id(
    filestore: Option<&FileStore>,
    mstore: Option<&Store>,
    block_id: &str,
    session_id_field: &str,
) -> Option<String> {
    if session_id_field.is_empty() {
        return None;
    }
    let (tail, starts_mid_line) = pane_history_tail(filestore, mstore, block_id, CONTINUATION_TAIL_BYTES)?;
    last_session_id_in_stream(&tail, starts_mid_line, session_id_field)
}

/// The last `max_bytes` of the history a pane renders: its own transcript,
/// else the agent's global zone. The `bool` is whether the window starts
/// partway into the file (so its first line is a fragment).
pub(crate) fn pane_history_tail(
    filestore: Option<&FileStore>,
    mstore: Option<&Store>,
    block_id: &str,
    max_bytes: i64,
) -> Option<(Vec<u8>, bool)> {
    if let Some(tail) = filestore.and_then(|fs| read_transcript_tail(fs, block_id, PERSISTENT_OUTPUT_SUBJECT, max_bytes)) {
        return Some(tail);
    }
    let zone = global_prior_zone(mstore, block_id)?;
    let gfs = crate::backend::agent_session::global_transcript_store()?;
    read_transcript_tail(gfs, &zone, crate::backend::agent_session::OUTPUT_FILE, max_bytes)
}

/// The transcript a continuation packet is built from: the agent's global
/// zone when the pane has one, else the pane's own transcript.
///
/// Deliberately NOT [`pane_history_tail`]'s local-first order. The zone
/// mirrors every append to the pane file, so it holds everything the pane
/// does, plus the conversation from earlier channels and builds. On a new
/// channel the pane file is created by a rejected resume's own settlement
/// (the session-outcome line and the CLI's error result), and reading it
/// first built no packet at all while the global record held the whole
/// conversation (REPORT_AGENT_HISTORY_LOST_ON_NEW_BUILD_2026_09_25.md §4).
pub(super) fn packet_history_tail(
    filestore: Option<&FileStore>,
    block_id: &str,
    global: Option<(&FileStore, &str)>,
    max_bytes: i64,
) -> Option<(Vec<u8>, bool)> {
    global
        .and_then(|(gfs, zone)| read_transcript_tail(gfs, zone, crate::backend::agent_session::OUTPUT_FILE, max_bytes))
        .or_else(|| filestore.and_then(|fs| read_transcript_tail(fs, block_id, PERSISTENT_OUTPUT_SUBJECT, max_bytes)))
}

/// The agent's global transcript zone, when this pane would render it:
/// agent-anchored and not archived.
pub(super) fn global_prior_zone(mstore: Option<&Store>, block_id: &str) -> Option<String> {
    let block = mstore?.must_get::<crate::backend::obj::Block>(block_id).ok()?;
    let archived = block
        .meta
        .get(crate::backend::session_archive::META_SESSION_ARCHIVED_AT)
        .and_then(|v| v.as_i64())
        .is_some_and(|v| v > 0);
    if archived {
        return None;
    }
    crate::backend::agent_session::agent_zone_for_block_meta(&block.meta)
}

/// The last `max_bytes` of a transcript, and whether that window starts
/// partway into the file (so its first line is a fragment).
fn read_transcript_tail(fs: &FileStore, zone: &str, name: &str, max_bytes: i64) -> Option<(Vec<u8>, bool)> {
    let file = fs.stat(zone, name).ok()??;
    if file.size <= 0 {
        return None;
    }
    let offset = (file.size - max_bytes).max(0);
    let (actual_offset, data) = fs.read_at(zone, name, offset, file.size - offset).ok()?;
    (!data.is_empty()).then_some((data, actual_offset > 0))
}