(MOT-4211, MOT-4212) feat(harness): coalesce capped fires; reactions inherit registrant policy#583
(MOT-4211, MOT-4212) feat(harness): coalesce capped fires; reactions inherit registrant policy#583andersonleal wants to merge 6 commits into
Conversation
…ate bindings The fire-rate breaker (10/60s per subscription) DROPPED capped fires. For recompute-style reactions that is the worst possible choice: a burst's TAIL is exactly what gets capped, so aggregates freeze one step behind the source of truth. Live run rctest-k7m3: 15 orders in ~30s across two overlapping bindings -> 17 dropped reactions, totals stuck at 4/3/3 vs truth 5/5/5, and a wall of "reaction failed — fire-rate cap" outcomes. Capped fires now COALESCE: the newest event parks in the gate slot (drops accumulate), one trailing task per key re-enters harness::react when the window frees, delivering the latest event stamped with __coalesced_fires. Bounded by construction — one pending event and one trailing task per key, so a true runaway costs at most MAX_FIRES_PER_WINDOW + 1 reactions per window and the loop-breaker property is preserved. The capped-fire outcome records "waiting" with the collapse count instead of "failed". Second half of the same postmortem: the run registered BOTH a keyed and a key-less state binding on one scope — the key-less one fires for EVERY key (done markers, completion signals, everything), doubling budget burn. New registration advisory (react + notify paths) warns on state bindings without a key filter, naming the scope.
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
📝 WalkthroughWalkthroughChangesReact subscriptions now stamp trusted registrant policies, coalesce rate-limited fires with newest-wins trailing delivery, apply policy subsets to spawned payloads, and add probe-driven E2E coverage for reaction, state-worker, and coalesced-fire flows. React trigger behavior
E2E and state-worker coverage
Estimated code review effort: 5 (Critical) | ~100 minutes Sequence Diagram(s)sequenceDiagram
participant ScenarioRunner
participant ScenarioProbe
participant StateWorker
participant ReactHandler
participant ControlledFunction
ScenarioRunner->>ScenarioProbe: dispatch state::set probe
StateWorker->>ReactHandler: deliver state reaction with metadata
ReactHandler->>ControlledFunction: dispatch recorder call
ControlledFunction-->>ScenarioProbe: record target call
ScenarioProbe-->>ScenarioRunner: satisfy call-count wait
Possibly related PRs
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
skill-check — worker0 verified, 48 skipped (no docs/).
Four for four. Nicely done. |
andersonleal
changed the title
A trigger-fired reaction spawned parentless and landed on the read-only baseline whenever the spec carried no options.functions. Live run rctest-k7m3: the wrap-up reaction — delivered into the ORCHESTRATOR'S OWN chat — was denied database::query (x6), state::set (x1), and engine::unregister_trigger (x4) in the same session whose earlier turns called all three freely. The reaction whose job was cleanup couldn't unregister, completing the binding-leak cycle across runs. The register interceptor now stamps the registering turn's dispatch policy onto the react binding (__registrant_functions — harness stamp, smuggled values stripped, both modes). At fire time the spawned turn inherits it; explicit options.functions are subset against it (narrow, never escalate — the in-turn child rule, subset_policy). Raw engine-side registrations carry no stamp and keep the read-only baseline fallback. CallerModel (already threading the registrant's model for inherit_model) now carries the policy from the same trusted TurnRecord source.
andersonleal
changed the title
|
Added registrant-policy inheritance ( |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@harness/src/functions/react.rs`:
- Around line 556-601: Gate the rate-limit warning and waiting outcome in the
!deps.react_gate.admit(...) branch on deferred.schedule_delay_ms.is_some(), so
only the first deferral in each cycle performs logging and
record_reaction_outcome. Keep defer, pending-event coalescing, trailing-task
scheduling, and the final return behavior unchanged for subsequent deferrals.
In `@harness/src/functions/subscribe.rs`:
- Around line 666-668: The join_canon canonicalization path must also remove the
__registrant_functions metadata stamp. Update join_canon to strip
REGISTRANT_FUNCTIONS_KEY alongside __subscription_id, __once,
__owner_session_id, and join.key, preserving predecessor fingerprints based only
on task/model/session/options.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 0528ffe4-9fc7-4272-9811-60914a980521
📒 Files selected for processing (2)
harness/src/functions/react.rsharness/src/functions/subscribe.rs
| if !deps.react_gate.admit(&gate_key, now_ms) { | ||
| // Coalesce instead of dropping: react patterns are recompute-style, | ||
| // so the newest event subsumes the capped ones. Park it and let one | ||
| // trailing fire deliver it when the window frees — without this, a | ||
| // burst's TAIL is exactly what gets dropped and aggregates freeze | ||
| // one step behind the source of truth (rctest-k7m3 postmortem). | ||
| let deferred = deps.react_gate.defer( | ||
| &gate_key, | ||
| event.clone(), | ||
| raw_metadata.unwrap_or(Value::Null), | ||
| now_ms, | ||
| ); | ||
| tracing::warn!( | ||
| subscription = %gate_key, | ||
| "harness::react: fire-rate breaker tripped ({MAX_FIRES_PER_WINDOW} fires/{FIRE_WINDOW_MS}ms); not reacting" | ||
| dropped = deferred.dropped, | ||
| "harness::react: fire-rate breaker tripped ({MAX_FIRES_PER_WINDOW} fires/{FIRE_WINDOW_MS}ms); coalescing" | ||
| ); | ||
| if let Some(delay_ms) = deferred.schedule_delay_ms { | ||
| let deps = deps.clone(); | ||
| let key = gate_key.clone(); | ||
| tokio::spawn(async move { | ||
| tokio::time::sleep(std::time::Duration::from_millis(delay_ms as u64)).await; | ||
| let Some(pending) = deps.react_gate.take_pending(&key) else { | ||
| return; | ||
| }; | ||
| let mut event = pending.event; | ||
| if let Some(obj) = event.as_object_mut() { | ||
| // Tell the reaction it stands in for a collapsed burst. | ||
| obj.insert("__coalesced_fires".to_string(), json!(pending.dropped)); | ||
| } | ||
| // Re-enters the gate: if the window is still hot the fire | ||
| // parks again and a fresh trailing task takes over. | ||
| if let Err(e) = handle_boxed(&deps, event, Some(pending.metadata)).await { | ||
| tracing::warn!(subscription = %key, error = %e, | ||
| "harness::react: trailing coalesced fire failed"); | ||
| } | ||
| }); | ||
| } | ||
| let note = format!( | ||
| "fire-rate cap ({MAX_FIRES_PER_WINDOW} per {FIRE_WINDOW_MS}ms) reached for this subscription; not spawning" | ||
| "fire-rate cap ({MAX_FIRES_PER_WINDOW} per {FIRE_WINDOW_MS}ms) reached; coalescing — \ | ||
| the latest event fires once the window frees ({} collapsed so far)", | ||
| deferred.dropped | ||
| ); | ||
| record_reaction_outcome(deps, &spec, &event, "failed", ¬e, None).await; | ||
| record_reaction_outcome(deps, &spec, &event, "waiting", ¬e, None).await; | ||
| return Ok(ReactResult::note(note)); | ||
| } |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟠 Major | ⚡ Quick win
Every capped fire writes an outcome record and a log line, not just the first per drain cycle.
record_reaction_outcome(..., "waiting", ...) and tracing::warn! run on every single call that hits admit() == false, regardless of whether this deferral is the first of the cycle (schedule_delay_ms.is_some()) or a mid-cycle overwrite (schedule_delay_ms.is_none()). During exactly the runaway-burst scenario this gate exists to bound, this reintroduces unbounded I/O/log volume proportional to event ingress rate — the doc comment's "MAX_FIRES_PER_WINDOW + 1 reactions per window" bound only covers spawned reactions, not the outcome-store writes and log lines on the drop path.
Consider gating the record/log to the first deferral of a cycle (deferred.schedule_delay_ms.is_some()), since the trailing fire's own outcome record (once it actually spawns/fails) already reflects the final state and the accumulated dropped count.
♻️ Suggested throttle
let deferred = deps.react_gate.defer(
&gate_key,
event.clone(),
raw_metadata.unwrap_or(Value::Null),
now_ms,
);
- tracing::warn!(
- subscription = %gate_key,
- dropped = deferred.dropped,
- "harness::react: fire-rate breaker tripped ({MAX_FIRES_PER_WINDOW} fires/{FIRE_WINDOW_MS}ms); coalescing"
- );
+ if deferred.schedule_delay_ms.is_some() {
+ tracing::warn!(
+ subscription = %gate_key,
+ "harness::react: fire-rate breaker tripped ({MAX_FIRES_PER_WINDOW} fires/{FIRE_WINDOW_MS}ms); coalescing"
+ );
+ }
if let Some(delay_ms) = deferred.schedule_delay_ms {
...
}
let note = format!(...);
- record_reaction_outcome(deps, &spec, &event, "waiting", ¬e, None).await;
+ if deferred.schedule_delay_ms.is_some() {
+ record_reaction_outcome(deps, &spec, &event, "waiting", ¬e, None).await;
+ }
return Ok(ReactResult::note(note));📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| if !deps.react_gate.admit(&gate_key, now_ms) { | |
| // Coalesce instead of dropping: react patterns are recompute-style, | |
| // so the newest event subsumes the capped ones. Park it and let one | |
| // trailing fire deliver it when the window frees — without this, a | |
| // burst's TAIL is exactly what gets dropped and aggregates freeze | |
| // one step behind the source of truth (rctest-k7m3 postmortem). | |
| let deferred = deps.react_gate.defer( | |
| &gate_key, | |
| event.clone(), | |
| raw_metadata.unwrap_or(Value::Null), | |
| now_ms, | |
| ); | |
| tracing::warn!( | |
| subscription = %gate_key, | |
| "harness::react: fire-rate breaker tripped ({MAX_FIRES_PER_WINDOW} fires/{FIRE_WINDOW_MS}ms); not reacting" | |
| dropped = deferred.dropped, | |
| "harness::react: fire-rate breaker tripped ({MAX_FIRES_PER_WINDOW} fires/{FIRE_WINDOW_MS}ms); coalescing" | |
| ); | |
| if let Some(delay_ms) = deferred.schedule_delay_ms { | |
| let deps = deps.clone(); | |
| let key = gate_key.clone(); | |
| tokio::spawn(async move { | |
| tokio::time::sleep(std::time::Duration::from_millis(delay_ms as u64)).await; | |
| let Some(pending) = deps.react_gate.take_pending(&key) else { | |
| return; | |
| }; | |
| let mut event = pending.event; | |
| if let Some(obj) = event.as_object_mut() { | |
| // Tell the reaction it stands in for a collapsed burst. | |
| obj.insert("__coalesced_fires".to_string(), json!(pending.dropped)); | |
| } | |
| // Re-enters the gate: if the window is still hot the fire | |
| // parks again and a fresh trailing task takes over. | |
| if let Err(e) = handle_boxed(&deps, event, Some(pending.metadata)).await { | |
| tracing::warn!(subscription = %key, error = %e, | |
| "harness::react: trailing coalesced fire failed"); | |
| } | |
| }); | |
| } | |
| let note = format!( | |
| "fire-rate cap ({MAX_FIRES_PER_WINDOW} per {FIRE_WINDOW_MS}ms) reached for this subscription; not spawning" | |
| "fire-rate cap ({MAX_FIRES_PER_WINDOW} per {FIRE_WINDOW_MS}ms) reached; coalescing — \ | |
| the latest event fires once the window frees ({} collapsed so far)", | |
| deferred.dropped | |
| ); | |
| record_reaction_outcome(deps, &spec, &event, "failed", ¬e, None).await; | |
| record_reaction_outcome(deps, &spec, &event, "waiting", ¬e, None).await; | |
| return Ok(ReactResult::note(note)); | |
| } | |
| if !deps.react_gate.admit(&gate_key, now_ms) { | |
| // Coalesce instead of dropping: react patterns are recompute-style, | |
| // so the newest event subsumes the capped ones. Park it and let one | |
| // trailing fire deliver it when the window frees — without this, a | |
| // burst's TAIL is exactly what gets dropped and aggregates freeze | |
| // one step behind the source of truth (rctest-k7m3 postmortem). | |
| let deferred = deps.react_gate.defer( | |
| &gate_key, | |
| event.clone(), | |
| raw_metadata.unwrap_or(Value::Null), | |
| now_ms, | |
| ); | |
| if deferred.schedule_delay_ms.is_some() { | |
| tracing::warn!( | |
| subscription = %gate_key, | |
| "harness::react: fire-rate breaker tripped ({MAX_FIRES_PER_WINDOW} fires/{FIRE_WINDOW_MS}ms); coalescing" | |
| ); | |
| } | |
| if let Some(delay_ms) = deferred.schedule_delay_ms { | |
| let deps = deps.clone(); | |
| let key = gate_key.clone(); | |
| tokio::spawn(async move { | |
| tokio::time::sleep(std::time::Duration::from_millis(delay_ms as u64)).await; | |
| let Some(pending) = deps.react_gate.take_pending(&key) else { | |
| return; | |
| }; | |
| let mut event = pending.event; | |
| if let Some(obj) = event.as_object_mut() { | |
| // Tell the reaction it stands in for a collapsed burst. | |
| obj.insert("__coalesced_fires".to_string(), json!(pending.dropped)); | |
| } | |
| // Re-enters the gate: if the window is still hot the fire | |
| // parks again and a fresh trailing task takes over. | |
| if let Err(e) = handle_boxed(&deps, event, Some(pending.metadata)).await { | |
| tracing::warn!(subscription = %key, error = %e, | |
| "harness::react: trailing coalesced fire failed"); | |
| } | |
| }); | |
| } | |
| let note = format!( | |
| "fire-rate cap ({MAX_FIRES_PER_WINDOW} per {FIRE_WINDOW_MS}ms) reached; coalescing — \ | |
| the latest event fires once the window frees ({} collapsed so far)", | |
| deferred.dropped | |
| ); | |
| if deferred.schedule_delay_ms.is_some() { | |
| record_reaction_outcome(deps, &spec, &event, "waiting", ¬e, None).await; | |
| } | |
| return Ok(ReactResult::note(note)); | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@harness/src/functions/react.rs` around lines 556 - 601, Gate the rate-limit
warning and waiting outcome in the !deps.react_gate.admit(...) branch on
deferred.schedule_delay_ms.is_some(), so only the first deferral in each cycle
performs logging and record_reaction_outcome. Keep defer, pending-event
coalescing, trailing-task scheduling, and the final return behavior unchanged
for subsequent deferrals.
| // Both modes: strips any smuggled registrant-policy stamp; inert for call | ||
| // reactions (they dispatch with worker authority, no spawned turn). | ||
| stamp_registrant_functions(&mut req.metadata, caller); |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
rg -n "fn join_canon" -A 40 harness/src/functions/react.rsRepository: iii-hq/workers
Length of output: 5860
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== subscribe relevant section =="
sed -n '620,700p' harness/src/functions/subscribe.rs
echo
echo "== stamp_registrant_functions definition/usages =="
rg -n "fn stamp_registrant_functions|stamp_registrant_functions|__registrant_functions" harness/src -g '*.rs'
echo
echo "== react join_canon tests =="
sed -n '1720,1742p' harness/src/functions/react.rs
echo
echo "== deterministic semantic probe for current join_canon strip behavior =="
python3 - <<'PY'
from pathlib import Path
import re
src = Path('harness/src/functions/react.rs').read_text()
m = re.search(r'pub fn join_canon\(metadata: Option<&Value>\) -> Option<\(String, Value\)> \{(?P<body>.*?)^}', src, re.S|re.M)
print("join_canon snippet:")
start = src.find('pub fn join_canon')
end = src.find('\n/// Registration-time validation', start)
print(src[start:end])
PYRepository: iii-hq/workers
Length of output: 6932
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== stamp_registrant_functions implementation =="
sed -n '288,320p' harness/src/functions/subscribe.rs
echo
echo "== subscription key constants =="
sed -n '150,170p' harness/src/functions/subscribe.rs
rg -n "OWNER_SESSION_KEY|owner_session|__owner_session_id|RegistrantModelPolicy|RegistrantFunctions|__" harness/src/functions/subscribe.rs harness/src/subscriptions*.rs harness/src -g '*.rs' | head -n 80
echo
echo "== dedup key if relevant =="
rg -n "registration_dedup_key" -A 60 harness/src/functions/subscribe.rs
echo
echo "== conflicting_join_spec shape =="
rg -n "conflicting_join_spec|join_canon" -A 35 harness/src -g '*.rs'Repository: iii-hq/workers
Length of output: 7873
Strip __registrant_functions in join_canon.
join_canon currently only removes __subscription_id, __once, __owner_session_id, and join.key, so the newly added __registrant_functions stamp makes join predecessors of identical specs fingerprint differently when registered from sessions with different dispatch policies. Add REGISTRANT_FUNCTIONS_KEY (__registrant_functions) to the canonical strip path so join predecessor equivalence depends only on task/model/session/options.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@harness/src/functions/subscribe.rs` around lines 666 - 668, The join_canon
canonicalization path must also remove the __registrant_functions metadata
stamp. Update join_canon to strip REGISTRANT_FUNCTIONS_KEY alongside
__subscription_id, __once, __owner_session_id, and join.key, preserving
predecessor fingerprints based only on task/model/session/options.
… policy Drives the rctest-k7m3 run-3 wrap-up failure through the real stack: the scripted model registers a turn-completed reaction on its own session with NO options, the turn completes, the binding fires, and the reaction seeds a second terminal turn in the same session whose scripted call to the controlled function must dispatch successfully — possible only under the inherited policy (the read-only baseline exposes no worker functions, so the pre-fix run fails both the tools pin and the no-policy-denial verify). Verified meaningful: fails against pre-inheritance sources (d34fc5f). Uses the two-terminal-turn await (terminal_turn_statuses) from E2E-003 and pins the reaction turn's steps with turn_request_step. Reaction turns run the sub-agent prompt, so those generations match the prompt by presence, not the default sha. Coalescing (the PR's other half) stays unit-pinned: its trailing fire lands when the 60s window frees, which cannot fit the e2e deadline.
|
Added E2E-005 |
…ract docs
rctest-k7m2 exposed the coalescing contract's sharp edge: only the
newest event survives a collapse, and a reactor doing per-key upserts
from single events silently skipped a writer whose events were all
among the collapsed (totals converged for w1/w3, never w2). The
recompute-on-coalesce rule existed only in this repo's heads.
Three places now carry it, closest first:
- the trailing event itself: stamp_coalesced adds __coalesced_note
("stands in for N fires ... recompute from the source of truth")
beside the machine-readable __coalesced_fires — guidance delivered
at the exact moment the reacting model reads the event
- ReactSpec.task docs: tasks that process single events per key must
recompute on a coalesced fire
- SubscribeRequest.metadata docs: the coalescing behavior is part of
the subscription contract agents read at registration time
Non-object events cannot carry the stamp and pass through unchanged.
|
Follow-up from the rctest-k7m2 live run ( |
The cross-worker sidecar path (MOT-4209/MOT-4213) couldn't be gated
deterministically: a reaction firing mid-turn races the strict-ordinal
scripted router. The probe hook fixes that — a scenario declares
`probe_after(n, function_id, payload)`, and the runner fires it once n
terminal turns are observed, so the reaction it trips runs while the
tracked session is idle (its turn is the only active one).
- ProbeAction on the fixture + `probe_after` DSL builder; the await
phase waits to each action's boundary, fires it (expanding
{{run_id}}/{{session_id}}), then awaits the rest.
- expected_traces for Direct is now 1 + probe_actions: a probe-tripped
reaction is a fresh externally-initiated flow, so it traces
separately (like a Playground external turn).
- The integration stack now runs the standalone `state` worker (the
engine builtin disabled) so the `state` trigger type is owned exactly
as in production — the fan-out seam under test.
E2E-006 (state-worker-sidecar): main turn registers a state-key
reaction and completes; the probe writes the key; the state worker
fans out to harness::react WITH the sidecar, seeding a second turn
that calls the recorder. Verified meaningful — reverting the state
worker to its pre-MOT-4209 sources times the scenario out (no sidecar
-> no reaction -> no second turn). 5/5 stable.
Coalescing's trailing-fire e2e stays deferred (MOT-4215): call
reactions leave no main-session trace and task reactions message-merge
when several target one session, so it also needs whole-run trace
evidence, not just the probe hook. Coalescing logic remains unit-pinned.
|
Implemented the probe post-completion hook ( New scenario E2E-006 Coalescing's trailing-fire e2e stays deferred to MOT-4215: it hits a second wall the probe hook doesn't clear (call reactions leave no main-session trace; task reactions message-merge when several target one session), so it also needs whole-run trace evidence. Coalescing logic remains unit-pinned + live-confirmed. |
There was a problem hiding this comment.
🧹 Nitpick comments (3)
harness/tests/e2e/src/fixtures/loading.rs (1)
140-145: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick win
validate()never exercises probe-action payloads or bounds.
compiled()only carriesscenario/script/system_prompt_template, sovalidate()'sexpand_compiled_fixturesanity check never touchesProbeAction.payload. A malformed{{placeholder}}in a probe action's payload, or anafter_turnsoutside[1, expected_terminal_turns), passesfixture.validate().unwrap()in unit tests and only surfaces as a real-engine E2E failure/timeout.♻️ Suggested addition to `validate()`
crate::expand::expand_compiled_fixture( &self.compiled(), "validate-run", "validate-session", )?; + for action in &self.probe_actions { + anyhow::ensure!( + !action.function_id.is_empty(), + "probe action has an empty function_id" + ); + anyhow::ensure!( + action.after_turns > 0 && action.after_turns < self.expected_terminal_turns, + "probe action after_turns ({}) must be within [1, {})", + action.after_turns, + self.expected_terminal_turns + ); + let mut payload = action.payload.clone(); + crate::expand::Placeholders::new("validate-run", "validate-session") + .expand_value(&mut payload)?; + } Ok(())Also applies to: 164-170
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@harness/tests/e2e/src/fixtures/loading.rs` around lines 140 - 145, Update the fixture validation method around expand_compiled_fixture so it also expands and validates every ProbeAction payload, including placeholder substitution and payload bounds. Ensure after_turns values are rejected unless they fall within [1, expected_terminal_turns), while preserving the existing compiled fixture validation behavior.harness/tests/e2e/src/scenarios/state_worker_sidecar.rs (1)
156-175: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDuplicates
message_textinstead of reusing it.This block manually re-extracts
contentblock text from afunction_resultmessage, duplicating whatcrate::evidence_data::message_textalready does — used for the identical check inreaction_policy_inheritance.rs(same cohort).♻️ Suggested fix
-use crate::fixtures::ScenarioFixture; +use crate::evidence_data::message_text; +use crate::fixtures::ScenarioFixture;if msg.get("role").and_then(Value::as_str) == Some("function_result") { - let text: String = msg - .get("content") - .and_then(Value::as_array) - .map(|blocks| { - blocks - .iter() - .filter_map(|b| b.get("text").and_then(Value::as_str)) - .collect::<Vec<_>>() - .concat() - }) - .unwrap_or_default(); + let text = message_text(&msg); anyhow::ensure!( !text.contains("not permitted"), "a dispatch was denied: {text}" ); }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@harness/tests/e2e/src/scenarios/state_worker_sidecar.rs` around lines 156 - 175, Replace the manual content-block text extraction inside the transcript loop with the existing crate::evidence_data::message_text helper, while preserving the function_result role filter and the anyhow::ensure! denial check. Reuse that helper’s result for the “not permitted” validation instead of constructing text locally.harness/tests/e2e/src/scenario/phases/completion.rs (1)
42-63: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winInconsistent deadline-expiry handling between the wait and the fire step.
The
wait_for_completion_turnserror path explicitly checksdeadline.is_expired()and degrades gracefully (active.timed_out = true; return Ok(())), but the immediately followingfire_probe_action(...).await?(Line 61) propagates any failure as a hardRunErrorregardless of whether the deadline expired. If the deadline lapses between the wait succeeding and the dispatch call, this turns what should be a graceful timeout into a hard CI failure.♻️ Suggested fix
- self.fire_probe_action(services, &action, deadline).await?; + if let Err(error) = self.fire_probe_action(services, &action, deadline).await { + if deadline.is_expired() { + active.timed_out = true; + return Ok(()); + } + return Err(error); + }Please confirm
call_with_deadline's error shape on deadline expiry to validate this is actually reachable.Also applies to: 136-161
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@harness/tests/e2e/src/scenario/phases/completion.rs` around lines 42 - 63, Update the probe-action dispatch path around fire_probe_action and its additional occurrence so failures are checked for deadline expiry before propagating. Confirm call_with_deadline’s deadline error shape, then mark active.timed_out and return Ok(()) when the deadline has expired; preserve RunError propagation for other failures and keep the existing wait_for_completion_turns handling consistent.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@harness/tests/e2e/src/fixtures/loading.rs`:
- Around line 140-145: Update the fixture validation method around
expand_compiled_fixture so it also expands and validates every ProbeAction
payload, including placeholder substitution and payload bounds. Ensure
after_turns values are rejected unless they fall within [1,
expected_terminal_turns), while preserving the existing compiled fixture
validation behavior.
In `@harness/tests/e2e/src/scenario/phases/completion.rs`:
- Around line 42-63: Update the probe-action dispatch path around
fire_probe_action and its additional occurrence so failures are checked for
deadline expiry before propagating. Confirm call_with_deadline’s deadline error
shape, then mark active.timed_out and return Ok(()) when the deadline has
expired; preserve RunError propagation for other failures and keep the existing
wait_for_completion_turns handling consistent.
In `@harness/tests/e2e/src/scenarios/state_worker_sidecar.rs`:
- Around line 156-175: Replace the manual content-block text extraction inside
the transcript loop with the existing crate::evidence_data::message_text helper,
while preserving the function_result role filter and the anyhow::ensure! denial
check. Reuse that helper’s result for the “not permitted” validation instead of
constructing text locally.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 5e8314e6-adb4-492d-81cf-f813ac17870e
📒 Files selected for processing (13)
harness/Makefileharness/src/functions/react.rsharness/src/functions/subscribe.rsharness/tests/e2e/README.mdharness/tests/e2e/src/fixtures.rsharness/tests/e2e/src/fixtures/loading.rsharness/tests/e2e/src/fixtures/tests.rsharness/tests/e2e/src/scenario/phases/completion.rsharness/tests/e2e/src/scenarios/dsl.rsharness/tests/e2e/src/scenarios/mod.rsharness/tests/e2e/src/scenarios/reaction_policy_inheritance.rsharness/tests/e2e/src/scenarios/state_worker_sidecar.rsharness/tests/e2e/src/stack/config.rs
🚧 Files skipped from review as they are similar to previous changes (2)
- harness/src/functions/subscribe.rs
- harness/src/functions/react.rs
Call-mode reactions are invisible to the runner's two evidence surfaces — they seed no turn (completions blind) and no session trace (trace floor blind) — which is what kept the coalescing e2e deferred. Two runner capabilities close that gap: * probe-side WHOLE-RUN call evidence: the probe serves the controlled function, so it records every dispatch payload regardless of origin (RunEvidence::target_calls, expect_target_calls), and * await_target_calls: the await phase holds until the controlled function served N calls — the only way to await a trailing coalesced fire that lands after the rate-cap window frees. The fire gate's limits gain env overrides (III_HARNESS_MAX_FIRES_PER_ WINDOW / III_HARNESS_FIRE_WINDOW_MS, read once), and fixtures can pass env to the harness under test plus override the floor's trace count (call reactions form no extra trace tree). E2E-007 bursts 5 state fires through a call-mode binding under a 3-per-2s gate: exactly cap + 1 dispatches, the first three unstamped with distinct seqs, the trailing one stamped __coalesced_fires: 2 with the self-describing recovery note, its event one of the CAPPED fires (newest wins — one seq is never observed, by design). Gate check: with the trailing-fire scheduling disabled (pre-MOT-4211 drop semantics) the scenario times out; with it, 6/6 passes at ~4.6s. All 7 direct scenarios pass.
|
MOT-4215 landed ( The two walls that kept it deferred are closed with runner capabilities rather than trace surgery:
The fire gate gained once-read env overrides ( Assertions: 5-fire burst → exactly cap+1 dispatches; first three unstamped with distinct seqs; trailing one stamped Gate check: with trailing-fire scheduling disabled (pre-MOT-4211 drop semantics) the scenario times out at the await; with it, 6/6 passes. All 7 direct scenarios green. |
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@harness/tests/e2e/src/fixtures/loading.rs`:
- Around line 33-42: Update ScenarioFixture::validate() to reject
await_target_calls values of Some(0) and any positive await_target_calls when
scenario.target is absent. Preserve valid positive counts when a target exists,
and allow None as before so malformed fixtures fail before execution.
In `@harness/tests/e2e/src/scenarios/coalesced_fire.rs`:
- Around line 172-178: Update the trailing event assertion in the coalesced-fire
scenario to require trailing_seq to equal BURST exactly, preserving the existing
uniqueness check and error context so the test verifies newest-wins rather than
accepting any capped sequence.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: a8fbc104-fe85-448a-98e1-079355bc076f
📒 Files selected for processing (15)
harness/src/functions/react.rsharness/tests/e2e/README.mdharness/tests/e2e/src/evidence_data.rsharness/tests/e2e/src/fixtures/loading.rsharness/tests/e2e/src/fixtures/tests.rsharness/tests/e2e/src/probe.rsharness/tests/e2e/src/scenario/floor.rsharness/tests/e2e/src/scenario/phases/arm.rsharness/tests/e2e/src/scenario/phases/completion.rsharness/tests/e2e/src/scenario/phases/evidence.rsharness/tests/e2e/src/scenarios/coalesced_fire.rsharness/tests/e2e/src/scenarios/dsl.rsharness/tests/e2e/src/scenarios/mod.rsharness/tests/e2e/src/stack/supervisor.rsharness/tests/e2e/tests/determinism.rs
🚧 Files skipped from review as they are similar to previous changes (3)
- harness/tests/e2e/src/scenarios/dsl.rs
- harness/tests/e2e/src/scenario/phases/completion.rs
- harness/src/functions/react.rs
| /// Await this many controlled-function invocations (probe-side, whole-run) | ||
| /// after every terminal turn arrived. This is the only way to await work | ||
| /// that produces no session turn — a call-mode reaction's dispatch, | ||
| /// including a trailing coalesced fire. | ||
| pub await_target_calls: Option<usize>, | ||
| /// Session-trace-count override for the floor. `None` keeps the driver | ||
| /// formula. Needed when probe actions trip CALL-mode reactions, which | ||
| /// dispatch a function without seeding any session turn — no extra trace | ||
| /// tree ever forms. | ||
| pub traces_override: Option<usize>, |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Validate controlled-call wait configuration.
Line 37 permits Some(0), which later fails in wait_for_target_calls; a positive count without scenario.target instead waits until timeout. Reject both combinations in ScenarioFixture::validate() so malformed fixtures fail before execution.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@harness/tests/e2e/src/fixtures/loading.rs` around lines 33 - 42, Update
ScenarioFixture::validate() to reject await_target_calls values of Some(0) and
any positive await_target_calls when scenario.target is absent. Preserve valid
positive counts when a target exists, and allow None as before so malformed
fixtures fail before execution.
| let trailing_seq = seq_of(trailing) | ||
| .ok_or_else(|| anyhow::anyhow!("trailing fire has no seq: {trailing}"))?; | ||
| anyhow::ensure!( | ||
| (1..=BURST as u64).contains(&trailing_seq) && !seen.contains(&trailing_seq), | ||
| "the trailing event must be one of the CAPPED fires (newest wins), got seq \ | ||
| {trailing_seq} vs delivered {seen:?}" | ||
| ); |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Assert the newest capped event exactly.
This accepts sequence 4 or 5; with CAP = 3, both are capped. A first-parked implementation would pass even though it violates newest-wins. Require the trailing event to be sequence BURST.
Proposed fix
anyhow::ensure!(
- (1..=BURST as u64).contains(&trailing_seq) && !seen.contains(&trailing_seq),
+ trailing_seq == BURST as u64 && !seen.contains(&trailing_seq),
"the trailing event must be one of the CAPPED fires (newest wins), got seq \
{trailing_seq} vs delivered {seen:?}"
);📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| let trailing_seq = seq_of(trailing) | |
| .ok_or_else(|| anyhow::anyhow!("trailing fire has no seq: {trailing}"))?; | |
| anyhow::ensure!( | |
| (1..=BURST as u64).contains(&trailing_seq) && !seen.contains(&trailing_seq), | |
| "the trailing event must be one of the CAPPED fires (newest wins), got seq \ | |
| {trailing_seq} vs delivered {seen:?}" | |
| ); | |
| let trailing_seq = seq_of(trailing) | |
| .ok_or_else(|| anyhow::anyhow!("trailing fire has no seq: {trailing}"))?; | |
| anyhow::ensure!( | |
| trailing_seq == BURST as u64 && !seen.contains(&trailing_seq), | |
| "the trailing event must be one of the CAPPED fires (newest wins), got seq \ | |
| {trailing_seq} vs delivered {seen:?}" | |
| ); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@harness/tests/e2e/src/scenarios/coalesced_fire.rs` around lines 172 - 178,
Update the trailing event assertion in the coalesced-fire scenario to require
trailing_seq to equal BURST exactly, preserving the existing uniqueness check
and error context so the test verifies newest-wins rather than accepting any
capped sequence.
1 participant
What
Follow-up to the rctest-k7m3 postmortem (the wall of
reaction failed — fire-rate cap (10 per 60000ms) reached for this subscription; not spawning). Two changes:1. Capped fires coalesce instead of dropping
The fire-rate breaker (loop breaker #3, 10/60s per subscription) silently discarded capped fires. For recompute-style reactions that's the worst choice: a burst's tail is exactly what gets capped, so aggregates freeze one step behind the source of truth. Observed live: 15 orders in ~30s → 17 dropped reactions → totals stuck at 4/3/3 vs truth 5/5/5, never reconciled.
Now a capped fire parks in the gate slot (newest event wins, drops accumulate) and one trailing task per key re-enters
harness::reactwhen the window frees, delivering the latest event stamped with__coalesced_fires: N. The reaction outcome recordswaiting("coalescing — the latest event fires once the window frees (N collapsed so far)") instead offailed.The loop-breaker property is preserved by construction: at most one pending event and one trailing task per key, so a true runaway costs at most
MAX_FIRES_PER_WINDOW + 1reactions per window. If the window is still hot when the trailing fire lands, it parks again and a fresh cycle takes over.Known edge (accepted): a binding unregistered while holding a parked fire still delivers that one trailing reaction — consistent with the at-least-once delivery posture.
2. Registration advisory for key-less
statebindingsThe same run registered BOTH a keyed and a key-less state binding on one scope; the key-less one fires for EVERY key (done markers, completion signals, everything), doubling the budget burn. Registrations of
statebindings without akeyfilter now return an advisory note (react and notify paths) naming the scope and the cost — same posture as the existing standing-binding / join-wiring advisories.Verification
cargo test: 267 unit + 7 integration pass — new tests pin the coalescing contract (newest-wins, single scheduler per drain cycle, drop-count reset, GC never evicts parked fires) and the advisory shape (scope-wide vs global vs keyed vs non-state)Replaying rctest-k7m3 against this: the 17 capped fires collapse into 2 trailing reactions (one per binding) that fire ~60s in with the final events — totals converge to 5/5/5 — and both registrations would have carried the catch-all warning telling the agent to key-filter the second binding.
Summary by CodeRabbit