diff --git a/.github/scripts/discover_changed_workers.py b/.github/scripts/discover_changed_workers.py
index 74d4b7f01..b20b88b23 100644
--- a/.github/scripts/discover_changed_workers.py
+++ b/.github/scripts/discover_changed_workers.py
@@ -89,8 +89,14 @@
     ".github/workflows/harness-e2e-main.yml",
     ".github/workflows/harness-e2e-nightly.yml",
 }
-INTEGRATION_EXCLUDED_PREFIXES = ("harness/tests/e2e/",)
-E2E_EXCLUDED_PREFIXES = ("harness/tests/integration/",)
+INTEGRATION_EXCLUDED_PREFIXES = (
+    "harness/tests/e2e/",
+    "harness/tests/quickstart/",
+)
+E2E_EXCLUDED_PREFIXES = (
+    "harness/tests/integration/",
+    "harness/tests/quickstart/",
+)
 
 # Shared Rust crates live under crates/<name>/ (no iii.worker.yaml — not
 # workers). A source change there (1) reports the crate in the `crates`
diff --git a/.github/scripts/tests/test_discover_changed_workers.py b/.github/scripts/tests/test_discover_changed_workers.py
index 92763e112..1fd2073f7 100644
--- a/.github/scripts/tests/test_discover_changed_workers.py
+++ b/.github/scripts/tests/test_discover_changed_workers.py
@@ -349,6 +349,24 @@ def test_integration_suite_change_does_not_run_e2e(self, tmp_path):
         assert data["integration_changed"] is True
         assert data["e2e_changed"] is False
 
+    def test_quickstart_change_does_not_run_e2e(self, tmp_path):
+        repo = make_repo_with_harness(tmp_path)
+        path = repo / "harness" / "tests" / "quickstart" / "run-ci.sh"
+        path.parent.mkdir(parents=True)
+        path.write_text("#!/usr/bin/env bash\n")
+        subprocess.run(["git", "add", "."], cwd=repo, check=True, env=GIT_HERMETIC_ENV)
+        subprocess.run(
+            ["git", "commit", "-q", "-m", "quickstart"],
+            cwd=repo,
+            check=True,
+            env=GIT_HERMETIC_ENV,
+        )
+        r = run_script(repo, "main~1")
+        assert r.returncode == 0, r.stderr
+        data = json.loads(r.stdout)
+        assert data["integration_changed"] is False
+        assert data["e2e_changed"] is False
+
     def test_integration_workflow_change_runs_integration(self, tmp_path):
         repo = make_repo_with_harness(tmp_path)
         workflow = repo / ".github" / "workflows" / "ci.yml"
diff --git a/harness/src/functions/metrics.rs b/harness/src/functions/metrics.rs
index a7e047897..fc4b63f82 100644
--- a/harness/src/functions/metrics.rs
+++ b/harness/src/functions/metrics.rs
@@ -1,6 +1,6 @@
 //! `harness::metrics` — aggregate durable model usage and function outcomes
-//! plus trace/span observability over one complete root-and-descendant session
-//! tree.
+//! plus trace/span observability over a root-and-descendant session tree.
+//! `complete=false` means the returned aggregates are a live partial snapshot.
 
 use std::collections::{BTreeSet, HashMap};
 
