Harden the reactive pipeline path: argument recovery, once on react bindings, wiring advisories, scoped scrapling renders

harness/src/functions/react.rs

@@ -189,6 +189,12 @@ pub struct ReactSpec {
     /// auto-unregister its predecessor subscriptions.
     #[serde(default, rename = "__subscription_id")]
     pub subscription_id: Option<String>,
+    /// Stamped by the registration interceptor when the agent asked `once:
+    /// true` (never caller-supplied): retire this binding after its first
+    /// successful spawn. Meaningless on join edges — the join lifecycle owns
+    /// predecessor bindings.
+    #[serde(default, rename = "__once")]
+    pub once: bool,
     /// Stamped by the interceptor at registration (never caller-supplied): the
     /// registering session. Console-tree parent fallback for fires whose event
     /// carries no session id (state/cron/stream).
@@ -362,14 +368,18 @@ pub async fn handle(
 
     match spec.join.clone() {
         None => {
-            spawn_reaction(
+            let res = spawn_reaction(
                 deps,
                 single_event_task(&spec.task, &event),
                 &spec,
                 parent,
                 spawn_depth,
             )
-            .await
+            .await;
+            if spec.once && matches!(&res, Ok(r) if r.spawned) {
+                once_unregister(deps, &spec).await;
+            }
+            res
         }
         Some(join) => join_edge(deps, event, &spec, &join, parent, spawn_depth).await,
     }
@@ -436,21 +446,8 @@ async fn join_edge(
         tracing::info!(join = %join.id, "harness::react: join re-armed; predecessor subscriptions stay registered");
     } else {
         for id in join_binding_ids(&rec) {
-            // Turn-event edges record the ENGINE binding id (stamped by the
-            // fan-out); state/cron/stream edges record the interceptor's
-            // local `sub_` handle — resolve it through the registry first.
-            let engine_id = if id.starts_with("sub_") {
-                deps.subscriptions.take(&id).and_then(|(_, t)| t)
-            } else {
-                deps.subscriptions.take_by_trigger_id(&id);
-                Some(id.clone())
-            };
-            let Some(engine_id) = engine_id else {
-                tracing::warn!(join = %join.id, subscription = %id, "harness::react: join predecessor has no resolvable engine binding; skipping unregister");
-                continue;
-            };
-            if let Err(e) = unregister_subscription(deps, &engine_id).await {
-                tracing::warn!(error = %e, join = %join.id, subscription = %engine_id, "harness::react: join subscription auto-unregister failed");
+            if let Err(e) = retire_binding(deps, &id).await {
+                tracing::warn!(error = %e, join = %join.id, subscription = %id, "harness::react: join predecessor auto-unregister failed");
             }
         }
     }
@@ -597,6 +594,57 @@ async fn resolve_root(deps: &Deps, session_id: &str) -> String {
     current
 }
 
+/// Retire a fired binding: engine unregister FIRST, local eviction only after
+/// it succeeds — evicting first would orphan the durable engine binding as a
+/// standing refire if the unregister call failed, with the `sub_` mapping gone
+/// so no later retry could resolve it. Turn-event fires stamp the engine
+/// binding id directly; state/cron/stream fires deliver the interceptor's
+/// local `sub_` handle — resolve it through the registry first.
+async fn retire_binding(deps: &Deps, id: &str) -> Result<(), HarnessError> {
+    let engine_id = if id.starts_with("sub_") {
+        match deps.subscriptions.trigger_id_of(id) {
+            Some(t) => t,
+            // Bind window: the binding fired before the registration
+            // round-trip recorded its engine id. Evict the slot so
+            // `set_trigger_id` finds it gone and the registration path
+            // unregisters the orphan engine trigger itself.
+            None if deps.subscriptions.session_of(id).is_some() => {
+                deps.subscriptions.take(id);
+                return Ok(());
+            }
+            None => {
+                return Err(HarnessError::Dependency(format!(
+                    "no local binding for subscription `{id}`"
+                )));
+            }
+        }
+    } else {
+        id.to_string()
+    };
+    unregister_subscription(deps, &engine_id).await?;
+    if id.starts_with("sub_") {
+        deps.subscriptions.take(id);
+    } else {
+        deps.subscriptions.take_by_trigger_id(id);
+    }
+    Ok(())
+}
+
+/// A `once: true` simple edge spawned: retire its binding so it never refires.
+/// Best-effort — a failed unregister only risks an extra fire, never the
+/// spawn, and the retained mapping lets the next fire retry the retirement.
+async fn once_unregister(deps: &Deps, spec: &ReactSpec) {
+    let Some(id) = spec.subscription_id.as_deref() else {
+        tracing::warn!(
+            "harness::react: once-binding fired without a subscription id; cannot auto-unregister"
+        );
+        return;
+    };
+    if let Err(e) = retire_binding(deps, id).await {
+        tracing::warn!(error = %e, subscription = %id, "harness::react: once-binding auto-unregister failed; retrying on the next fire");
+    }
+}
+
 async fn unregister_subscription(deps: &Deps, id: &str) -> Result<(), HarnessError> {
     deps.iii
         .trigger(TriggerRequest {
@@ -816,10 +864,20 @@ mod tests {
             parent_session_id: None,
             join: None,
             subscription_id: None,
+            once: false,
             owner_session_id: None,
         }
     }
 
+    #[test]
+    fn once_stamp_parses_and_defaults_off() {
+        let s: ReactSpec = serde_json::from_value(json!({ "model": "m", "task": "t" })).unwrap();
+        assert!(!s.once);
+        let s: ReactSpec =
+            serde_json::from_value(json!({ "model": "m", "task": "t", "__once": true })).unwrap();
+        assert!(s.once);
+    }
+
     #[test]
     fn simple_event_task_embeds_event() {
         let ev = json!({ "session_id": "s_child", "turn_id": "t1", "status": "completed" });