From 4958f8f9cb13597d2f936634e8a3d2de3f9a826a Mon Sep 17 00:00:00 2001 From: Anderson Leal Date: Thu, 23 Jul 2026 08:29:07 -0300 Subject: [PATCH 1/4] fix(harness): reseed a turn when the finalize drain delivers a parked message A notification or steer that parked during a turn's final step was drained to the transcript by the finalize drain but had no turn to react to it, so an autonomous run that ended its turn expecting the fire to wake it would stall. finalize_completed/finalize_failed now seed a fresh turn when the finalize drain delivered a model-visible row; custom-only rows never reseed and a cancelled turn drains but does not reseed. Gated on the drain actually delivering a row, so a redelivered finalize cannot double-seed and a racing external send is resolved by run_step's stale-turn guard. --- harness/src/functions/send.rs | 6 +- harness/src/turn_loop.rs | 162 ++++++++++++++++++++++++++++++---- 2 files changed, 151 insertions(+), 17 deletions(-) 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..0e74ceae4 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 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 { 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,59 @@ 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) +} + +/// 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 + } + } } -/// 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"); +/// 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 +1283,7 @@ async fn finalize_completed( record: &mut TurnRecord, result: Option, ) -> Result { - 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 +1313,12 @@ async fn finalize_completed( if let Some(parent) = record.parent.clone() { crate::deferred::resolve_parent(deps, &parent, "completed", result.as_ref(), None).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 +1334,7 @@ async fn finalize_failed( reason: &str, failure: FailureInfo, ) -> Result { - 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 +1415,13 @@ async fn finalize_failed( notify_parent_of_child_failure(deps, &parent.session_id, record, reason, failure).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 +1610,10 @@ async fn finalize_cancelled( record: &mut TurnRecord, reason: &str, ) -> Result { - 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(); @@ -2236,9 +2309,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(¬ice)), + 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(¬ification)), + 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() { From df2a992ba69ed587d90456792fda1a6ce2607cf3 Mon Sep 17 00:00:00 2001 From: Anderson Leal Date: Thu, 23 Jul 2026 10:22:40 -0300 Subject: [PATCH 2/4] test(harness): deterministic e2e coverage for the finalize-drain reseed (E2E-003) A message parked during a completing terminal generation is always delivered by the turn loop's steering check, so the completed-finalize drain cannot be reached deterministically from the public boundary. The failed finalize has no steering check and shares the same drain-and- reseed, so E2E-003 steers a message mid-generation and then fails that generation with a scripted router error: the failed finalize drains the parked row and must reseed a fresh turn, pinned by generation 2 matching a new turn's :0 request over [user, empty assistant residue, parked follow-up] and by the follow-up keeping its durable e_q_ queue entry id. With the reseed disabled the scenario times out. Framework support: scripted generation failures (fails()/failure), declared per-turn terminal statuses enforced positionally by the floor, driver-derived expected trace count (harness-seeded turns chain into the originating send's trace), and latest-terminal-turn evidence binding for multi-turn direct scenarios. --- harness/tests/e2e/README.md | 17 ++- harness/tests/e2e/src/fixtures/loading.rs | 48 ++++++- harness/tests/e2e/src/fixtures/tests.rs | 6 +- harness/tests/e2e/src/scenario/floor.rs | 132 +++++++++++++++--- .../e2e/src/scenario/phases/completion.rs | 33 ++++- .../tests/e2e/src/scenario/phases/evidence.rs | 11 +- harness/tests/e2e/src/scenarios/dsl.rs | 101 ++++++++++++-- harness/tests/e2e/src/scenarios/mod.rs | 6 +- .../src/scenarios/reseed_parked_message.rs | 132 ++++++++++++++++++ harness/tests/e2e/src/scripted_router.rs | 55 +++++++- harness/tests/e2e/src/types/script.rs | 36 +++++ 11 files changed, 521 insertions(+), 56 deletions(-) create mode 100644 harness/tests/e2e/src/scenarios/reseed_parked_message.rs 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, 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::>(); 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..757bc49ce 100644 --- a/harness/tests/e2e/src/scenario/floor.rs +++ b/harness/tests/e2e/src/scenario/floor.rs @@ -10,18 +10,40 @@ 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 { - 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 { +pub fn floor_failure_for(run: &RunEvidence, expected: &FloorExpectations) -> Option { 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 +90,28 @@ fn status_list_len(run: &RunEvidence, key: &str) -> usize { .unwrap_or(0) } -fn trace_failure(run: &RunEvidence, expected_terminal_turns: usize) -> Option { +fn trace_failure(run: &RunEvidence, expected: &FloorExpectations) -> Option { 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 +120,8 @@ fn trace_failure(run: &RunEvidence, expected_terminal_turns: usize) -> Option Option { +fn lifecycle_failure(run: &RunEvidence, expected: &FloorExpectations) -> Option { + let expected_terminal_turns = expected.turns(); let lifecycle_name = format!("execute {LIFECYCLE_FUNCTION_ID}"); let lifecycle: Vec> = run .spans_named(&lifecycle_name) @@ -134,14 +166,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 +291,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 +375,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 +391,41 @@ 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..5669ff352 100644 --- a/harness/tests/e2e/src/scenario/phases/completion.rs +++ b/harness/tests/e2e/src/scenario/phases/completion.rs @@ -24,10 +24,35 @@ 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 +66,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, target: Option, generations: Vec, - expected_terminal_turns: usize, + expected_turn_statuses: Vec, verify: Option, } @@ -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, + ) -> 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, response: Option, + parked_message: Option, + failure: Option, } 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 { 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..f7eb85d2c --- /dev/null +++ b/harness/tests/e2e/src/scenarios/reseed_parked_message.rs @@ -0,0 +1,132 @@ +//! 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..6fa1e9b21 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>, address: String, input: Value) -> Result { +async fn chat( + state: Arc>, + address: String, + client: Arc, + input: Value, +) -> Result { // 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>, 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,27 @@ async fn chat(state: Arc>, 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 terminal frame +/// will drive the turn into finalize with that row present. +async fn perform_serve_effect(client: &IIIClient, effect: &ServeEffectV1) -> Result<(), Error> { + let steer = &effect.steer; + 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 + .map(|_| ()) + .map_err(|e| Error::Handler(format!("integration/serve_effect harness::send: {e}"))) +} + 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, 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, + /// 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, +} + +/// 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, } From bbb8a38172b06d4f7049045b79771dddb2c12bca Mon Sep 17 00:00:00 2001 From: Anderson Leal Date: Thu, 23 Jul 2026 14:50:00 -0300 Subject: [PATCH 3/4] style(harness): apply cargo fmt --- harness/src/turn_loop.rs | 6 +---- harness/tests/e2e/src/scenario/floor.rs | 22 ++++++++++++++----- .../e2e/src/scenario/phases/completion.rs | 4 +--- .../src/scenarios/reseed_parked_message.rs | 5 ++++- 4 files changed, 22 insertions(+), 15 deletions(-) diff --git a/harness/src/turn_loop.rs b/harness/src/turn_loop.rs index 0e74ceae4..52947df28 100644 --- a/harness/src/turn_loop.rs +++ b/harness/src/turn_loop.rs @@ -1105,11 +1105,7 @@ async fn drain_queued( /// 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 { +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) => { diff --git a/harness/tests/e2e/src/scenario/floor.rs b/harness/tests/e2e/src/scenario/floor.rs index 757bc49ce..195052e45 100644 --- a/harness/tests/e2e/src/scenario/floor.rs +++ b/harness/tests/e2e/src/scenario/floor.rs @@ -24,7 +24,9 @@ impl FloorExpectations<'_> { } fn declares_failure(&self) -> bool { - self.turn_statuses.iter().any(|status| status != "completed") + self.turn_statuses + .iter() + .any(|status| status != "completed") } } @@ -392,7 +394,10 @@ mod tests { }); evidence.traces = TraceEvidenceV1::new(evidence.traces.traces); let statuses = vec!["completed".to_string(), "completed".to_string()]; - assert_eq!(floor_failure_for(&evidence, &expectations(&statuses, 2)), None); + assert_eq!( + floor_failure_for(&evidence, &expectations(&statuses, 2)), + None + ); } /// The reseed shape: one send, one trace, a failed turn whose finalize @@ -412,13 +417,18 @@ mod tests { ], }]); let statuses = vec!["failed".to_string(), "completed".to_string()]; - assert_eq!(floor_failure_for(&evidence, &expectations(&statuses, 1)), None); + 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")); + 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()]; diff --git a/harness/tests/e2e/src/scenario/phases/completion.rs b/harness/tests/e2e/src/scenario/phases/completion.rs index 5669ff352..0affc4d3e 100644 --- a/harness/tests/e2e/src/scenario/phases/completion.rs +++ b/harness/tests/e2e/src/scenario/phases/completion.rs @@ -36,9 +36,7 @@ impl ScenarioRunner<'_> { .probe() .wait_for_completion_turns(expected, deadline) .await - .map(|events| { - latest_terminal_observation(&events).map(|o| o.event.turn_id.clone()) - }) + .map(|events| latest_terminal_observation(&events).map(|o| o.event.turn_id.clone())) } else { services .probe() diff --git a/harness/tests/e2e/src/scenarios/reseed_parked_message.rs b/harness/tests/e2e/src/scenarios/reseed_parked_message.rs index f7eb85d2c..fe8d38e94 100644 --- a/harness/tests/e2e/src/scenarios/reseed_parked_message.rs +++ b/harness/tests/e2e/src/scenarios/reseed_parked_message.rs @@ -113,7 +113,10 @@ mod tests { // 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"); + 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()); From a3191405ff056ef16a3b1fc75d972ab60e779191 Mon Sep 17 00:00:00 2001 From: Anderson Leal Date: Thu, 23 Jul 2026 15:07:25 -0300 Subject: [PATCH 4/4] fix(harness): close the enqueue-after-drain window with a post-terminal sweep Review follow-up (CodeRabbit). A send racing a finalize could enqueue after the finalize drain and recheck the turn record before the terminal write landed, concluding queued against a turn that would never drain again. A second drain sweep after the terminal put_turn pairs with try_enqueue's post-enqueue recheck: a recheck that still saw Running implies the row landed before the sweep; a recheck that sees the terminal record seeds its own turn. Applied to completed, failed, and cancelled finalize (the latter still drain-only). Also wrap the scripted router's steer trigger in the same outer timeout Client::call_with_timeout uses, so a connection that never establishes cannot park the chat handler. --- harness/src/turn_loop.rs | 14 +++++++++++ harness/tests/e2e/src/scripted_router.rs | 30 +++++++++++++++++------- 2 files changed, 36 insertions(+), 8 deletions(-) diff --git a/harness/src/turn_loop.rs b/harness/src/turn_loop.rs index 52947df28..3165bc71e 100644 --- a/harness/src/turn_loop.rs +++ b/harness/src/turn_loop.rs @@ -1309,6 +1309,13 @@ 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). @@ -1411,6 +1418,9 @@ 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 @@ -1648,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, diff --git a/harness/tests/e2e/src/scripted_router.rs b/harness/tests/e2e/src/scripted_router.rs index 6fa1e9b21..94d9137fb 100644 --- a/harness/tests/e2e/src/scripted_router.rs +++ b/harness/tests/e2e/src/scripted_router.rs @@ -461,12 +461,17 @@ async fn chat( /// 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 terminal frame -/// will drive the turn into finalize with that row present. +/// 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; - client - .trigger(TriggerRequest { + 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, @@ -474,10 +479,19 @@ async fn perform_serve_effect(client: &IIIClient, effect: &ServeEffectV1) -> Res }), action: None, timeout_ms: Some(DEFAULT_CALL_TIMEOUT_MS), - }) - .await - .map(|_| ()) - .map_err(|e| Error::Handler(format!("integration/serve_effect harness::send: {e}"))) + }), + ) + .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(