From f5c90d161346a6cc6d15fcdce8772dab652c82b1 Mon Sep 17 00:00:00 2001 From: Salman Mohammed Date: Fri, 21 Aug 2026 12:08:21 -0400 Subject: [PATCH 1/2] fix(benchmarks): wait for scripted turns to settle Signed-off-by: Salman Mohammed --- .../container_runtime.py | 46 ++++++++----- .../harbor_buzz_orchestra/task_fixtures.py | 2 - .../tests/test_container_runtime.py | 65 +++++++++++++++---- 3 files changed, 84 insertions(+), 29 deletions(-) diff --git a/benchmarks/harbor-buzz-orchestra/src/harbor_buzz_orchestra/container_runtime.py b/benchmarks/harbor-buzz-orchestra/src/harbor_buzz_orchestra/container_runtime.py index e4415755de9..4c26ddb608d 100644 --- a/benchmarks/harbor-buzz-orchestra/src/harbor_buzz_orchestra/container_runtime.py +++ b/benchmarks/harbor-buzz-orchestra/src/harbor_buzz_orchestra/container_runtime.py @@ -49,7 +49,7 @@ FORWARDER_LOG = f"{REMOTE_LOGS}/relay-forwarder.log" # How many done-poll iterations between in-container liveness probes. LIVENESS_EVERY = 10 -SCRIPTED_EVENT_IDLE_POLLS = 5 +SCRIPTED_TURN_SETTLE_POLLS = 5 TRANSCRIPT_LIMIT = 1000 TURN_ENDED_MARKERS = ( "turn complete for", @@ -206,9 +206,7 @@ async def run( trial, agents + infra, solo=agents[0] if len(agents) == 1 else None, - minimum_agent_messages=fixture_for( - trial.task_name - ).minimum_agent_messages, + wait_for_scripted_turns=bool(scripted_events), ), timeout=manifest.trial_budget.timeout_seconds, ) @@ -506,16 +504,17 @@ async def _wait_for_done( trial: TrialHandle, agents: list[_Agent], solo: _Agent | None = None, - minimum_agent_messages: int = 0, + wait_for_scripted_turns: bool = False, ) -> dict[str, Any] | None: - """Observe until a team posts DONE or a solo agent finishes its one turn. + """Observe until a team posts DONE or a solo agent finishes its work. Observation only: the harness never speaks as any agent. If the team stalls, the trial times out and the stall is the measured result. A solo - agent cannot be woken by a teammate, so its logged turn end is final. + task without scripted events finishes at its first logged turn end. """ polls = 0 - idle_terminal_polls = 0 + settled_polls = 0 + previous_scripted_state: tuple[int, int, tuple[str, ...]] | None = None while True: if polls % LIVENESS_EVERY == 0: await self._raise_for_dead_agents(environment, agents) @@ -531,9 +530,11 @@ async def _wait_for_done( "100", ) for message in messages: - if message.get("pubkey") == orchestrator.nostr_pubkey and str( - message.get("content", "") - ).startswith("DONE:"): + if ( + message.get("pubkey") == orchestrator.nostr_pubkey + and str(message.get("content", "")).startswith("DONE:") + and (solo is None or not wait_for_scripted_turns) + ): return message if solo is not None: starts, ends = await self._turn_counts(environment, solo) @@ -542,14 +543,27 @@ async def _wait_for_done( for message in messages if message.get("pubkey") == orchestrator.nostr_pubkey ] - if ends > 0 and len(authored) >= minimum_agent_messages: + if not wait_for_scripted_turns and ends > 0: return authored[-1] if authored else None - if starts > 0 and starts == ends: - idle_terminal_polls += 1 - if idle_terminal_polls >= SCRIPTED_EVENT_IDLE_POLLS: + + scripted_state = ( + starts, + ends, + tuple(sorted(str(message.get("id", "")) for message in authored)), + ) + if wait_for_scripted_turns and starts > 0 and starts == ends: + # A queued scripted event can start a new turn just after one ends. + # Stop only after the idle state stays unchanged for several polls. + settled_polls = ( + settled_polls + 1 + if scripted_state == previous_scripted_state + else 1 + ) + if settled_polls >= SCRIPTED_TURN_SETTLE_POLLS: return authored[-1] if authored else None else: - idle_terminal_polls = 0 + settled_polls = 0 + previous_scripted_state = scripted_state await asyncio.sleep(self.poll_seconds) @staticmethod diff --git a/benchmarks/harbor-buzz-orchestra/src/harbor_buzz_orchestra/task_fixtures.py b/benchmarks/harbor-buzz-orchestra/src/harbor_buzz_orchestra/task_fixtures.py index da94871beac..451b97f9c12 100644 --- a/benchmarks/harbor-buzz-orchestra/src/harbor_buzz_orchestra/task_fixtures.py +++ b/benchmarks/harbor-buzz-orchestra/src/harbor_buzz_orchestra/task_fixtures.py @@ -38,7 +38,6 @@ class BuzzTaskFixture: directory: tuple[DirectoryEntry, ...] = () scripted_messages: tuple[ScriptedMessage, ...] = () - minimum_agent_messages: int = 0 observe_channel_names: tuple[str, ...] = () user_display_name: str | None = None # Whether the task's verifier grades the exported relay snapshot. Only @@ -137,7 +136,6 @@ class BuzzTaskFixture: ), ), ), - minimum_agent_messages=2, user_display_name="Priya Simone Patel", requires_evidence=True, ) diff --git a/benchmarks/harbor-buzz-orchestra/tests/test_container_runtime.py b/benchmarks/harbor-buzz-orchestra/tests/test_container_runtime.py index 53bf268499e..4019efdf5b8 100644 --- a/benchmarks/harbor-buzz-orchestra/tests/test_container_runtime.py +++ b/benchmarks/harbor-buzz-orchestra/tests/test_container_runtime.py @@ -13,6 +13,7 @@ REMOTE_BIN, REMOTE_EVIDENCE, REMOTE_LOGS, + SCRIPTED_TURN_SETTLE_POLLS, THINKING_EFFORT, BuzzContainerRuntime, EndpointLaunchConfig, @@ -549,26 +550,29 @@ async def buzz_json(*args, **kwargs): assert await rt._wait_for_done(environment, orch, trial, [], solo=solo) is None -async def test_scripted_events_wait_for_second_agent_message(tmp_path, monkeypatch): +async def test_scripted_events_wait_for_delayed_follow_up_turn(tmp_path, monkeypatch): from harbor_buzz_orchestra.container_runtime import _Agent rt = runtime(tmp_path, poll_seconds=0) orch = credential("orch-1", "orchestrator", "orch-model") trial = trial_handle((orch,)) solo = _Agent(orch, 7, "stdout.log", "stderr.log") + alpha = {"id": "alpha", "pubkey": orch.nostr_pubkey, "content": "ALPHA"} + beta = {"id": "beta", "pubkey": orch.nostr_pubkey, "content": "BETA"} message_rounds = iter( - [ - [{"id": "alpha", "pubkey": orch.nostr_pubkey, "content": "ALPHA"}], - [{"id": "alpha", "pubkey": orch.nostr_pubkey, "content": "ALPHA"}], - [ - {"id": "alpha", "pubkey": orch.nostr_pubkey, "content": "ALPHA"}, - {"id": "beta", "pubkey": orch.nostr_pubkey, "content": "BETA"}, - ], - ] + [[alpha]] * SCRIPTED_TURN_SETTLE_POLLS + + [[alpha, beta]] * SCRIPTED_TURN_SETTLE_POLLS ) - turn_rounds = iter([(1, 1), (2, 1), (2, 2)]) + turn_rounds = iter( + [(1, 1)] * (SCRIPTED_TURN_SETTLE_POLLS - 1) + + [(2, 1)] + + [(2, 2)] * SCRIPTED_TURN_SETTLE_POLLS + ) + polls = 0 async def buzz_json(*args, **kwargs): + nonlocal polls + polls += 1 return next(message_rounds) async def turn_counts(*args, **kwargs): @@ -583,10 +587,49 @@ async def turn_counts(*args, **kwargs): trial, [], solo=solo, - minimum_agent_messages=2, + wait_for_scripted_turns=True, + ) + + assert result["id"] == "beta" + assert polls == SCRIPTED_TURN_SETTLE_POLLS * 2 + + +async def test_scripted_events_do_not_stop_an_active_turn(tmp_path, monkeypatch): + from harbor_buzz_orchestra.container_runtime import _Agent + + rt = runtime(tmp_path, poll_seconds=0) + orch = credential("orch-1", "orchestrator", "orch-model") + trial = trial_handle((orch,)) + solo = _Agent(orch, 7, "stdout.log", "stderr.log") + messages = [ + {"id": "alpha", "pubkey": orch.nostr_pubkey, "content": "ALPHA"}, + {"id": "beta", "pubkey": orch.nostr_pubkey, "content": "DONE: BETA"}, + ] + turn_rounds = iter([(2, 1)] + [(2, 2)] * SCRIPTED_TURN_SETTLE_POLLS) + polls = 0 + + async def buzz_json(*args, **kwargs): + nonlocal polls + polls += 1 + return messages + + async def turn_counts(*args, **kwargs): + return next(turn_rounds) + + monkeypatch.setattr(rt, "_buzz_json", buzz_json) + monkeypatch.setattr(rt, "_turn_counts", turn_counts) + + result = await rt._wait_for_done( + Environment(), + orch, + trial, + [], + solo=solo, + wait_for_scripted_turns=True, ) assert result["id"] == "beta" + assert polls == SCRIPTED_TURN_SETTLE_POLLS + 1 async def test_collect_evidence_uploads_verifier_artifact(tmp_path, monkeypatch): From 9eb291b6c479a2273e1b42014e6e6a04b4e252ef Mon Sep 17 00:00:00 2001 From: Salman Mohammed Date: Fri, 21 Aug 2026 13:07:23 -0400 Subject: [PATCH 2/2] fix(benchmarks): require scripted delivery receipts Signed-off-by: Salman Mohammed --- .../container_runtime.py | 74 +++++++++++------- .../tests/test_container_runtime.py | 78 ++++++++++++++----- crates/buzz-acp/src/pool.rs | 49 +++++++++++- 3 files changed, 151 insertions(+), 50 deletions(-) diff --git a/benchmarks/harbor-buzz-orchestra/src/harbor_buzz_orchestra/container_runtime.py b/benchmarks/harbor-buzz-orchestra/src/harbor_buzz_orchestra/container_runtime.py index 4c26ddb608d..797e3a860c2 100644 --- a/benchmarks/harbor-buzz-orchestra/src/harbor_buzz_orchestra/container_runtime.py +++ b/benchmarks/harbor-buzz-orchestra/src/harbor_buzz_orchestra/container_runtime.py @@ -13,6 +13,7 @@ import asyncio import json import os +import re import shlex import traceback from dataclasses import dataclass, field @@ -49,8 +50,10 @@ FORWARDER_LOG = f"{REMOTE_LOGS}/relay-forwarder.log" # How many done-poll iterations between in-container liveness probes. LIVENESS_EVERY = 10 -SCRIPTED_TURN_SETTLE_POLLS = 5 TRANSCRIPT_LIMIT = 1000 +DELIVERY_RECEIPT_MARKER = "turn delivered Buzz events for channel" +EVENT_ID_PATTERN = re.compile(r"(? dict[str, Any] | None: """Observe until a team posts DONE or a solo agent finishes its work. @@ -513,8 +518,6 @@ async def _wait_for_done( task without scripted events finishes at its first logged turn end. """ polls = 0 - settled_polls = 0 - previous_scripted_state: tuple[int, int, tuple[str, ...]] | None = None while True: if polls % LIVENESS_EVERY == 0: await self._raise_for_dead_agents(environment, agents) @@ -533,37 +536,26 @@ async def _wait_for_done( if ( message.get("pubkey") == orchestrator.nostr_pubkey and str(message.get("content", "")).startswith("DONE:") - and (solo is None or not wait_for_scripted_turns) + and (solo is None or not scripted_event_ids) ): return message if solo is not None: - starts, ends = await self._turn_counts(environment, solo) + starts, ends, delivered_event_ids = await self._turn_status( + environment, solo + ) authored = [ message for message in messages if message.get("pubkey") == orchestrator.nostr_pubkey ] - if not wait_for_scripted_turns and ends > 0: + if not scripted_event_ids and ends > 0: + return authored[-1] if authored else None + if ( + starts > 0 + and starts == ends + and scripted_event_ids <= delivered_event_ids + ): return authored[-1] if authored else None - - scripted_state = ( - starts, - ends, - tuple(sorted(str(message.get("id", "")) for message in authored)), - ) - if wait_for_scripted_turns and starts > 0 and starts == ends: - # A queued scripted event can start a new turn just after one ends. - # Stop only after the idle state stays unchanged for several polls. - settled_polls = ( - settled_polls + 1 - if scripted_state == previous_scripted_state - else 1 - ) - if settled_polls >= SCRIPTED_TURN_SETTLE_POLLS: - return authored[-1] if authored else None - else: - settled_polls = 0 - previous_scripted_state = scripted_state await asyncio.sleep(self.poll_seconds) @staticmethod @@ -575,14 +567,36 @@ async def _turn_ended(environment: BaseEnvironment, agent: _Agent) -> bool: async def _turn_counts( environment: BaseEnvironment, agent: _Agent ) -> tuple[int, int]: + starts, ends, _ = await BuzzContainerRuntime._turn_status(environment, agent) + return starts, ends + + @staticmethod + async def _turn_status( + environment: BaseEnvironment, agent: _Agent + ) -> tuple[int, int, set[str]]: result = await environment.exec( f"cat {shlex.quote(agent.stdout_log)} " f"{shlex.quote(agent.stderr_log)} 2>/dev/null" ) - output = result.stdout or "" + return BuzzContainerRuntime._parse_turn_status(result.stdout or "") + + @staticmethod + def _parse_turn_status(output: str) -> tuple[int, int, set[str]]: + output = ANSI_ESCAPE_PATTERN.sub("", output) + delivered_event_ids: set[str] = set() + for line in output.splitlines(): + if DELIVERY_RECEIPT_MARKER in line: + delivered_event_ids.update(EVENT_ID_PATTERN.findall(line)) + elif ( + "non-cancelling steer ack received" in line and "ack=Ok(Success" in line + ): + match = re.search(r"event_id=([0-9a-f]{64})", line) + if match is not None: + delivered_event_ids.add(match.group(1)) return ( output.count("turn starting for"), sum(output.count(marker) for marker in TURN_ENDED_MARKERS), + delivered_event_ids, ) async def _raise_for_dead_agents( @@ -838,6 +852,10 @@ async def _send_scripted_messages( and isinstance(response.get("event_id"), str) else None ) + if event_id is None: + raise RuntimeLaunchError( + f"scripted event {message.label!r} did not return an event ID" + ) recorded.append( { "label": message.label, diff --git a/benchmarks/harbor-buzz-orchestra/tests/test_container_runtime.py b/benchmarks/harbor-buzz-orchestra/tests/test_container_runtime.py index 4019efdf5b8..182db9893f6 100644 --- a/benchmarks/harbor-buzz-orchestra/tests/test_container_runtime.py +++ b/benchmarks/harbor-buzz-orchestra/tests/test_container_runtime.py @@ -13,7 +13,6 @@ REMOTE_BIN, REMOTE_EVIDENCE, REMOTE_LOGS, - SCRIPTED_TURN_SETTLE_POLLS, THINKING_EFFORT, BuzzContainerRuntime, EndpointLaunchConfig, @@ -503,6 +502,25 @@ async def send(actor, trial_arg, content, **kwargs): assert all(call[3]["reply_to"] == "task-root" for call in calls) +async def test_scripted_message_requires_an_event_id(tmp_path, monkeypatch): + rt = runtime(tmp_path) + orch = credential("solo-1", "orchestrator", "orch-model") + trial = replace( + trial_handle((orch,)), + task_name="cross-thread-requests", + ) + + async def send(*args, **kwargs): + return {} + + monkeypatch.setattr(rt, "_send", send) + + with pytest.raises(RuntimeLaunchError, match="did not return an event ID"): + await rt._send_scripted_messages( + trial=trial, orchestrator=orch, task_event_id="task-root" + ) + + async def test_wait_for_done_requires_orchestrator_authorship(tmp_path, monkeypatch): rt = runtime(tmp_path, poll_seconds=0) orch = credential("orch-1", "orchestrator", "orch-model") @@ -550,7 +568,7 @@ async def buzz_json(*args, **kwargs): assert await rt._wait_for_done(environment, orch, trial, [], solo=solo) is None -async def test_scripted_events_wait_for_delayed_follow_up_turn(tmp_path, monkeypatch): +async def test_scripted_events_wait_for_delivery_receipt(tmp_path, monkeypatch): from harbor_buzz_orchestra.container_runtime import _Agent rt = runtime(tmp_path, poll_seconds=0) @@ -559,14 +577,10 @@ async def test_scripted_events_wait_for_delayed_follow_up_turn(tmp_path, monkeyp solo = _Agent(orch, 7, "stdout.log", "stderr.log") alpha = {"id": "alpha", "pubkey": orch.nostr_pubkey, "content": "ALPHA"} beta = {"id": "beta", "pubkey": orch.nostr_pubkey, "content": "BETA"} - message_rounds = iter( - [[alpha]] * SCRIPTED_TURN_SETTLE_POLLS - + [[alpha, beta]] * SCRIPTED_TURN_SETTLE_POLLS - ) + scripted_event_id = "b" * 64 + message_rounds = iter([[alpha]] * 8 + [[alpha, beta]] * 2) turn_rounds = iter( - [(1, 1)] * (SCRIPTED_TURN_SETTLE_POLLS - 1) - + [(2, 1)] - + [(2, 2)] * SCRIPTED_TURN_SETTLE_POLLS + [(1, 1, set())] * 8 + [(2, 1, set()), (2, 2, {scripted_event_id})] ) polls = 0 @@ -575,11 +589,11 @@ async def buzz_json(*args, **kwargs): polls += 1 return next(message_rounds) - async def turn_counts(*args, **kwargs): + async def turn_status(*args, **kwargs): return next(turn_rounds) monkeypatch.setattr(rt, "_buzz_json", buzz_json) - monkeypatch.setattr(rt, "_turn_counts", turn_counts) + monkeypatch.setattr(rt, "_turn_status", turn_status) result = await rt._wait_for_done( Environment(), @@ -587,11 +601,11 @@ async def turn_counts(*args, **kwargs): trial, [], solo=solo, - wait_for_scripted_turns=True, + scripted_event_ids={scripted_event_id}, ) assert result["id"] == "beta" - assert polls == SCRIPTED_TURN_SETTLE_POLLS * 2 + assert polls == 10 async def test_scripted_events_do_not_stop_an_active_turn(tmp_path, monkeypatch): @@ -605,7 +619,8 @@ async def test_scripted_events_do_not_stop_an_active_turn(tmp_path, monkeypatch) {"id": "alpha", "pubkey": orch.nostr_pubkey, "content": "ALPHA"}, {"id": "beta", "pubkey": orch.nostr_pubkey, "content": "DONE: BETA"}, ] - turn_rounds = iter([(2, 1)] + [(2, 2)] * SCRIPTED_TURN_SETTLE_POLLS) + scripted_event_id = "b" * 64 + turn_rounds = iter([(2, 1, {scripted_event_id}), (2, 2, {scripted_event_id})]) polls = 0 async def buzz_json(*args, **kwargs): @@ -613,11 +628,11 @@ async def buzz_json(*args, **kwargs): polls += 1 return messages - async def turn_counts(*args, **kwargs): + async def turn_status(*args, **kwargs): return next(turn_rounds) monkeypatch.setattr(rt, "_buzz_json", buzz_json) - monkeypatch.setattr(rt, "_turn_counts", turn_counts) + monkeypatch.setattr(rt, "_turn_status", turn_status) result = await rt._wait_for_done( Environment(), @@ -625,11 +640,38 @@ async def turn_counts(*args, **kwargs): trial, [], solo=solo, - wait_for_scripted_turns=True, + scripted_event_ids={scripted_event_id}, ) assert result["id"] == "beta" - assert polls == SCRIPTED_TURN_SETTLE_POLLS + 1 + assert polls == 2 + + +def test_turn_status_parses_completed_batch_and_successful_steer_receipts(): + batch_event_id = "a" * 64 + steer_event_id = "b" * 64 + rejected_event_id = "c" * 64 + output = "\n".join( + [ + "turn starting for channel test", + f"turn delivered Buzz events for channel test: {batch_event_id}", + "turn complete for channel test: end_turn", + ( + "non-cancelling steer ack received " + f"event_id={steer_event_id} ack=Ok(Success {{ session_id: session }})" + ), + ( + "non-cancelling steer ack received " + f"event_id={rejected_event_id} ack=Ok(Err(OutcomeRejected))" + ), + ] + ) + + assert BuzzContainerRuntime._parse_turn_status(output) == ( + 1, + 1, + {batch_event_id, steer_event_id}, + ) async def test_collect_evidence_uploads_verifier_artifact(tmp_path, monkeypatch): diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 2c173646bac..f3916b5fdab 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -2619,12 +2619,14 @@ pub async fn run_prompt_task( "control signal arrived but turn already completed — treating as success" ); } + log_stop_reason(&source, &StopReason::EndTurn); if let PromptSource::Channel(cid) = &source { let standing_sent = !agent.has_system_prompt_support(); - agent.state.mark_channel_delivery_success( + record_channel_delivery_success( + &mut agent, *cid, standing_sent, - pending_delivered_event_ids.iter().cloned(), + &pending_delivered_event_ids, ); } apply_completed_before_control_signal( @@ -2663,10 +2665,11 @@ pub async fn run_prompt_task( if let PromptSource::Channel(cid) = &source { let standing_sent = !agent.has_system_prompt_support(); - agent.state.mark_channel_delivery_success( + record_channel_delivery_success( + &mut agent, *cid, standing_sent, - pending_delivered_event_ids.iter().cloned(), + &pending_delivered_event_ids, ); } else if !agent.has_system_prompt_support() { agent.state.heartbeat_standing_context_sent = true; @@ -4085,6 +4088,33 @@ fn log_stop_reason(source: &PromptSource, stop_reason: &StopReason) { } } +fn delivery_receipt_line(channel_id: Uuid, event_ids: &HashSet) -> String { + let mut event_ids: Vec<&str> = event_ids.iter().map(String::as_str).collect(); + event_ids.sort_unstable(); + format!( + "turn delivered Buzz events for channel {channel_id}: {}", + event_ids.join(",") + ) +} + +fn record_channel_delivery_success( + agent: &mut OwnedAgent, + channel_id: Uuid, + standing_context_sent: bool, + event_ids: &HashSet, +) { + tracing::info!( + target: "pool::prompt", + "{}", + delivery_receipt_line(channel_id, event_ids) + ); + agent.state.mark_channel_delivery_success( + channel_id, + standing_context_sent, + event_ids.iter().cloned(), + ); +} + // // Two-phase lifecycle visible to users: // 👀 "seen" — event was queued and an agent will handle it @@ -4709,6 +4739,17 @@ mod tests { } } + #[test] + fn delivery_receipt_line_sorts_event_ids() { + let channel_id = Uuid::nil(); + let event_ids = HashSet::from(["beta".to_string(), "alpha".to_string()]); + + assert_eq!( + delivery_receipt_line(channel_id, &event_ids), + format!("turn delivered Buzz events for channel {channel_id}: alpha,beta") + ); + } + // MINOR (#2884): the permission-mode RPC is gated on agent_supports_mode. // An advertised mode issues set_config_option; an absent one is skipped so // the harness falls back to per-tool auto-approval. Pin both edges directly.