diff --git a/harness/src/functions/send.rs b/harness/src/functions/send.rs
index 5ebe188dc..16e12aa45 100644
--- a/harness/src/functions/send.rs
+++ b/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,
diff --git a/harness/src/turn_loop.rs b/harness/src/turn_loop.rs
index 7abd66656..3165bc71e 100644
--- a/harness/src/turn_loop.rs
+++ b/harness/src/turn_loop.rs
@@ -1058,10 +1058,23 @@ 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,
@@ -1069,7 +1082,7 @@ async fn drain_queued(
 ) -> 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() {
diff --git a/harness/tests/e2e/README.md b/harness/tests/e2e/README.md
index 6b38d5bbc..437eb14fb 100644
--- a/harness/tests/e2e/README.md
+++ b/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:
 
diff --git a/harness/tests/e2e/src/fixtures/loading.rs b/harness/tests/e2e/src/fixtures/loading.rs
index 9e973eaf0..42b47fb59 100644
--- a/harness/tests/e2e/src/fixtures/loading.rs
+++ b/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(),
diff --git a/harness/tests/e2e/src/fixtures/tests.rs b/harness/tests/e2e/src/fixtures/tests.rs
index baaaad2a7..b88ecbad7 100644
--- a/harness/tests/e2e/src/fixtures/tests.rs
+++ b/harness/tests/e2e/src/fixtures/tests.rs
@@ -1,7 +1,7 @@
 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()
@@ -9,14 +9,14 @@ fn all_selection_returns_the_four_checked_in_fixtures() {
         .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
     );
 }
 
diff --git a/harness/tests/e2e/src/scenario/floor.rs b/harness/tests/e2e/src/scenario/floor.rs
index b6ffd554d..195052e45 100644
--- a/harness/tests/e2e/src/scenario/floor.rs
+++ b/harness/tests/e2e/src/scenario/floor.rs
@@ -10,18 +10,42 @@ use crate::probe::LIFECYCLE_FUNCTION_ID;
 use crate::scenarios::VerifyFn;
 use crate::types::trace::strip_engine_fields;
 
-/// First violated floor post-condition, if any.
+/// Scenario-declared shape of a passing run.
+pub struct FloorExpectations<'a> {
+    /// Each terminal turn's lifecycle status, in completion order.
+    pub turn_statuses: &'a [String],
+    /// Distinct trace trees (one per externally initiated send).
+    pub traces: usize,
+}
+
+impl FloorExpectations<'_> {
+    fn turns(&self) -> usize {
+        self.turn_statuses.len()
+    }
+
+    fn declares_failure(&self) -> bool {
+        self.turn_statuses
+            .iter()
+            .any(|status| status != "completed")
+    }
+}
+
+/// First violated floor post-condition for a single-completed-turn run.
 pub fn floor_failure(run: &RunEvidence) -> Option<String> {
-    floor_failure_for_turns(run, 1)
+    let statuses = ["completed".to_string()];
+    floor_failure_for(
+        run,
+        &FloorExpectations {
+            turn_statuses: &statuses,
+            traces: 1,
+        },
+    )
 }
 
