(MOT-4269) fix(harness): harden reactive E2E validation

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 {

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<String>,
     pub(super) max_parallel_calls: usize,
+    pub(super) max_concurrent_sessions: usize,
     pub(super) sessions_in_tree: BTreeSet<String>,
 }
 

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

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<BTreeMap<String, ActivityWindow>> {
+    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<ActivityWindow> {
+    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<String, ActivityWindow>) -> 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);
+    }
 }