@@ -126,25 +126,24 @@ pub async fn handle(
     req: SessionMetricsRequestV1,
 ) -> Result<SessionMetricsResponseV1, HarnessError> {
     let tree = session_tree::collect(deps, &req.root_session_id).await?;
-    if !tree.complete {
-        return Ok(incomplete(&req.root_session_id));
-    }
-
     let session = deps.session().await;
     let cfg = deps.cfg().await;
+    let mut complete = tree.complete;
     let mut total = UsageAccumulator::default();
     let mut by_session = Vec::with_capacity(tree.sessions.len());
     for node in &tree.sessions {
         if !session.exists(&node.session_id).await? {
-            return Ok(incomplete(&req.root_session_id));
+            complete = false;
+            continue;
         }
         let Some(turn) =
             crate::state::get_turn(&deps.iii, &node.session_id, cfg.session_timeout_ms).await?
         else {
-            return Ok(incomplete(&req.root_session_id));
+            complete = false;
+            continue;
         };
         if !terminal_status(Some(turn.status)) {
-            return Ok(incomplete(&req.root_session_id));
+            complete = false;
         }
         let entries = session.messages_strict(&node.session_id).await?;
         let mut current = UsageAccumulator::default();
@@ -156,11 +155,18 @@ pub async fn handle(
         }
         by_session.push(current.finish_session(node));
     }
-    let traces = collect_trace_metrics(deps, &tree.sessions, &req.root_session_id).await;
+    // Partial snapshots are polled as progress signals. Trace aggregation is
+    // comparatively expensive and does not help the watchdog decide whether
+    // work advanced, so collect it only for the final complete response.
+    let traces = if complete {
+        collect_trace_metrics(deps, &tree.sessions, &req.root_session_id).await
+    } else {
+        None
+    };
 
     Ok(SessionMetricsResponseV1 {
         root_session_id: req.root_session_id,
-        complete: true,
+        complete,
         totals: total.finish_totals(tree.sessions.len() as u64),
         by_session,
         traces,
@@ -171,16 +177,6 @@ fn terminal_status(status: Option<crate::types::turn::TurnStatus>) -> bool {
     status.is_some_and(crate::types::turn::TurnStatus::is_terminal)
 }
 
-fn incomplete(root_session_id: &str) -> SessionMetricsResponseV1 {
-    SessionMetricsResponseV1 {
-        root_session_id: root_session_id.to_string(),
-        complete: false,
-        totals: SessionUsageTotalsV1::default(),
-        by_session: Vec::new(),
-        traces: None,
-    }
-}
-
 async fn collect_trace_metrics(
     deps: &Deps,
     sessions: &[SessionTreeNodeV1],
@@ -486,11 +482,33 @@ mod tests {
     }
 
     #[test]
-    fn incomplete_metrics_never_contain_partial_values() {
-        let response = incomplete("root");
+    fn incomplete_metrics_can_preserve_observed_values() {
+        let mut usage = UsageAccumulator::default();
+        usage.observe(&message(json!({
+            "role": "assistant",
+            "content": [{
+                "type": "function_call",
+                "id": "call-1",
+                "function_id": "state::get",
+                "arguments": {}
+            }],
+            "stop_reason": "function_call",
+            "usage": {"input": 10, "output": 2},
+            "model": "model",
+            "provider": "provider",
+            "timestamp": 1
+        })));
+        let response = SessionMetricsResponseV1 {
+            root_session_id: "root".into(),
+            complete: false,
+            totals: usage.finish_totals(1),
+            by_session: Vec::new(),
+            traces: None,
+        };
         assert!(!response.complete);
         assert!(response.by_session.is_empty());
-        assert_eq!(response.totals, SessionUsageTotalsV1::default());
+        assert_eq!(response.totals.function_calls, 1);
+        assert_eq!(response.totals.input_tokens, Some(10));
         assert_eq!(response.traces, None);
     }
 
diff --git a/harness/src/functions/react.rs b/harness/src/functions/react.rs
index 54fb18e51..b86f63750 100644
--- a/harness/src/functions/react.rs
+++ b/harness/src/functions/react.rs
@@ -32,10 +32,11 @@
 //! by design; deployments additionally deny it in their permissions policy
 //! (see the repo's iii-permissions.yaml conventions).
 //!
-//! ponytail: a trigger fires with no live parent turn, so spawned sub-agents
-//! are unparented (depth 0). Emergent joins are fixed-arity (`expect` lists the
-//! predecessors) — no fan-out over arrays, no retries, no central run record;
-//! that heavier machinery stays in the `workflow` worker.
+//! ponytail: a trigger fires with no live parent turn, so reactive children
+//! inherit policy and budgets from the registering session through a trusted
+//! owner stamp rather than a live parent call. Emergent joins are fixed-arity
+//! (`expect` lists the predecessors) — no fan-out over arrays, no retries, no
+//! central run record; that heavier machinery stays in the `workflow` worker.
 
 use iii_helpers::observability::opentelemetry::trace::{Status, TraceContextExt as _};
 use iii_helpers::observability::opentelemetry::{Context, KeyValue};
@@ -1219,6 +1220,34 @@ pub(crate) fn kind_name(v: &Value) -> &'static str {
     }
 }
 
+async fn notify_owner_of_reaction_failure(deps: &Deps, spec: &ReactSpec, reason: &str) {
+    let Some(owner_session_id) = spec.owner_session_id.as_deref() else {
+        return;
+    };
+    let subscription = spec.subscription_id.as_deref().unwrap_or("untracked");
+    let message = format!(
+        "[reaction-failure] Reactive execution `{subscription}` failed and could not make \
+         progress. Reason: {reason}. Inspect or replace the binding, then clean up its trigger."
+    );
+    let req = crate::functions::send::SendRequest {
+        session_id: Some(owner_session_id.to_string()),
+        message: crate::functions::send::MessageInput::Text(message),
+        model: None,
+        provider: None,
+        idempotency_key: Some(format!("reaction_failure_{subscription}")),
+        session: None,
+        options: None,
+    };
+    if let Err(error) = crate::functions::send::handle(deps, req).await {
+        tracing::warn!(
+            owner_session = %owner_session_id,
+            subscription,
+            error = %error,
+            "reaction failure notification failed"
+        );
+    }
+}
+
 async fn spawn_reaction(
     deps: &Deps,
     task: String,
@@ -1228,25 +1257,22 @@ async fn spawn_reaction(
 ) -> Result<ReactResult, HarnessError> {
     let mut payload = build_spawn_payload(task, spec, parent.as_deref());
     payload["reactive_depth"] = json!(reactive_depth);
-    match deps
-        .iii
-        .trigger(TriggerRequest {
-            function_id: super::SPAWN_ID.to_string(),
-            payload,
-            action: None,
-            timeout_ms: None,
-        })
-        .await
-    {
-        Ok(v) => {
-            let child = v
-                .get("child_session_id")
-                .and_then(Value::as_str)
-                .map(str::to_string);
-            let turn = v
-                .get("child_turn_id")
-                .and_then(Value::as_str)
-                .map(str::to_string);
+    let req: crate::functions::spawn::SpawnRequest =
+        serde_json::from_value(payload).map_err(|error| {
+            HarnessError::InvalidRequest(format!("invalid reaction spawn: {error}"))
+        })?;
+    let spawned = match spec.owner_session_id.as_deref() {
+        Some(owner_session_id) => {
+            crate::subagent::spawn_reactive_child(deps, &req, owner_session_id).await
+        }
+        // Raw engine registrations have no harness turn to inherit from.
+        // Preserve their historical parentless behavior.
+        None => crate::subagent::spawn_child(deps, &req, None).await,
+    };
+    match spawned {
+        Ok(ids) => {
+            let child = Some(ids.session_id);
+            let turn = Some(ids.turn_id);
             // Reaction sessions may need to clean up their own run while the
             // registrant is parked on a wake: record child → registrant so
             // the unregister ownership check accepts lineage descendants.
@@ -1264,7 +1290,7 @@ async fn spawn_reaction(
             Ok(ReactResult::spawned(child, turn))
         }
         Err(e) => {
-            tracing::warn!(error = %e, "harness::react: harness::spawn dispatch failed");
+            tracing::warn!(error = %e, "harness::react: reactive child spawn failed");
             Ok(ReactResult::note(format!("spawn failed: {e}")))
         }
     }
@@ -1333,6 +1359,10 @@ async fn record_reaction_outcome(
         );
     }
 
+    if status == "failed" {
+        notify_owner_of_reaction_failure(deps, spec, summary).await;
+    }
+
     // Outcomes belong in the chat that wired the pipeline even when the
     // downstream turn is explicitly delivered somewhere else. Raw external
     // registrations have no owner stamp, so fall back to their target session.
@@ -1710,13 +1740,11 @@ fn build_spawn_payload(task: String, spec: &ReactSpec, parent_session_id: Option
     if let Some(o) = &spec.options {
         payload["options"] = o.clone();
     }
-    // Interceptor-registered reactions inherit the REGISTRANT's dispatch
-    // policy, subset when the spec narrows it (the in-turn child rule).
-    // Without this the spawned turn is parentless and lands on the read-only
-    // baseline — a wrap-up reaction delivered into the registrant's own chat
-    // suddenly can't call what every earlier turn there could (rctest-k7m3:
-    // database::query / state::set / engine::unregister_trigger all denied).
-    // Raw engine-side registrations carry no stamp and keep the baseline.
+    // Preserve the interceptor-stamped effective policy in the spawn request.
+    // The trusted reactive spawn path also loads the durable owner record and
+    // subsets against it, so policy and budget inheritance share one source
+    // of truth. Raw engine-side registrations carry no stamp or owner and keep
+    // the configured parentless baseline.
     if let Some(registrant) = &spec.registrant_functions {
         let requested: Option<crate::types::turn::FunctionPolicy> = spec
             .options
diff --git a/harness/src/functions/send.rs b/harness/src/functions/send.rs
index ea62630dc..ddfc3ae24 100644
--- a/harness/src/functions/send.rs
+++ b/harness/src/functions/send.rs
@@ -685,6 +685,7 @@ pub(crate) async fn seed_new(
         display_parent_session_id: None,
         spawned_by_subscription_id: None,
         reactive_depth: None,
+        reactive_owner_session_id: None,
         functions_generation,
         result: None,
         result_error: None,
diff --git a/harness/src/subagent.rs b/harness/src/subagent.rs
index 1cbda0839..eedebfb84 100644
--- a/harness/src/subagent.rs
+++ b/harness/src/subagent.rs
@@ -87,7 +87,7 @@ pub async fn spawn_from_turn(
         turn_id: parent.turn_id.clone(),
         function_call_id: call_id.to_string(),
     };
-    seed_child(deps, &cfg, &req, Some(&parent_link), Some(parent))
+    seed_child(deps, &cfg, &req, Some(&parent_link), Some(parent), None)
         .await
         .map_err(|e| is_error(e.code(), e.to_string()))
 }
@@ -124,7 +124,29 @@ pub async fn spawn_child(
     parent: Option<&ParentLink>,
 ) -> Result<ChildIds, HarnessError> {
     let cfg = deps.cfg().await;
-    seed_child(deps, &cfg, req, parent, None).await
+    seed_child(deps, &cfg, req, parent, None, None).await
+}
+
+/// Trusted spawn path for a reaction registered by a harness turn.
+///
+/// Reactive children remain fire-and-forget and therefore do not get a
+/// `ParentLink`, but they inherit the registering turn's policy, filesystem
+/// scope, model limits, and shared token/cost ledger. The durable owner id is
+/// also retained so terminal failures wake the orchestration session.
+pub async fn spawn_reactive_child(
+    deps: &Deps,
+    req: &SpawnRequest,
+    owner_session_id: &str,
+) -> Result<ChildIds, HarnessError> {
+    let cfg = deps.cfg().await;
+    let owner = crate::state::get_turn(&deps.iii, owner_session_id, cfg.session_timeout_ms)
+        .await?
+        .ok_or_else(|| {
+            HarnessError::InvalidRequest(format!(
+                "reaction owner session `{owner_session_id}` has no durable turn record"
+            ))
+        })?;
+    seed_child(deps, &cfg, req, None, Some(&owner), Some(owner_session_id)).await
 }
 
 /// Resolve a child's dispatch policy. An in-turn child inherits the PARENT'S
@@ -132,8 +154,9 @@ pub async fn spawn_child(
 /// `subset_policy` still forbids escalation and honours an explicit narrower
 /// request. Dumbness is a prompt/data-flow property, not a capability wall — a
 /// child CAN spawn/register, it just shouldn't (subagent.txt). A parentless
-/// (direct/CLI/trigger-fired) spawn has no parent to mirror: explicit options
-/// win, else the configured default policy. Whatever was resolved, an ask-mode
+/// direct/CLI/raw-trigger spawn has no owner to mirror: explicit options win,
+/// else the configured default policy. Harness-owned reactions use the
+/// registering turn as `parent_record`. Whatever was resolved, an ask-mode
 /// child is then capped at that policy.
 fn child_functions(
     cfg: &WorkerConfig,
@@ -157,6 +180,7 @@ async fn seed_child(
     req: &SpawnRequest,
     parent: Option<&ParentLink>,
     parent_record: Option<&TurnRecord>,
+    reactive_owner_session_id: Option<&str>,
 ) -> Result<ChildIds, HarnessError> {
     let session = deps.session().await;
 
@@ -166,9 +190,9 @@ async fn seed_child(
         .or_else(|| parent_record.map(|p| p.options.model.clone()))
         .ok_or_else(|| {
             HarnessError::InvalidRequest(
-                "spawn requires a model — only an IN-TURN spawn inherits its parent's; a spawn \
-                 dispatched from a reaction, trigger, or CLI is parentless and inherits \
-                 nothing, so name `model` explicitly in that spawn payload"
+                "spawn requires a model — only an in-turn spawn or a harness-owned reaction \
+                 inherits it; a direct, CLI, or raw-trigger spawn has no owner, so name `model` \
+                 explicitly in that spawn payload"
                     .into(),
             )
         })?;
@@ -345,6 +369,7 @@ async fn seed_child(
         },
         spawned_by_subscription_id: req.spawned_by_subscription_id.clone(),
         reactive_depth: req.reactive_depth,
+        reactive_owner_session_id: reactive_owner_session_id.map(str::to_string),
         functions_generation: None,
         result: None,
         result_error: None,
@@ -463,6 +488,7 @@ mod tests {
             display_parent_session_id: None,
             spawned_by_subscription_id: None,
             reactive_depth: None,
+            reactive_owner_session_id: None,
             functions_generation: None,
             result: None,
             result_error: None,
diff --git a/harness/src/turn_loop.rs b/harness/src/turn_loop.rs
index 3989db397..1f0842ae7 100644
--- a/harness/src/turn_loop.rs
+++ b/harness/src/turn_loop.rs
@@ -1512,6 +1512,15 @@ async fn finalize_failed(
             notify_parent_of_child_failure(deps, &parent.session_id, record, reason, failure).await;
         }
     }
+    if let Some(owner_session_id) = record.reactive_owner_session_id.clone() {
+        let already_notified = record
+            .parent
+            .as_ref()
+            .is_some_and(|parent| parent.session_id == owner_session_id);
+        if !already_notified && owner_session_id != record.session_id {
+            notify_parent_of_child_failure(deps, &owner_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;
diff --git a/harness/src/types/turn.rs b/harness/src/types/turn.rs
index ead505eba..4a36afc21 100644
--- a/harness/src/types/turn.rs
+++ b/harness/src/types/turn.rs
@@ -250,6 +250,13 @@ pub struct TurnRecord {
     /// so `harness::react` can refuse chains past `MAX_REACTIVE_DEPTH`.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub reactive_depth: Option<u32>,
+    /// Session that registered the reaction which spawned this turn.
+    ///
+    /// Unlike `parent`, this does not resolve a pending function call. It is
+    /// durable ownership used to inherit execution limits and to deliver a
+    /// terminal child failure back to the orchestration session.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub reactive_owner_session_id: Option<String>,
     /// Function-registry generation this session last acknowledged; a mismatch
     /// at generate time appends a registry-change notice so session-cached
     /// contracts get re-fetched.
@@ -343,6 +350,7 @@ mod tests {
             display_parent_session_id: None,
             spawned_by_subscription_id: None,
             reactive_depth: None,
+            reactive_owner_session_id: None,
             functions_generation: None,
             result: None,
             result_error: None,
@@ -424,6 +432,7 @@ mod tests {
         r.display_parent_session_id = Some("s_display_parent".into());
         r.spawned_by_subscription_id = Some("sub_1".into());
         r.reactive_depth = Some(3);
+        r.reactive_owner_session_id = Some("s_owner".into());
         let back: TurnRecord = serde_json::from_value(serde_json::to_value(&r).unwrap()).unwrap();
         assert_eq!(back, r);
     }
diff --git a/harness/tests/e2e/src/context.rs b/harness/tests/e2e/src/context.rs
index 593deb7a4..0e9d19721 100644
--- a/harness/tests/e2e/src/context.rs
+++ b/harness/tests/e2e/src/context.rs
@@ -13,6 +13,50 @@ use serde_json::{json, Value};
 
 pub const INVOCATION_TIMEOUT: Duration = Duration::from_secs(120);
 const POLL_INTERVAL: Duration = Duration::from_millis(100);
+const METRICS_POLL_INTERVAL: Duration = Duration::from_secs(1);
+const PROGRESS_SAMPLE_INTERVAL: Duration = Duration::from_secs(15);
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct RootProgress {
+    turn_id: Option<String>,
+    status: TurnStatus,
+    step: u64,
+    pending_function_calls: usize,
+    children: usize,
+    queued_messages: usize,
+    expects_wake: bool,
+}
+
+impl From<&StatusReport> for RootProgress {
+    fn from(status: &StatusReport) -> Self {
+        Self {
+            turn_id: status.turn_id.clone(),
+            status: status.status,
+            step: status.step,
+            pending_function_calls: status.pending_function_calls.len(),
+            children: status.children.len(),
+            queued_messages: status.queued.len(),
+            expects_wake: status.expects_wake,
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct MetricsProgress {
+    sessions: u64,
+    function_calls: u64,
+    function_call_errors: u64,
+}
+
+impl From<&SessionMetricsResponseV1> for MetricsProgress {
+    fn from(metrics: &SessionMetricsResponseV1) -> Self {
+        Self {
+            sessions: metrics.totals.sessions,
+            function_calls: metrics.totals.function_calls,
+            function_call_errors: metrics.totals.function_call_errors,
+        }
+    }
+}
 
 pub struct E2eContext {
     client: IIIClient,
@@ -70,15 +114,19 @@ impl E2eContext {
 
     pub async fn wait_for_turn(
         &self,
+        scenario_id: &str,
         session_id: &str,
         initial_turn_id: &str,
-        timeout: Duration,
+        stuck_timeout: Duration,
         progress_interval: Option<Duration>,
     ) -> Result<StatusReport> {
         let started = tokio::time::Instant::now();
-        let deadline = started + timeout;
+        let mut last_progress = started;
+        let mut next_sample = started;
         let mut next_progress = progress_interval.map(|interval| started + interval);
         let mut active_turn_id = initial_turn_id.to_string();
+        let mut root_progress = None;
+        let mut metrics_progress = None;
         loop {
             let status: Option<StatusReport> = self
                 .trigger("harness::status", json!({ "session_id": session_id }))
@@ -90,10 +138,17 @@ impl E2eContext {
                 if session_is_terminal(&status)? {
                     return Ok(status);
                 }
+                let observed = RootProgress::from(&status);
+                if root_progress.as_ref() != Some(&observed) {
+                    root_progress = Some(observed);
+                    last_progress = tokio::time::Instant::now();
+                }
                 if progress_due(&mut next_progress, progress_interval) {
                     tracing::info!(
+                        scenario = scenario_id,
                         session_id,
                         elapsed_seconds = started.elapsed().as_secs(),
+                        inactive_seconds = last_progress.elapsed().as_secs(),
                         status = ?status.status,
                         step = status.step,
                         turns = status.turn_count,
@@ -106,13 +161,31 @@ impl E2eContext {
                     );
                 }
             }
-            if tokio::time::Instant::now() >= deadline {
-                let _ = self
-                    .stop_session(session_id, Some(active_turn_id.as_str()))
-                    .await;
+            let now = tokio::time::Instant::now();
+            if now >= next_sample {
+                match self.metrics(session_id).await {
+                    Ok(metrics) => {
+                        let observed = MetricsProgress::from(&metrics);
+                        if metrics_progress.as_ref() != Some(&observed) {
+                            metrics_progress = Some(observed);
+                            last_progress = tokio::time::Instant::now();
+                        }
+                    }
+                    Err(error) => tracing::debug!(
+                        scenario = scenario_id,
+                        session_id,
+                        %error,
+                        "could not sample E2E progress metrics"
+                    ),
+                }
+                next_sample = tokio::time::Instant::now() + PROGRESS_SAMPLE_INTERVAL;
+            }
+            if last_progress.elapsed() >= stuck_timeout {
+                self.stop_session_tree(session_id).await;
                 bail!(
-                    "scenario exceeded {}s while waiting for session {session_id}",
-                    timeout.as_secs()
+                    "scenario {scenario_id} made no observable progress for {}s while waiting for \
+                     session {session_id} (last active turn {active_turn_id})",
+                    stuck_timeout.as_secs()
                 );
             }
             tokio::time::sleep(POLL_INTERVAL).await;
@@ -126,20 +199,23 @@ impl E2eContext {
 
     pub async fn wait_for_complete_metrics(
         &self,
+        scenario_id: &str,
         session_id: &str,
-        timeout: Duration,
+        stuck_timeout: Duration,
         progress_interval: Option<Duration>,
     ) -> Result<SessionMetricsResponseV1> {
         let started = tokio::time::Instant::now();
-        let deadline = started + timeout;
+        let mut last_progress = started;
+        let mut metrics_progress = None;
         let mut next_progress = progress_interval.map(|interval| started + interval);
         loop {
-            let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
+            let remaining = stuck_timeout.saturating_sub(last_progress.elapsed());
             if remaining.is_zero() {
                 self.stop_session_tree(session_id).await;
                 bail!(
-                    "scenario exceeded {}s while waiting for the complete session tree {session_id}",
-                    timeout.as_secs()
+                    "scenario {scenario_id} made no observable progress for {}s while waiting \
+                     for the complete session tree {session_id}",
+                    stuck_timeout.as_secs()
                 );
             }
             let metrics = match tokio::time::timeout(remaining, self.metrics(session_id)).await {
@@ -147,14 +223,20 @@ impl E2eContext {
                 Err(_) => {
                     self.stop_session_tree(session_id).await;
                     bail!(
-                        "scenario exceeded {}s while waiting for the complete session tree {session_id}",
-                        timeout.as_secs()
+                        "scenario {scenario_id} made no observable progress for {}s while waiting \
+                         for the complete session tree {session_id}",
+                        stuck_timeout.as_secs()
                     );
                 }
             };
             if metrics.complete {
                 return Ok(metrics);
             }
+            let observed = MetricsProgress::from(&metrics);
+            if metrics_progress.as_ref() != Some(&observed) {
+                metrics_progress = Some(observed);
+                last_progress = tokio::time::Instant::now();
+            }
             if progress_due(&mut next_progress, progress_interval) {
                 let tree = self
                     .trigger::<_, harness::functions::session_tree::SessionTreeResponseV1>(
@@ -164,20 +246,23 @@ impl E2eContext {
                     .await;
                 match tree {
                     Ok(tree) => tracing::info!(
+                        scenario = scenario_id,
                         session_id,
                         elapsed_seconds = started.elapsed().as_secs(),
+                        inactive_seconds = last_progress.elapsed().as_secs(),
                         sessions = tree.sessions.len(),
                         tree_complete = tree.complete,
                         "waiting for descendant sessions to finish"
                     ),
                     Err(error) => tracing::debug!(
+                        scenario = scenario_id,
                         session_id,
                         %error,
                         "could not collect progress for descendant sessions"
                     ),
                 }
             }
-            tokio::time::sleep(POLL_INTERVAL.min(remaining)).await;
+            tokio::time::sleep(METRICS_POLL_INTERVAL.min(remaining)).await;
         }
     }
 
diff --git a/harness/tests/e2e/src/report.rs b/harness/tests/e2e/src/report.rs
index 958733cbb..8aeed822d 100644
--- a/harness/tests/e2e/src/report.rs
+++ b/harness/tests/e2e/src/report.rs
@@ -459,7 +459,7 @@ mod tests {
                 max_turns: 1,
                 max_output_tokens: Some(1),
                 max_total_tokens: 1,
-                timeout_seconds: 1,
+                stuck_timeout_seconds: 1,
             },
             runs,
         )
diff --git a/harness/tests/e2e/src/scenarios/direct_answer.rs b/harness/tests/e2e/src/scenarios/direct_answer.rs
index 1f0dbf720..d0fdaf1fe 100644
--- a/harness/tests/e2e/src/scenarios/direct_answer.rs
+++ b/harness/tests/e2e/src/scenarios/direct_answer.rs
@@ -13,7 +13,7 @@ pub fn scenario(_run_id: &str) -> ScenarioSpec {
             max_turns: 2,
             max_output_tokens: Some(2_048),
             max_total_tokens: 32_768,
-            timeout_seconds: 120,
+            stuck_timeout_seconds: 120,
         },
         threshold: 80,
         criteria: vec![
diff --git a/harness/tests/e2e/src/scenarios/mod.rs b/harness/tests/e2e/src/scenarios/mod.rs
index d19144c7a..796ae3aa0 100644
--- a/harness/tests/e2e/src/scenarios/mod.rs
+++ b/harness/tests/e2e/src/scenarios/mod.rs
@@ -37,7 +37,9 @@ pub struct ExecutionPolicy {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub max_output_tokens: Option<u64>,
     pub max_total_tokens: u64,
-    pub timeout_seconds: u64,
+    /// Stop only after this many seconds without observable useful progress.
+    /// Large scenarios have no fixed wall-clock deadline.
+    pub stuck_timeout_seconds: u64,
 }
 
 impl ExecutionPolicy {
@@ -45,7 +47,7 @@ impl ExecutionPolicy {
         if self.max_turns == 0
             || self.max_output_tokens == Some(0)
             || self.max_total_tokens == 0
-            || self.timeout_seconds == 0
+            || self.stuck_timeout_seconds == 0
         {
             bail!("scenario {scenario_id} has an invalid execution policy");
         }
diff --git a/harness/tests/e2e/src/scenarios/persistent_state.rs b/harness/tests/e2e/src/scenarios/persistent_state.rs
index 8ba8ff4ba..359e645ee 100644
--- a/harness/tests/e2e/src/scenarios/persistent_state.rs
+++ b/harness/tests/e2e/src/scenarios/persistent_state.rs
@@ -24,7 +24,7 @@ pub fn scenario(run_id: &str) -> ScenarioSpec {
             max_turns: 12,
             max_output_tokens: Some(8_192),
             max_total_tokens: 122_880,
-            timeout_seconds: 240,
+            stuck_timeout_seconds: 240,
         },
         threshold: 90,
         criteria: vec![
diff --git a/harness/tests/e2e/src/scenarios/reactive_automation/evaluate.rs b/harness/tests/e2e/src/scenarios/reactive_automation/evaluate.rs
index 82e1b97d7..8cee016c7 100644
--- a/harness/tests/e2e/src/scenarios/reactive_automation/evaluate.rs
+++ b/harness/tests/e2e/src/scenarios/reactive_automation/evaluate.rs
@@ -99,16 +99,7 @@ pub(super) fn score(evidence: &Evidence, names: &ScenarioNames) -> ObjectiveEval
         .iter()
         .all(|check| *check == Some(true))
     });
-    let fallback_documented = report.is_some_and(|report| {
-        report
-            .watch_mechanism
-            .as_deref()
-            .is_some_and(|mechanism| mechanism.to_ascii_lowercase().contains("state"))
-            && report
-                .fallback_reason
-                .as_deref()
-                .is_some_and(|reason| !reason.trim().is_empty())
-    });
+    let fallback_documented = fallback_documented(report);
     let report_identity_matches = report.is_some_and(|report| {
         report.run_id.as_deref() == Some(names.run_label.as_str())
             && report.finalizer_session_id.as_deref() == Some(names.finalizer_session.as_str())
@@ -204,6 +195,26 @@ fn single_report(reports: &[FinalReport]) -> Option<&FinalReport> {
     reports.first().filter(|_| reports.len() == 1)
 }
 
+fn fallback_documented(report: Option<&FinalReport>) -> bool {
+    let Some(report) = report else {
+        return false;
+    };
+    let contains = |value: Option<&str>, needle: &str| {
+        value.is_some_and(|value| value.to_ascii_lowercase().contains(needle))
+    };
+    let state_fallback_used = contains(report.watch_mechanism.as_deref(), "state")
+        || contains(report.spawning_event.as_deref(), "state")
+        || contains(report.reactor_session_id.as_deref(), "fallback");
+    if !state_fallback_used {
+        return contains(report.watch_mechanism.as_deref(), "database");
+    }
+    contains(report.watch_mechanism.as_deref(), "state")
+        && report
+            .fallback_reason
+            .as_deref()
+            .is_some_and(|reason| !reason.trim().is_empty())
+}
+
 #[cfg(test)]
 mod tests {
     use std::collections::BTreeMap;
@@ -309,4 +320,44 @@ mod tests {
             0
         );
     }
+
+    #[test]
+    fn database_watch_does_not_require_an_unused_fallback_reason() {
+        let report = FinalReport {
+            watch_mechanism: Some("database::row-changed".to_string()),
+            reactor_session_id: Some("rctest-aB19-reactor".to_string()),
+            spawning_event: Some("database::row-changed".to_string()),
+            ..Default::default()
+        };
+
+        assert!(fallback_documented(Some(&report)));
+    }
+
+    #[test]
+    fn unrecognized_watch_mechanism_is_rejected() {
+        let report = FinalReport {
+            watch_mechanism: Some("custom watcher".to_string()),
+            reactor_session_id: Some("rctest-aB19-reactor".to_string()),
+            spawning_event: Some("custom event".to_string()),
+            ..Default::default()
+        };
+
+        assert!(!fallback_documented(Some(&report)));
+    }
+
+    #[test]
+    fn state_fallback_requires_matching_mechanism_and_reason() {
+        let mut report = FinalReport {
+            watch_mechanism: Some("database::row-changed".to_string()),
+            reactor_session_id: Some("rctest-aB19-reactor-fallback".to_string()),
+            spawning_event: Some("state:order_signal".to_string()),
+            ..Default::default()
+        };
+
+        assert!(!fallback_documented(Some(&report)));
+        report.watch_mechanism = Some("state".to_string());
+        assert!(!fallback_documented(Some(&report)));
+        report.fallback_reason = Some("database trigger did not fire".to_string());
+        assert!(fallback_documented(Some(&report)));
+    }
 }
diff --git a/harness/tests/e2e/src/scenarios/reactive_automation/mod.rs b/harness/tests/e2e/src/scenarios/reactive_automation/mod.rs
index ec622b1c6..e6cad1ba8 100644
--- a/harness/tests/e2e/src/scenarios/reactive_automation/mod.rs
+++ b/harness/tests/e2e/src/scenarios/reactive_automation/mod.rs
@@ -16,8 +16,11 @@ pub(super) const ORDERS_PER_WRITER: i64 = 5;
 pub(super) const EXPECTED_ORDERS: i64 = EXPECTED_WRITERS as i64 * ORDERS_PER_WRITER;
 
 const STUCK_WATCHDOG_SECONDS: u64 = 600;
-const SCENARIO_TIMEOUT_SECONDS: u64 = 1_800;
-const SCENARIO_MAX_TOTAL_TOKENS: u64 = 1_000_000;
+// Shared across the long-lived root and every writer/reactor/finalizer turn.
+// Discovery alone can approach one million input tokens on large-context
+// models before the three writers start, so retain enough room for the actual
+// workload without relaxing any behavioral gate.
+const SCENARIO_MAX_TOTAL_TOKENS: u64 = 2_000_000;
 
 pub fn scenario(run_id: &str) -> ScenarioSpec {
     let names = ScenarioNames::new(run_id);
@@ -28,7 +31,7 @@ pub fn scenario(run_id: &str) -> ScenarioSpec {
             max_turns: 64,
             max_output_tokens: None,
             max_total_tokens: SCENARIO_MAX_TOTAL_TOKENS,
-            timeout_seconds: SCENARIO_TIMEOUT_SECONDS,
+            stuck_timeout_seconds: STUCK_WATCHDOG_SECONDS,
         },
         threshold: 90,
         criteria: vec![
diff --git a/harness/tests/e2e/src/scenarios/security_review.rs b/harness/tests/e2e/src/scenarios/security_review.rs
index 2bb6ed9df..9cd561a2f 100644
--- a/harness/tests/e2e/src/scenarios/security_review.rs
+++ b/harness/tests/e2e/src/scenarios/security_review.rs
@@ -20,7 +20,7 @@ For each snippet, identify the vulnerability, explain its impact, and recommend
             max_turns: 2,
             max_output_tokens: Some(4_096),
             max_total_tokens: 49_152,
-            timeout_seconds: 120,
+            stuck_timeout_seconds: 120,
         },
         threshold: 80,
         criteria: vec![
diff --git a/harness/tests/e2e/src/suite.rs b/harness/tests/e2e/src/suite.rs
index 47b38c807..4d685cf98 100644
--- a/harness/tests/e2e/src/suite.rs
+++ b/harness/tests/e2e/src/suite.rs
@@ -350,8 +350,7 @@ async fn execute(
         spec,
         progress_interval,
     } = request;
-    let scenario_timeout = Duration::from_secs(spec.execution.timeout_seconds);
-    let scenario_deadline = tokio::time::Instant::now() + scenario_timeout;
+    let stuck_timeout = Duration::from_secs(spec.execution.stuck_timeout_seconds);
     let response: SendResponse = context
         .trigger(
             "harness::send",
@@ -391,19 +390,29 @@ async fn execute(
         ));
     }
 
-    context
+    if let Err(error) = context
         .wait_for_turn(
+            spec.id,
             session_id,
             &response.turn_id,
-            remaining(scenario_deadline),
+            stuck_timeout,
             progress_interval,
         )
         .await
-        .map_err(|error| subject_failure(FailurePhase::Execute, error.to_string()))?;
-    let metrics = context
-        .wait_for_complete_metrics(session_id, remaining(scenario_deadline), progress_interval)
+    {
+        capture_partial_observation(context, session_id, report).await;
+        return Err(subject_failure(FailurePhase::Execute, error.to_string()));
+    }
+    let metrics = match context
+        .wait_for_complete_metrics(spec.id, session_id, stuck_timeout, progress_interval)
         .await
-        .map_err(|error| collection_failure(FailurePhase::Collect, error.to_string()))?;
+    {
+        Ok(metrics) => metrics,
+        Err(error) => {
+            capture_partial_observation(context, session_id, report).await;
+            return Err(collection_failure(FailurePhase::Collect, error.to_string()));
+        }
+    };
     let transcript = context.transcript(session_id).await.map_err(|error| {
         RunFailure::new(
             RunStatus::InfrastructureError,
@@ -467,8 +476,27 @@ async fn execute(
     Ok(())
 }
 
-fn remaining(deadline: tokio::time::Instant) -> Duration {
-    deadline.saturating_duration_since(tokio::time::Instant::now())
+async fn capture_partial_observation(
+    context: &E2eContext,
+    session_id: &str,
+    report: &mut E2eRunReport,
+) {
+    match context.metrics(session_id).await {
+        Ok(metrics) => report.metrics = Some(metrics),
+        Err(error) => tracing::warn!(
+            session_id,
+            %error,
+            "could not capture partial E2E metrics"
+        ),
+    }
+    match context.transcript(session_id).await {
+        Ok(transcript) => report.transcript = Some(transcript),
+        Err(error) => tracing::warn!(
+            session_id,
+            %error,
+            "could not capture partial E2E transcript"
+        ),
+    }
 }
 
 fn is_resource_limit(message: &str) -> bool {
@@ -479,6 +507,7 @@ fn is_resource_limit(message: &str) -> bool {
         "max_total_tokens",
         "cost budget",
         "scenario exceeded",
+        "no observable progress",
         "maximum turn",
         "turn limit",
         "context length",
@@ -634,7 +663,7 @@ mod tests {
                 max_turns: 1,
                 max_output_tokens: Some(1),
                 max_total_tokens: 1,
-                timeout_seconds: 1,
+                stuck_timeout_seconds: 1,
             },
             threshold: 80,
             criteria: vec![CriterionSpec {