(MOT-4202) fix(harness): reseed a turn when the finalize drain delivers a parked message

harness/src/functions/send.rs

@@ -638,7 +638,11 @@ async fn seed_or_merge(
     }
 }
 
-async fn seed_new(
+/// Seed a fresh turn record and enqueue its first step. Exposed to the turn
+/// loop's finalize-drain reseed path (`turn_loop::reseed_after_finalize_drain`)
+/// so a notification that parked during a turn's final step gets a turn to
+/// react to it, instead of being drained to the transcript and stranded.
+pub(crate) async fn seed_new(
     deps: &Deps,
     cfg: &WorkerConfig,
     session_id: &str,

harness/src/turn_loop.rs

@@ -1058,18 +1058,31 @@ async fn has_queued(deps: &Deps, record: &TurnRecord) -> Result<bool, HarnessErr
         .any(|r| !matches!(r.message, AgentMessage::Custom(_))))
 }
 
-/// Drain the session's message queue into the transcript in arrival order.
-/// Idempotent: each row appends under its stored deterministic entry id, and
-/// rows are deleted only after the append lands — a redelivered step re-drains
-/// as a no-op.
+/// How many of these parked rows are MODEL-VISIBLE (non-custom). Custom-role
+/// rows are transcript-only status notices that never enter the model context,
+/// so they neither steer a live turn (`has_queued`) nor, at finalize, warrant
+/// waking a fresh one — a re-generate over an assistant-tailed context is a
+/// guaranteed provider prefill rejection. A parked notification arrives as a
+/// user-role message, so it counts.
+fn count_model_visible(rows: &[crate::state::QueuedMessage]) -> usize {
+    rows.iter()
+        .filter(|r| !matches!(r.message, AgentMessage::Custom(_)))
+        .count()
+}
+
+/// Drain the session's message queue into the transcript in arrival order,
+/// returning how many drained rows were MODEL-VISIBLE. Idempotent: each row
+/// appends under its stored deterministic entry id, and rows are deleted only
+/// after the append lands — a redelivered step re-drains as a no-op and reports
+/// zero, since there is nothing left to drain.
 async fn drain_queued(
     deps: &Deps,
     session: &SessionClient,
     session_id: &str,
 ) -> Result<usize, HarnessError> {
     let cfg = deps.cfg().await;
     let rows = crate::state::list_queued(&deps.iii, session_id, cfg.session_timeout_ms).await?;
-    let drained = rows.len();
+    let model_visible = count_model_visible(&rows);
     for row in rows {
         session
             .append(
@@ -1082,15 +1095,55 @@ async fn drain_queued(
             .await?;
         crate::state::delete_queued(&deps.iii, session_id, &row.id, cfg.session_timeout_ms).await?;
     }
-    Ok(drained)
+    Ok(model_visible)
 }
 
-/// Best-effort finalize drain: a message enqueued after the loop's last queue
-/// check still lands in the transcript (unreacted — the turn is over, same as
-/// a pre-queue merged send racing completion). Never blocks the finalise.
-async fn drain_queued_best_effort(deps: &Deps, session: &SessionClient, session_id: &str) {
-    if let Err(e) = drain_queued(deps, session, session_id).await {
-        tracing::warn!(session_id = %session_id, error = %e, "finalize queue drain failed");
+/// Finalize drain: a message that parked after the loop's last in-step queue
+/// check still lands in the transcript here. Returns `true` when it delivered a
+/// MODEL-VISIBLE message — the signal that the finalizing turn must reseed
+/// (via [`reseed_after_finalize_drain`]) so something reacts to it. Without the
+/// reseed a parked notification sits unread with no turn to process it, which
+/// strands an autonomous run that ended its turn expecting the fire to wake it.
+/// Never blocks the finalise.
+async fn drain_queued_best_effort(deps: &Deps, session: &SessionClient, session_id: &str) -> bool {
+    match drain_queued(deps, session, session_id).await {
+        Ok(model_visible) => model_visible > 0,
+        Err(e) => {
+            tracing::warn!(session_id = %session_id, error = %e, "finalize queue drain failed");
+            false
+        }
+    }
+}
+
+/// Seed a fresh turn after a finalize drain delivered a model-visible message
+/// with no turn to react to it. Reuses the finalized turn's frozen options
+/// (model / provider / dispatch policy / prompt) and last-acked registry
+/// generation so the woken turn keeps the agent's capabilities — the same
+/// outcome an external `harness::send` produces against a now-terminal session.
+///
+/// MUST be called AFTER the terminal `put_turn`: the turn slot is keyed per
+/// session, so seeding before the finalize write would be clobbered by it. The
+/// caller gates on the drain actually delivering a row, so a redelivered
+/// finalize (queue at-least-once) drains nothing and does not double-seed; a
+/// concurrent external send racing the same slot is resolved by `run_step`'s
+/// stale-turn guard, exactly as two racing sends already are.
+async fn reseed_after_finalize_drain(deps: &Deps, record: &TurnRecord) {
+    let cfg = deps.cfg().await;
+    if let Err(e) = crate::functions::send::seed_new(
+        deps,
+        &cfg,
+        &record.session_id,
+        record.options.clone(),
+        record.functions_generation,
+        None,
+    )
+    .await
+    {
+        tracing::warn!(
+            session_id = %record.session_id,
+            error = %e,
+            "reseed after finalize drain failed; a parked notification may be stranded",
+        );
     }
 }
 
@@ -1226,7 +1279,7 @@ async fn finalize_completed(
     record: &mut TurnRecord,
     result: Option<Value>,
 ) -> Result<TurnStepResult, HarnessError> {
-    drain_queued_best_effort(deps, session, &record.session_id).await;
+    let woke = drain_queued_best_effort(deps, session, &record.session_id).await;
     let cfg = deps.cfg().await;
     record.status = TurnStatus::Completed;
     record.result = result.clone();
@@ -1256,6 +1309,19 @@ async fn finalize_completed(
     if let Some(parent) = record.parent.clone() {
         crate::deferred::resolve_parent(deps, &parent, "completed", result.as_ref(), None).await;
     }
+    // Second sweep, AFTER the terminal write, pairing with `try_enqueue`'s
+    // post-enqueue recheck: a send whose recheck still saw `Running` must have
+    // enqueued before the terminal write landed, so this sweep collects its
+    // row; a recheck that sees the terminal record seeds its own turn. Without
+    // it, a row enqueued between the first drain and the terminal write would
+    // strand — queued against a turn that will never drain again.
+    let woke = woke || drain_queued_best_effort(deps, session, &record.session_id).await;
+    // A message parked during this turn's final step was just drained to the
+    // transcript with no turn to react to it; seed one now (after the terminal
+    // write above, or it would clobber the fresh turn's slot).
+    if woke {
+        reseed_after_finalize_drain(deps, record).await;
+    }
     Ok(TurnStepResult {
         session_id: record.session_id.clone(),
         status: TurnStatus::Completed,
@@ -1271,7 +1337,7 @@ async fn finalize_failed(
     reason: &str,
     failure: FailureInfo,
 ) -> Result<TurnStepResult, HarnessError> {
-    drain_queued_best_effort(deps, session, &record.session_id).await;
+    let woke = drain_queued_best_effort(deps, session, &record.session_id).await;
     let cfg = deps.cfg().await;
     record.status = TurnStatus::Failed;
     record.result_error = Some(reason.to_string());
@@ -1352,6 +1418,16 @@ async fn finalize_failed(
             notify_parent_of_child_failure(deps, &parent.session_id, record, reason, failure).await;
         }
     }
+    // Second post-terminal sweep, as in `finalize_completed`: closes the
+    // enqueue-after-drain window against `try_enqueue`'s recheck.
+    let woke = woke || drain_queued_best_effort(deps, session, &record.session_id).await;
+    // As in `finalize_completed`: a message that parked during the failing
+    // turn's final step is genuine new input (a notification, a steer) and
+    // deserves a turn, the same as an external send arriving at a failed
+    // session. Gated on the drain, so it cannot loop on the failure itself.
+    if woke {
+        reseed_after_finalize_drain(deps, record).await;
+    }
     Ok(TurnStepResult {
         session_id: record.session_id.clone(),
         status: TurnStatus::Failed,
@@ -1540,7 +1616,10 @@ async fn finalize_cancelled(
     record: &mut TurnRecord,
     reason: &str,
 ) -> Result<TurnStepResult, HarnessError> {
-    drain_queued_best_effort(deps, session, &record.session_id).await;
+    // Deliver any parked rows to the transcript but do NOT reseed: the user
+    // stopped this turn, so a parked notification waits for the next explicit
+    // send rather than auto-waking a turn they just cancelled.
+    let _ = drain_queued_best_effort(deps, session, &record.session_id).await;
     let cfg = deps.cfg().await;
     record.status = TurnStatus::Cancelled;
     record.updated_at = AgentMessage::now_ms();
@@ -1579,6 +1658,10 @@ async fn finalize_cancelled(
     if let Some(parent) = record.parent.clone() {
         crate::deferred::resolve_parent(deps, &parent, "cancelled", None, Some(reason)).await;
     }
+    // Second post-terminal sweep (see `finalize_completed`): a row enqueued
+    // between the first drain and the terminal write still reaches the
+    // transcript. Still no reseed — the user cancelled.
+    let _ = drain_queued_best_effort(deps, session, &record.session_id).await;
     Ok(TurnStepResult {
         session_id: record.session_id.clone(),
         status: TurnStatus::Cancelled,
@@ -2236,9 +2319,66 @@ impl Clone for SessionStreamSink {
 
 #[cfg(test)]
 mod tests {
-    use super::{cancel_requested, transient_resume_allowed};
+    use super::{cancel_requested, count_model_visible, transient_resume_allowed};
     use crate::types::content::ContentBlock;
     use crate::types::event::{ErrorKind, StopReason};
+    use crate::types::message::{AgentMessage, CustomMessage, CustomRoleTag};
+
+    fn queued(message: AgentMessage) -> crate::state::QueuedMessage {
+        crate::state::QueuedMessage {
+            id: "q".into(),
+            session_id: "s_1".into(),
+            message,
+            entry_id: "e".into(),
+            origin: None,
+            queued_at: 0,
+        }
+    }
+
+    fn custom_notice(text: &str) -> AgentMessage {
+        AgentMessage::Custom(CustomMessage {
+            role: CustomRoleTag::Custom,
+            custom_type: "notice".into(),
+            content: vec![ContentBlock::text(text)],
+            display: None,
+            details: None,
+            timestamp: 0,
+        })
+    }
+
+    /// The gate that fixes the "notification parked during a turn's final step
+    /// is stranded" bug: `finalize_completed`/`finalize_failed` reseed a turn
+    /// only when the finalize drain delivered a MODEL-VISIBLE message. A
+    /// notification arrives as a user-role message, so it counts and wakes a
+    /// turn; a custom-role status notice drains to the transcript but must not
+    /// reseed (a re-generate over an assistant-tailed context would be a
+    /// provider prefill rejection). A redelivered finalize drains nothing, so
+    /// it reports zero and cannot double-seed.
+    #[test]
+    fn finalize_reseed_gate_counts_only_model_visible_rows() {
+        assert_eq!(count_model_visible(&[]), 0, "empty queue never reseeds");
+
+        let notice = queued(custom_notice("scanning…"));
+        assert_eq!(
+            count_model_visible(std::slice::from_ref(&notice)),
+            0,
+            "a custom-only queue drains but must not reseed",
+        );
+
+        let notification = queued(AgentMessage::user_text("[notification] chunk-done"));
+        assert_eq!(
+            count_model_visible(std::slice::from_ref(&notification)),
+            1,
+            "a parked notification (user role) reseeds so the agent reacts",
+        );
+
+        let steer = queued(AgentMessage::user_text("also check the tests"));
+        assert_eq!(
+            count_model_visible(&[notification, notice, steer]),
+            2,
+            "only the model-visible rows gate the reseed",
+        );
+    }
 
     #[test]
     fn request_overhead_always_reserves_provider_framing() {

harness/tests/e2e/README.md

@@ -13,6 +13,7 @@ No provider key or network access is required.
 |---|---|---|---|
 | E2E-001 | `streamed-text` | direct | streamed text reaches durable completion |
 | E2E-002 | `exactly-once-function` | direct | a native function executes exactly once |
+| E2E-003 | `reseed-parked-message` | direct | a message parked during a turn's failing final step is delivered by a harness-reseeded turn |
 | UI-001 | `console-streamed-text` | playground | a message sent by the Console streams to durable completion |
 | UI-002 | `multi-turn-traces` | playground | a native function turn and a Console turn expose distinct traces and function-call events |
 
@@ -93,9 +94,19 @@ cargo clippy --manifest-path harness/Cargo.toml \
   -p harness-integration --all-targets -- -D warnings
 ```
 
-`validate --scenario all` checks exactly the four fixtures. `run --scenario
-all` executes only E2E-001 and E2E-002; UI-001 and UI-002 must use
-`playground`.
+`validate --scenario all` checks every fixture. `run --scenario all` executes
+the direct scenarios (E2E-001, E2E-002, E2E-003); UI-001 and UI-002 must use
+`playground`. E2E-003 produces two terminal turns from one send: generation 1
+steers a message into the running session (it parks durably) and then fails,
+so the harness's failed finalize drains the parked row and reseeds a turn to
+react to it. The failed route is deliberate — a park during a *completing*
+terminal generation is always delivered earlier by the loop's steering check,
+so only the failed finalize (which has no steering check) reaches the drain
+deterministically from the public boundary; both finalize paths share the
+drain-and-reseed under test. The fixture declares the per-turn statuses
+(`failed`, then `completed`) and the floor enforces them positionally, along
+with a single trace covering both turns (harness-seeded turns chain into the
+originating send's trace).
 
 The fixture tests pin:
 

harness/tests/e2e/src/fixtures/loading.rs

@@ -7,8 +7,14 @@ use crate::types::script::RouterScriptV1;
 pub struct ScenarioFixture {
     pub slug: String,
     pub driver: ScenarioDriver,
-    /// Number of distinct terminal turns the playground must observe.
+    /// Number of distinct terminal turns the run must observe. Usually 1;
+    /// Playground turns and harness-initiated follow-on turns (e.g. a reseed
+    /// after a message parks mid-final-step) push it higher. Always equals
+    /// `expected_turn_statuses.len()`.
     pub expected_terminal_turns: usize,
+    /// Each terminal turn's lifecycle status, in completion order. The last
+    /// must be `completed` — the floor's durable-status check binds to it.
+    pub expected_turn_statuses: Vec<String>,
     pub scenario: CompiledScenarioV1,
     pub script: RouterScriptV1,
     /// Compiled Harness default plus inferred session/policy aid.
@@ -48,9 +54,25 @@ impl ScenarioFixture {
             "scenario must expect at least one terminal turn"
         );
         anyhow::ensure!(
-            self.driver == ScenarioDriver::Playground || self.expected_terminal_turns == 1,
-            "direct scenarios must expect exactly one terminal turn"
+            self.expected_turn_statuses.len() == self.expected_terminal_turns,
+            "scenario declares {} turn status(es) for {} terminal turn(s)",
+            self.expected_turn_statuses.len(),
+            self.expected_terminal_turns
         );
+        for status in &self.expected_turn_statuses {
+            anyhow::ensure!(
+                matches!(status.as_str(), "completed" | "failed" | "cancelled"),
+                "unknown terminal turn status {status:?}"
+            );
+        }
+        anyhow::ensure!(
+            self.expected_turn_statuses.last().map(String::as_str) == Some("completed"),
+            "the last terminal turn must be completed"
+        );
+        // A direct scenario is one external send, but the harness may seed
+        // further turns from it (a reseed after a parked message); those extra
+        // terminal turns are declared with `terminal_turns(n)` and awaited the
+        // same way Playground awaits its externally driven turns.
         if self.script.scenario_id != self.scenario.id {
             anyhow::bail!(
                 "script scenario_id {:?} does not match scenario id {:?}",
@@ -76,6 +98,14 @@ impl ScenarioFixture {
                 "duplicate router generation ordinal {}",
                 generation.ordinal
             );
+            if generation.failure.is_some() {
+                anyhow::ensure!(
+                    generation.frames.is_empty(),
+                    "failing generation {} must stream no frames",
+                    generation.ordinal
+                );
+                continue;
+            }
             anyhow::ensure!(
                 generation
                     .frames
@@ -100,6 +130,18 @@ impl ScenarioFixture {
         Ok(())
     }
 
+    /// Distinct trace trees the run must produce. Turns chain into their
+    /// initiating trace through the durable queue, so a direct scenario's one
+    /// send yields exactly one trace no matter how many turns it seeds (a
+    /// finalize reseed enqueues from inside the finalizing step); Playground
+    /// turns are each externally initiated and trace separately.
+    pub fn expected_traces(&self) -> usize {
+        match self.driver {
+            ScenarioDriver::Direct => 1,
+            ScenarioDriver::Playground => self.expected_terminal_turns,
+        }
+    }
+
     pub fn compiled(&self) -> CompiledFixtureV1 {
         CompiledFixtureV1 {
             scenario: self.scenario.clone(),

harness/tests/e2e/src/fixtures/tests.rs

@@ -1,22 +1,22 @@
 use super::*;
 
 #[test]
-fn all_selection_returns_the_four_checked_in_fixtures() {
+fn all_selection_returns_the_checked_in_fixtures() {
     let fixtures = scenario_fixtures("all").unwrap();
     let ids = fixtures
         .iter()
         .map(|fixture| fixture.scenario.id.as_str())
         .collect::<std::collections::BTreeSet<_>>();
     assert_eq!(
         ids,
-        std::collections::BTreeSet::from(["E2E-001", "E2E-002", "UI-001", "UI-002"])
+        std::collections::BTreeSet::from(["E2E-001", "E2E-002", "E2E-003", "UI-001", "UI-002"])
     );
     assert_eq!(
         fixtures
             .iter()
             .filter(|fixture| fixture.driver == crate::scenarios::ScenarioDriver::Direct)
             .count(),
-        2
+        3
     );
 }