-pub fn floor_failure_for_turns(
-    run: &RunEvidence,
-    expected_terminal_turns: usize,
-) -> Option<String> {
+pub fn floor_failure_for(run: &RunEvidence, expected: &FloorExpectations) -> Option<String> {
     terminal_failure(run)
-        .or_else(|| trace_failure(run, expected_terminal_turns))
-        .or_else(|| lifecycle_failure(run, expected_terminal_turns))
+        .or_else(|| trace_failure(run, expected))
+        .or_else(|| lifecycle_failure(run, expected))
         .or_else(|| generations_failure(run))
         .or_else(|| send_flags_failure(run))
 }
@@ -68,19 +92,28 @@ fn status_list_len(run: &RunEvidence, key: &str) -> usize {
         .unwrap_or(0)
 }
 
-fn trace_failure(run: &RunEvidence, expected_terminal_turns: usize) -> Option<String> {
+fn trace_failure(run: &RunEvidence, expected: &FloorExpectations) -> Option<String> {
     let summary = &run.traces.summary;
-    let trace_count = summary.trace_count == expected_terminal_turns;
-    let turn_count = summary.turn_ids.len() == expected_terminal_turns;
+    let trace_count = summary.trace_count == expected.traces;
+    let turn_count = summary.turn_ids.len() == expected.turns();
     let complete = summary.pending_span_count == 0;
-    let clean = summary.error_count == 0;
+    // A declared-failed turn must surface its failure in the traces; a run
+    // declared all-completed must stay clean.
+    let clean = if expected.declares_failure() {
+        summary.error_count > 0
+    } else {
+        summary.error_count == 0
+    };
     let latest_bound = summary.turn_ids.last().map(String::as_str) == run.turn_id.as_deref();
     if trace_count && turn_count && complete && clean && latest_bound {
         return None;
     }
     Some(format!(
-        "floor: traces must cover {expected_terminal_turns} completed clean turn(s) (trace count: \
+        "floor: {} trace(s) must cover {} terminal turn(s) with statuses {:?} (trace count: \
          {}, turn count: {}, pending spans: {}, error spans: {}, latest turn bound: {latest_bound})",
+        expected.traces,
+        expected.turns(),
+        expected.turn_statuses,
         summary.trace_count,
         summary.turn_ids.len(),
         summary.pending_span_count,
@@ -89,7 +122,8 @@ fn trace_failure(run: &RunEvidence, expected_terminal_turns: usize) -> Option<St
 }
 
 /// Validate `harness::turn-completed` from the lifecycle sink's trace events.
-fn lifecycle_failure(run: &RunEvidence, expected_terminal_turns: usize) -> Option<String> {
+fn lifecycle_failure(run: &RunEvidence, expected: &FloorExpectations) -> Option<String> {
+    let expected_terminal_turns = expected.turns();
     let lifecycle_name = format!("execute {LIFECYCLE_FUNCTION_ID}");
     let lifecycle: Vec<Option<Value>> = run
         .spans_named(&lifecycle_name)
@@ -134,14 +168,25 @@ fn lifecycle_failure(run: &RunEvidence, expected_terminal_turns: usize) -> Optio
         }
     }
     let turn_count = turns.len() == expected_terminal_turns;
+    // Each turn's status must equal the declared status for its position in
+    // trace order (summary turn ids are ordered by first span start).
+    let expected_by_turn: BTreeMap<&str, &str> = run
+        .traces
+        .summary
+        .turn_ids
+        .iter()
+        .map(String::as_str)
+        .zip(expected.turn_statuses.iter().map(String::as_str))
+        .collect();
     let bound = payloads.iter().all(|payload| {
-        payload.get("status").and_then(Value::as_str) == Some("completed")
+        let declared = payload
+            .get("turn_id")
+            .and_then(Value::as_str)
+            .and_then(|turn_id| expected_by_turn.get(turn_id).copied());
+        declared.is_some()
+            && payload.get("status").and_then(Value::as_str) == declared
             && payload.get("terminal").and_then(Value::as_bool) == Some(true)
             && payload.get("session_id").and_then(Value::as_str) == Some(run.session_id.as_str())
-            && payload
-                .get("turn_id")
-                .and_then(Value::as_str)
-                .is_some_and(|turn_id| run.traces.summary.turn_ids.iter().any(|id| id == turn_id))
     });
 
     let allowed_keys = BTreeSet::from([
@@ -248,16 +293,20 @@ mod tests {
         }
     }
 
-    fn completed_lifecycle(turn_id: &str, timestamp: i64) -> Value {
+    fn terminal_lifecycle(turn_id: &str, status: &str, timestamp: i64) -> Value {
         json!({
             "session_id": "session-1",
             "turn_id": turn_id,
-            "status": "completed",
+            "status": status,
             "terminal": true,
             "timestamp": timestamp
         })
     }
 
+    fn completed_lifecycle(turn_id: &str, timestamp: i64) -> Value {
+        terminal_lifecycle(turn_id, "completed", timestamp)
+    }
+
     fn clean_evidence() -> RunEvidence {
         RunEvidence {
             run_id: "run-1".into(),
@@ -328,6 +377,13 @@ mod tests {
             .contains("consistent retries: false"));
     }
 
+    fn expectations<'a>(turn_statuses: &'a [String], traces: usize) -> FloorExpectations<'a> {
+        FloorExpectations {
+            turn_statuses,
+            traces,
+        }
+    }
+
     #[test]
     fn multiple_turns_are_bound_in_trace_order() {
         let mut evidence = clean_evidence();
@@ -337,7 +393,49 @@ mod tests {
             roots: vec![lifecycle_span("turn-2", completed_lifecycle("turn-2", 2))],
         });
         evidence.traces = TraceEvidenceV1::new(evidence.traces.traces);
-        assert_eq!(floor_failure_for_turns(&evidence, 2), None);
+        let statuses = vec!["completed".to_string(), "completed".to_string()];
+        assert_eq!(
+            floor_failure_for(&evidence, &expectations(&statuses, 2)),
+            None
+        );
+    }
+
+    /// The reseed shape: one send, one trace, a failed turn whose finalize
+    /// drain seeded a completing follow-on turn — with the failure visible as
+    /// an error span.
+    #[test]
+    fn declared_failed_turn_passes_with_error_spans_in_one_trace() {
+        let mut evidence = clean_evidence();
+        evidence.turn_id = Some("turn-2".into());
+        let mut failed_root = lifecycle_span("turn-1", terminal_lifecycle("turn-1", "failed", 1));
+        failed_root.status = "error".into();
+        evidence.traces = TraceEvidenceV1::new(vec![TraceTreeV1 {
+            trace_id: "trace-turn-1".into(),
+            roots: vec![
+                failed_root,
+                lifecycle_span("turn-2", completed_lifecycle("turn-2", 2)),
+            ],
+        }]);
+        let statuses = vec!["failed".to_string(), "completed".to_string()];
+        assert_eq!(
+            floor_failure_for(&evidence, &expectations(&statuses, 1)),
+            None
+        );
+
+        // The same evidence fails an all-completed expectation.
+        let all_completed = vec!["completed".to_string(), "completed".to_string()];
+        assert!(
+            floor_failure_for(&evidence, &expectations(&all_completed, 1))
+                .unwrap()
+                .contains("error spans: 1")
+        );
+
+        // Statuses are positional: swapping the declaration fails binding.
+        let swapped = vec!["completed".to_string(), "failed".to_string()];
+        assert!(
+            floor_failure_for(&evidence, &expectations(&swapped, 1)).is_some(),
+            "swapped statuses must not pass"
+        );
     }
 
     #[test]
diff --git a/harness/tests/e2e/src/scenario/phases/completion.rs b/harness/tests/e2e/src/scenario/phases/completion.rs
index 9eac53b33..0affc4d3e 100644
--- a/harness/tests/e2e/src/scenario/phases/completion.rs
+++ b/harness/tests/e2e/src/scenario/phases/completion.rs
@@ -24,10 +24,33 @@ impl ScenarioRunner<'_> {
 
         // The event only wakes collection; trace evidence verifies the
         // lifecycle delivery itself. Status remains the durable-state check.
-        match services.probe().wait_for_completion(deadline).await {
-            Ok(observation) => {
+        //
+        // A single send can still produce more than one terminal turn when the
+        // harness seeds a follow-on turn from it (a reseed after a message
+        // parks mid-final-step). Such a scenario declares `terminal_turns(n)`;
+        // wait for all n and bind evidence to the latest, mirroring how
+        // Playground awaits its externally driven turns.
+        let expected = self.fixture.expected_terminal_turns;
+        let latest_turn_id = if expected > 1 {
+            services
+                .probe()
+                .wait_for_completion_turns(expected, deadline)
+                .await
+                .map(|events| latest_terminal_observation(&events).map(|o| o.event.turn_id.clone()))
+        } else {
+            services
+                .probe()
+                .wait_for_completion(deadline)
+                .await
+                .map(|observation| Some(observation.event.turn_id))
+        };
+        match latest_turn_id {
+            // Evidence binds to the LATEST terminal turn: with harness-seeded
+            // follow-on turns, Send's own turn id is the first, not the last.
+            Ok(Some(turn_id)) if expected > 1 => active.turn_id = Some(turn_id),
+            Ok(turn_id) => {
                 if active.turn_id.is_none() {
-                    active.turn_id = Some(observation.event.turn_id);
+                    active.turn_id = turn_id;
                 }
             }
             Err(error) if deadline.is_expired() => {
@@ -41,7 +64,7 @@ impl ScenarioRunner<'_> {
             Err(error) => {
                 return Err(RunError::runner(
                     phase,
-                    "wait for harness::turn-completed signal",
+                    "wait for harness::turn-completed signal(s)",
                     error,
                 ));
             }
diff --git a/harness/tests/e2e/src/scenario/phases/evidence.rs b/harness/tests/e2e/src/scenario/phases/evidence.rs
index 51a07902e..911fbe4fe 100644
--- a/harness/tests/e2e/src/scenario/phases/evidence.rs
+++ b/harness/tests/e2e/src/scenario/phases/evidence.rs
@@ -98,9 +98,14 @@ impl ScenarioRunner<'_> {
         evidence: &RunEvidence,
         timed_out: bool,
     ) -> Result<(), RunError> {
-        let failure =
-            floor::floor_failure_for_turns(evidence, self.fixture.expected_terminal_turns)
-                .or_else(|| floor::verify_failure(self.fixture.verify, evidence));
+        let failure = floor::floor_failure_for(
+            evidence,
+            &floor::FloorExpectations {
+                turn_statuses: &self.fixture.expected_turn_statuses,
+                traces: self.fixture.expected_traces(),
+            },
+        )
+        .or_else(|| floor::verify_failure(self.fixture.verify, evidence));
         self.failure = failure.map(|message| evidence.scrub(&message));
 
         if timed_out {
diff --git a/harness/tests/e2e/src/scenarios/dsl.rs b/harness/tests/e2e/src/scenarios/dsl.rs
index 575851a0a..556d5ba98 100644
--- a/harness/tests/e2e/src/scenarios/dsl.rs
+++ b/harness/tests/e2e/src/scenarios/dsl.rs
@@ -9,8 +9,8 @@ use serde_json::{json, Value};
 use super::{ScenarioDriver, VerifyFn};
 use crate::fixtures::ScenarioFixture;
 use crate::types::frames::{
-    AssistantMessage, AssistantMessageEvent, AssistantRoleTag, ContentBlock, RouterChatResponse,
-    StopReason, Usage,
+    AssistantMessage, AssistantMessageEvent, AssistantRoleTag, ContentBlock, ErrorShape,
+    RouterChatResponse, StopReason, Usage,
 };
 use crate::types::probe::ControlledTargetV1;
 use crate::types::scenario::{
@@ -19,7 +19,7 @@ use crate::types::scenario::{
 };
 use crate::types::script::{
     GenerationMatchV1, JsonMatcherV1, JsonNormalizerV1, ModelFixtureV1, NormalizerOperation,
-    RouterScriptV1, SchemaVersion1, ScriptedGenerationV1,
+    RouterScriptV1, SchemaVersion1, ScriptedGenerationV1, ServeEffectV1, SteerSendV1,
 };
 
 const DEFAULT_SYSTEM_PROMPT: &str = include_str!("../../../../prompts/default.txt");
@@ -53,7 +53,7 @@ pub(super) struct Scenario {
     send: Option<Send>,
     target: Option<ControlledTargetV1>,
     generations: Vec<Generation>,
-    expected_terminal_turns: usize,
+    expected_turn_statuses: Vec<String>,
     verify: Option<VerifyFn>,
 }
 
@@ -74,7 +74,7 @@ impl Scenario {
             send: None,
             target: None,
             generations: Vec::new(),
-            expected_terminal_turns: 1,
+            expected_turn_statuses: vec!["completed".to_string()],
             verify: None,
         }
     }
@@ -96,7 +96,23 @@ impl Scenario {
 
     pub(super) fn terminal_turns(mut self, count: usize) -> Self {
         assert!(count > 0, "scenario must expect at least one terminal turn");
-        self.expected_terminal_turns = count;
+        self.expected_turn_statuses = vec!["completed".to_string(); count];
+        self
+    }
+
+    /// Declare each terminal turn's status, in completion order, when not
+    /// every turn completes (e.g. a failed turn whose finalize drain reseeds a
+    /// completing follow-on turn). The last turn must complete: the floor's
+    /// durable-status check has no meaning for a run that ends failed.
+    pub(super) fn terminal_turn_statuses<'a>(
+        mut self,
+        statuses: impl IntoIterator<Item = &'a str>,
+    ) -> Self {
+        self.expected_turn_statuses = statuses.into_iter().map(str::to_string).collect();
+        assert!(
+            !self.expected_turn_statuses.is_empty(),
+            "scenario must expect at least one terminal turn"
+        );
         self
     }
 
@@ -118,7 +134,8 @@ impl Scenario {
         ScenarioFixture {
             slug: self.slug,
             driver: self.driver,
-            expected_terminal_turns: self.expected_terminal_turns,
+            expected_terminal_turns: self.expected_turn_statuses.len(),
+            expected_turn_statuses: self.expected_turn_statuses,
             scenario: CompiledScenarioV1 {
                 schema_version: SchemaVersion1::V1,
                 id: self.id.clone(),
@@ -357,6 +374,8 @@ pub(super) struct Generation {
     ordinal: u64,
     request: Option<Request>,
     response: Option<Response>,
+    parked_message: Option<String>,
+    failure: Option<String>,
 }
 
 impl Generation {
@@ -365,6 +384,8 @@ impl Generation {
             ordinal,
             request: None,
             response: None,
+            parked_message: None,
+            failure: None,
         }
     }
 
@@ -378,11 +399,33 @@ impl Generation {
         self
     }
 
+    /// Before this generation resolves, steer the scenario session with
+    /// `text`. Because the harness is blocked awaiting this generation (turn
+    /// `Running`), the message parks durably in the turn's queue before the
+    /// generation's outcome — streamed frames or a scripted failure —
+    /// proceeds.
+    pub(super) fn parked_message(mut self, text: &str) -> Self {
+        self.parked_message = Some(text.to_string());
+        self
+    }
+
+    /// Instead of streaming, fail this generation with a handler error. The
+    /// harness treats a router handler error as permanent and finalizes the
+    /// turn `failed` WITHOUT the completed path's steering check — the only
+    /// deterministic public-path route into the finalize drain with a parked
+    /// message present (a park during a *completing* terminal generation is
+    /// always seen first by the steering check and delivered by an advance).
+    pub(super) fn fails(mut self, message: &str) -> Self {
+        self.failure = Some(message.to_string());
+        self
+    }
+
     fn compile(self, model: &ModelFixtureV1) -> ScriptedGenerationV1 {
-        let (frames, response) = self
-            .response
-            .expect("generation response is required")
-            .compile(model, self.ordinal);
+        let (frames, response) = match (self.response, &self.failure) {
+            (Some(response), None) => response.compile(model, self.ordinal),
+            (None, Some(message)) => (Vec::new(), failure_response(model, message)),
+            _ => panic!("generation needs exactly one of respond/fails"),
+        };
         ScriptedGenerationV1 {
             ordinal: self.ordinal,
             match_: self
@@ -391,6 +434,13 @@ impl Generation {
                 .compile(model, self.ordinal),
             frames,
             response,
+            on_serve: self.parked_message.map(|message| ServeEffectV1 {
+                steer: SteerSendV1 {
+                    session_id: "{{session_id}}".to_string(),
+                    message,
+                },
+            }),
+            failure: self.failure,
         }
     }
 }
@@ -554,6 +604,19 @@ impl Message {
         })
     }
 
+    /// The durable residue of a failed generation: the harness appends an empty
+    /// assistant row before streaming, and a scripted failure never fills it.
+    /// A later turn in the session sees it in its assembled context.
+    pub(super) fn assistant_empty(model: &ModelFixtureV1) -> Value {
+        json!({
+            "role": "assistant",
+            "content": [],
+            "stop_reason": "end",
+            "model": model.id,
+            "provider": model.provider
+        })
+    }
+
     pub(super) fn function_result(
         call_id: &str,
         function: &ControlledFunction,
@@ -578,6 +641,22 @@ impl Tool {
     }
 }
 
+/// The response slot of a scripted-failure generation. Never sent — the
+/// router fails the call before responding — but kept honest in the fixture.
+fn failure_response(model: &ModelFixtureV1, message: &str) -> RouterChatResponse {
+    RouterChatResponse {
+        ok: false,
+        provider: model.provider.clone(),
+        model: model.id.clone(),
+        stop_reason: None,
+        usage: None,
+        error: Some(ErrorShape {
+            code: "scripted_failure".to_string(),
+            message: message.to_string(),
+        }),
+    }
+}
+
 fn usage(input: u64, output: u64) -> Usage {
     Usage {
         input: Some(input),
diff --git a/harness/tests/e2e/src/scenarios/mod.rs b/harness/tests/e2e/src/scenarios/mod.rs
index 9388261a7..d7f92b72e 100644
--- a/harness/tests/e2e/src/scenarios/mod.rs
+++ b/harness/tests/e2e/src/scenarios/mod.rs
@@ -1,9 +1,10 @@
-//! The three checked-in integration fixtures.
+//! The checked-in integration fixtures.
 
 mod console_streamed_text;
 mod dsl;
 mod exactly_once_function;
 mod multi_turn_traces;
+mod reseed_parked_message;
 mod streamed_text;
 
 use crate::evidence_data::RunEvidence;
@@ -24,6 +25,7 @@ pub fn all() -> Vec<ScenarioFixture> {
         console_streamed_text::scenario(),
         exactly_once_function::scenario(),
         multi_turn_traces::scenario(),
+        reseed_parked_message::scenario(),
         streamed_text::scenario(),
     ]
 }
@@ -35,7 +37,7 @@ mod tests {
     #[test]
     fn every_fixture_is_unique_and_valid() {
         let fixtures = all();
-        assert_eq!(fixtures.len(), 4);
+        assert_eq!(fixtures.len(), 5);
         let mut slugs = std::collections::BTreeSet::new();
         let mut ids = std::collections::BTreeSet::new();
         for fixture in fixtures {
diff --git a/harness/tests/e2e/src/scenarios/reseed_parked_message.rs b/harness/tests/e2e/src/scenarios/reseed_parked_message.rs
new file mode 100644
index 000000000..fe8d38e94
--- /dev/null
+++ b/harness/tests/e2e/src/scenarios/reseed_parked_message.rs
@@ -0,0 +1,135 @@
+//! E2E-003 — a message that parks during a turn's final step is delivered by a
+//! reseeded turn.
+//!
+//! Regression guard for the harness finalize-drain reseed. The *completed*
+//! finalize cannot be pinned from the public boundary: a message parked while
+//! the terminal generation is in flight is seen by the loop's steering check
+//! and delivered by an advance — the drain's window only opens after that
+//! check, and no public actor can act inside it. The *failed* finalize has no
+//! steering check, so a generation that parks a steer and then fails routes
+//! the parked message deterministically through the finalize drain, which must
+//! reseed a turn to react to it (both finalize paths share the same drain +
+//! reseed). Without the reseed the parked message is stranded, no second
+//! terminal turn arrives, and the run times out.
+
+use super::dsl::{Generation, Message, Model, Request, Response, Scenario, Send};
+use super::ScenarioDriver;
+use crate::fixtures::ScenarioFixture;
+
+pub(super) fn scenario() -> ScenarioFixture {
+    const ID: &str = "E2E-003";
+    const MESSAGE: &str = "Answer the first question.";
+    const PARKED: &str = "Follow-up that parked during finalize.";
+    const SECOND_TEXT: &str = "handled the parked follow-up";
+
+    let model = Model::scripted("fixture-model");
+
+    Scenario::new(
+        ID,
+        "reseed-parked-message",
+        "A message parked during a turn's final step is delivered by a reseeded turn.",
+        ScenarioDriver::Direct,
+        model.clone(),
+    )
+    .send(Send::message(MESSAGE).idempotency_key("{{run_id}}:e2e-003"))
+    .terminal_turn_statuses(["failed", "completed"])
+    .generation(
+        // Turn 1's only step: the steer parks while the harness awaits this
+        // generation, then the scripted failure drives finalize_failed — whose
+        // drain delivers the parked row and must reseed.
+        Generation::new(1)
+            .expect(
+                Request::new()
+                    .turn_request_step(0)
+                    .system_prompt_sha256("{{system_prompt_sha256}}")
+                    .messages_exact([Message::user(MESSAGE)])
+                    .without_tools(),
+            )
+            .parked_message(PARKED)
+            .fails("scripted generation failure while the steered follow-up sits parked"),
+    )
+    .generation(
+        // The reseeded turn's own step 0 (`:0` pins a fresh turn id, not an
+        // advance of turn 1). Its context carries the original user message,
+        // the failed generation's empty assistant residue, and the parked
+        // follow-up delivered by the finalize drain.
+        Generation::new(2)
+            .expect(
+                Request::new()
+                    .turn_request_step(0)
+                    .system_prompt_sha256("{{system_prompt_sha256}}")
+                    .messages_exact([
+                        Message::user(MESSAGE),
+                        Message::assistant_empty(&model),
+                        Message::user(PARKED),
+                    ])
+                    .without_tools(),
+            )
+            .respond(Response::text(SECOND_TEXT, 12, 4)),
+    )
+    .verify(|run| {
+        // The empty assistant residue carries no text; only the reseeded
+        // turn's answer does.
+        run.expect_assistant_texts([SECOND_TEXT])?;
+        run.expect_message_counts(2, 2, 0)?;
+        run.expect_no_duplicate_messages()?;
+
+        // The follow-up must have PARKED (drained queue rows keep their
+        // durable `e_q_` entry id) — not landed as a direct send.
+        let parked_entry = run.transcript.iter().find(|item| {
+            item.get("message")
+                .and_then(|message| message.get("content"))
+                .and_then(|content| content.as_array())
+                .and_then(|blocks| blocks.first())
+                .and_then(|block| block.get("text"))
+                .and_then(|text| text.as_str())
+                == Some(PARKED)
+        });
+        let entry_id = parked_entry
+            .and_then(|item| item.get("entry_id"))
+            .and_then(|entry_id| entry_id.as_str())
+            .ok_or_else(|| anyhow::anyhow!("parked follow-up not found in transcript"))?;
+        anyhow::ensure!(
+            entry_id.starts_with("e_q_"),
+            "follow-up was not delivered from the queue: entry {entry_id}"
+        );
+        Ok(())
+    })
+    .build()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::types::script::JsonMatcherV1;
+
+    #[test]
+    fn parks_before_a_failing_terminal_generation_and_expects_a_reseeded_turn() {
+        let fixture = scenario();
+        assert_eq!(fixture.expected_terminal_turns, 2);
+        assert_eq!(fixture.expected_turn_statuses, ["failed", "completed"]);
+        assert_eq!(fixture.expected_traces(), 1);
+
+        // The parked steer rides generation 1, which then fails without
+        // streaming — the deterministic route into the finalize drain.
+        let first = &fixture.script.generations[0];
+        let effect = first
+            .on_serve
+            .as_ref()
+            .expect("generation 1 parks a message");
+        assert_eq!(effect.steer.session_id, "{{session_id}}");
+        assert!(!effect.steer.message.is_empty());
+        assert!(first.failure.is_some());
+        assert!(first.frames.is_empty());
+        assert!(!first.response.ok);
+
+        // Generation 2 is the reseeded turn's own step 0, with no side effect.
+        let second = &fixture.script.generations[1];
+        assert!(second.on_serve.is_none());
+        assert!(second.failure.is_none());
+        let JsonMatcherV1::Regex { pattern } = &second.match_.request_id else {
+            panic!("request id must be a regex");
+        };
+        assert!(pattern.ends_with(":0$"), "{pattern}");
+    }
+}
diff --git a/harness/tests/e2e/src/scripted_router.rs b/harness/tests/e2e/src/scripted_router.rs
index 39a6da3d1..94d9137fb 100644
--- a/harness/tests/e2e/src/scripted_router.rs
+++ b/harness/tests/e2e/src/scripted_router.rs
@@ -18,12 +18,13 @@ use std::sync::{Arc, Mutex};
 
 use iii_sdk::channel::{ChannelWriter, StreamChannelRef};
 use iii_sdk::errors::Error;
-use iii_sdk::RegisterFunction;
+use iii_sdk::protocol::TriggerRequest;
+use iii_sdk::{IIIClient, RegisterFunction};
 use serde_json::{json, Value};
 
-use crate::client::Client;
+use crate::client::{Client, DEFAULT_CALL_TIMEOUT_MS};
 use crate::matcher::{evaluate, FieldMatchResult};
-use crate::types::script::{ModelFixtureV1, RouterScriptV1, ScriptedGenerationV1};
+use crate::types::script::{ModelFixtureV1, RouterScriptV1, ScriptedGenerationV1, ServeEffectV1};
 
 /// One `router::chat` call's evidence, stored raw (never normalized).
 #[derive(Debug, Clone, serde::Serialize)]
@@ -86,13 +87,15 @@ impl ScriptedRouter {
         {
             let state = self.state.clone();
             let address = address.clone();
+            let client = iii.clone();
             iii.register_function(
                 "router::chat",
                 with_router_contract(
                     RegisterFunction::new_async(move |input: Value| {
                         let state = state.clone();
                         let address = address.clone();
-                        async move { chat(state, address, input).await }
+                        let client = client.clone();
+                        async move { chat(state, address, client, input).await }
                     }),
                     "router::chat",
                 )
@@ -359,7 +362,12 @@ fn model_supports(model: &ModelFixtureV1, capability: &str) -> bool {
     }
 }
 
-async fn chat(state: Arc<Mutex<State>>, address: String, input: Value) -> Result<Value, Error> {
+async fn chat(
+    state: Arc<Mutex<State>>,
+    address: String,
+    client: Arc<IIIClient>,
+    input: Value,
+) -> Result<Value, Error> {
     // Match under the lock; stream outside it.
     let (generation, writer_ref, request_id) = {
         let mut state = state.lock().expect("router state");
@@ -424,6 +432,22 @@ async fn chat(state: Arc<Mutex<State>>, address: String, input: Value) -> Result
         (generation, writer_ref, request_id)
     };
 
+    // Serve-time side effect, BEFORE streaming or failing: the harness is now
+    // blocked awaiting this generation with its turn `Running`, so an awaited
+    // steer parks in the turn's queue before the outcome proceeds.
+    if let Some(effect) = &generation.on_serve {
+        perform_serve_effect(&client, effect).await?;
+    }
+
+    // Scripted failure: the generation is consumed and recorded as matched;
+    // the error is the scripted subject behavior, not a contract violation.
+    if let Some(message) = &generation.failure {
+        state.lock().expect("router state").live.remove(&request_id);
+        return Err(Error::Handler(format!(
+            "integration/scripted_failure: {message}"
+        )));
+    }
+
     let stream_result =
         stream_frames(&address, &writer_ref, &generation, &state, &request_id).await;
 
@@ -435,6 +459,41 @@ async fn chat(state: Arc<Mutex<State>>, address: String, input: Value) -> Result
         .map_err(|e| Error::Handler(format!("integration/response_serialize: {e}")))
 }
 
+/// Run a generation's serve-time side effect. The steer is *awaited*:
+/// `harness::send` enqueues the parked row before it returns, so once this
+/// resolves the message is durably queued and the subsequent generation
+/// outcome (frames or scripted failure) drives the turn into finalize with
+/// that row present. Outer timeout as in `Client::call_with_timeout`: the
+/// SDK timeout covers the engine round-trip, but a connection that never
+/// establishes can park the future — and with it this `chat` handler.
+async fn perform_serve_effect(client: &IIIClient, effect: &ServeEffectV1) -> Result<(), Error> {
+    let steer = &effect.steer;
+    let outer = std::time::Duration::from_millis(DEFAULT_CALL_TIMEOUT_MS + 5_000);
+    match tokio::time::timeout(
+        outer,
+        client.trigger(TriggerRequest {
+            function_id: "harness::send".to_string(),
+            payload: json!({
+                "session_id": steer.session_id,
+                "message": steer.message,
+            }),
+            action: None,
+            timeout_ms: Some(DEFAULT_CALL_TIMEOUT_MS),
+        }),
+    )
+    .await
+    {
+        Ok(Ok(_)) => Ok(()),
+        Ok(Err(e)) => Err(Error::Handler(format!(
+            "integration/serve_effect harness::send: {e}"
+        ))),
+        Err(_) => Err(Error::Handler(format!(
+            "integration/serve_effect harness::send: no response within {}ms",
+            outer.as_millis()
+        ))),
+    }
+}
+
 async fn stream_frames(
     address: &str,
     writer_ref: &StreamChannelRef,
diff --git a/harness/tests/e2e/src/types/script.rs b/harness/tests/e2e/src/types/script.rs
index ab2862361..e3cabea7e 100644
--- a/harness/tests/e2e/src/types/script.rs
+++ b/harness/tests/e2e/src/types/script.rs
@@ -144,4 +144,40 @@ pub struct ScriptedGenerationV1 {
     pub match_: GenerationMatchV1,
     pub frames: Vec<AssistantMessageEvent>,
     pub response: RouterChatResponse,
+    /// Optional side effect the router performs *before* it streams this
+    /// generation's frames (or fails the call): an awaited steer send. Because
+    /// the harness is blocked awaiting this generation (its turn is `Running`),
+    /// the steered message parks durably in the turn's queue before the
+    /// generation's outcome proceeds.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub on_serve: Option<ServeEffectV1>,
+    /// When set, the router streams nothing: after `on_serve`, the call fails
+    /// with this handler-error message. The harness classifies a router
+    /// handler error as permanent and finalizes the turn `failed` WITHOUT the
+    /// completed path's steering check — the only deterministic public-path
+    /// route into the finalize drain with a parked message present, which the
+    /// reseed regression scenario requires. `frames` must be empty and
+    /// `response` is never sent.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub failure: Option<String>,
+}
+
+/// A serve-time side effect. Kept as a struct (not a bare field) so future
+/// effects can be added without another optional column.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
+#[serde(deny_unknown_fields)]
+pub struct ServeEffectV1 {
+    pub steer: SteerSendV1,
+}
+
+/// An awaited `harness::send` into a live session. The router issues it before
+/// streaming, so the message is durably enqueued before the frames (and thus
+/// the turn's finalize) proceed.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
+#[serde(deny_unknown_fields)]
+pub struct SteerSendV1 {
+    /// Target session, placeholder-expanded (e.g. `"{{session_id}}"`).
+    pub session_id: String,
+    /// User-message text steered into the session before streaming.
+    pub message: String,
 }