(MOT-4269) fix(harness): stabilize reactive E2E execution

.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`

.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"

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);
     }
 

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

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,

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,16 +124,39 @@ 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
 /// full policy by default (same permissions as its main agent);
 /// `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,