From 58b5c8918fe4aeffa6f0417ce7c74e1c93591df1 Mon Sep 17 00:00:00 2001 From: Ytallo Layon Date: Wed, 29 Jul 2026 08:37:43 -0300 Subject: [PATCH] (MOT-4269) fix(harness): harden reactive E2E validation --- .../scenarios/reactive_automation/evaluate.rs | 6 +- .../scenarios/reactive_automation/evidence.rs | 1 + .../scenarios/reactive_automation/prompt.rs | 27 +++- .../scenarios/reactive_automation/queries.rs | 144 +++++++++++++++++- 4 files changed, 170 insertions(+), 8 deletions(-) diff --git a/harness/tests/e2e/src/scenarios/reactive_automation/evaluate.rs b/harness/tests/e2e/src/scenarios/reactive_automation/evaluate.rs index 8cee016c7..2335f6aa8 100644 --- a/harness/tests/e2e/src/scenarios/reactive_automation/evaluate.rs +++ b/harness/tests/e2e/src/scenarios/reactive_automation/evaluate.rs @@ -27,7 +27,8 @@ pub(super) fn score(evidence: &Evidence, names: &ScenarioNames) -> ObjectiveEval let parallel_writers = evidence.writer_spawns.call_count == EXPECTED_WRITERS && evidence.writer_spawns.session_ids == expected_writer_sessions - && evidence.writer_spawns.max_parallel_calls == EXPECTED_WRITERS + && (evidence.writer_spawns.max_parallel_calls == EXPECTED_WRITERS + || evidence.writer_spawns.max_concurrent_sessions == EXPECTED_WRITERS) && expected_writer_sessions .iter() .all(|session| evidence.writer_spawns.sessions_in_tree.contains(session)); @@ -247,7 +248,8 @@ mod tests { writer_spawns: WriterSpawnEvidence { call_count: 3, session_ids: expected_sessions, - max_parallel_calls: 3, + max_parallel_calls: 1, + max_concurrent_sessions: 3, sessions_in_tree, }, watch: WatchEvidence { diff --git a/harness/tests/e2e/src/scenarios/reactive_automation/evidence.rs b/harness/tests/e2e/src/scenarios/reactive_automation/evidence.rs index ddbeeaf57..c52e14e67 100644 --- a/harness/tests/e2e/src/scenarios/reactive_automation/evidence.rs +++ b/harness/tests/e2e/src/scenarios/reactive_automation/evidence.rs @@ -24,6 +24,7 @@ pub(super) struct WriterSpawnEvidence { pub(super) call_count: usize, pub(super) session_ids: BTreeSet, pub(super) max_parallel_calls: usize, + pub(super) max_concurrent_sessions: usize, pub(super) sessions_in_tree: BTreeSet, } diff --git a/harness/tests/e2e/src/scenarios/reactive_automation/prompt.rs b/harness/tests/e2e/src/scenarios/reactive_automation/prompt.rs index 576950412..750c27d2e 100644 --- a/harness/tests/e2e/src/scenarios/reactive_automation/prompt.rs +++ b/harness/tests/e2e/src/scenarios/reactive_automation/prompt.rs @@ -35,10 +35,14 @@ Required execution: 4. Each notification must wake a trigger-spawned reactor session namespaced with `{run_label}`. Reactors must recompute and upsert `{totals}` from `{orders}` so replaying an event cannot - double-count. + double-count. Use per-writer upserts; do not delete all totals before reinserting them, because + overlapping notifications would expose a transient empty or partial aggregate table. -5. Once all writers are done and totals cover 15 orders, a trigger-spawned finalizer in - `{finalizer}` must write exactly one row to `{report}` with: +5. Before spawning writers, arm a namespaced completion reaction covering all three exact writer + sessions. It must evaluate finalization only after every writer has marked itself `done`; do + not rely only on an order-insert notification, because the final insert happens before that + writer's status update. Once all writers are done and totals cover 15 orders, a + trigger-spawned finalizer in `{finalizer}` must write exactly one row to `{report}` with: `run_id`, `watch_mechanism`, `fallback_reason`, `events_received`, `rows_written`, `elapsed_ms`, `totals_match`, `no_notification_loss`, `no_double_counting`, @@ -48,8 +52,21 @@ Required execution: Use `{run_label}` as `run_id` and `{finalizer}` as `finalizer_session_id`. Report the actual watch mechanism, a non-empty fallback reason when fallback is used, 15 events and rows, positive elapsed milliseconds, and one trigger-spawned reactor session and spawning event. - -6. Unregister every trigger and subscription created for this run, then wake the root session. + Record a numeric start time before the writers run so the finalizer can compute `elapsed_ms`; + if exact clock arithmetic is unavailable, use a conservative positive elapsed estimate rather + than zero. Before signaling completion, query the single report row and verify + `events_received = 15`, `rows_written = 15`, and `elapsed_ms > 0`; correct the row if any of + those checks fails. Also verify that `totals_match`, `no_notification_loss`, + `no_double_counting`, `trigger_spawned_reactor`, and `no_inline_waiting` are all true. The + finalizer must recompute and upsert the complete 15-order totals before this comparison so it + does not record a transient reactor result. + +6. After the trigger-spawned finalizer writes the report, wake the existing root session + explicitly. Set the wake reaction's `metadata.session_id` to the current root session id; + setting only `parent_session_id` spawns a new unnamed child and does not wake the root. In the + resumed root turn, unregister every trigger and subscription created for this run, then list + the registered triggers and verify that none contains `{run_label}` or `{namespace}`. Do not + give the final response before the root has resumed and this cleanup check has passed. This is a deliberately large execution. A deadline is only a stuck-execution watchdog, not a normal completion deadline. If you register a watchdog, use at least {watchdog_seconds} seconds diff --git a/harness/tests/e2e/src/scenarios/reactive_automation/queries.rs b/harness/tests/e2e/src/scenarios/reactive_automation/queries.rs index 1a25bbb5f..354232283 100644 --- a/harness/tests/e2e/src/scenarios/reactive_automation/queries.rs +++ b/harness/tests/e2e/src/scenarios/reactive_automation/queries.rs @@ -1,4 +1,4 @@ -use std::collections::BTreeSet; +use std::collections::{BTreeMap, BTreeSet}; use anyhow::Context; use serde_json::{json, Value}; @@ -43,6 +43,9 @@ pub(super) async fn collect( }) .collect(), max_parallel_calls: max_parallel_calls(&observation.transcript, "harness::spawn"), + max_concurrent_sessions: max_concurrent_sessions( + &writer_activity_windows(context, &names.writer_sessions).await?, + ), sessions_in_tree, }; @@ -308,6 +311,67 @@ fn max_parallel_calls(transcript: &Value, function_id: &str) -> usize { .unwrap_or(0) } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct ActivityWindow { + started_at: i64, + finished_at: i64, +} + +async fn writer_activity_windows( + context: &E2eContext, + session_ids: &[String], +) -> anyhow::Result> { + let mut windows = BTreeMap::new(); + for session_id in session_ids { + let transcript = context.transcript(session_id).await?; + if let Some(window) = activity_window(&transcript) { + windows.insert(session_id.clone(), window); + } + } + Ok(windows) +} + +fn activity_window(transcript: &Value) -> Option { + let mut timestamps = transcript + .get("messages") + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter_map(|entry| entry.get("message")) + .filter(|message| { + matches!( + message.get("role").and_then(Value::as_str), + Some("assistant" | "function_result") + ) + }) + .filter_map(|message| message.get("timestamp").and_then(Value::as_i64)); + let first = timestamps.next()?; + let (started_at, finished_at) = + timestamps.fold((first, first), |(started_at, finished_at), timestamp| { + (started_at.min(timestamp), finished_at.max(timestamp)) + }); + Some(ActivityWindow { + started_at, + finished_at, + }) +} + +fn max_concurrent_sessions(windows: &BTreeMap) -> usize { + windows + .values() + .map(|candidate| { + windows + .values() + .filter(|window| { + window.started_at <= candidate.started_at + && candidate.started_at < window.finished_at + }) + .count() + }) + .max() + .unwrap_or(0) +} + #[cfg(test)] mod tests { use super::*; @@ -340,4 +404,82 @@ mod tests { }); assert_eq!(max_parallel_calls(&transcript, "harness::spawn"), 3); } + + #[test] + fn activity_window_uses_model_execution_messages() { + let transcript = json!({ + "messages": [ + {"message": {"role": "user", "timestamp": 1}}, + {"message": {"role": "assistant", "timestamp": 10}}, + {"message": {"role": "function_result", "timestamp": 20}}, + {"message": {"role": "assistant", "timestamp": 30}} + ] + }); + + assert_eq!( + activity_window(&transcript), + Some(ActivityWindow { + started_at: 10, + finished_at: 30, + }) + ); + } + + #[test] + fn concurrent_session_detection_accepts_overlapping_writer_activity() { + let windows = BTreeMap::from([ + ( + "writer-1".to_string(), + ActivityWindow { + started_at: 10, + finished_at: 40, + }, + ), + ( + "writer-2".to_string(), + ActivityWindow { + started_at: 20, + finished_at: 50, + }, + ), + ( + "writer-3".to_string(), + ActivityWindow { + started_at: 30, + finished_at: 60, + }, + ), + ]); + + assert_eq!(max_concurrent_sessions(&windows), 3); + } + + #[test] + fn concurrent_session_detection_rejects_non_overlapping_writer_activity() { + let windows = BTreeMap::from([ + ( + "writer-1".to_string(), + ActivityWindow { + started_at: 10, + finished_at: 20, + }, + ), + ( + "writer-2".to_string(), + ActivityWindow { + started_at: 20, + finished_at: 30, + }, + ), + ( + "writer-3".to_string(), + ActivityWindow { + started_at: 30, + finished_at: 40, + }, + ), + ]); + + assert_eq!(max_concurrent_sessions(&windows), 1); + } }