fix(benchmarks): wait for scripted event delivery (#6487)

## Why

The Buzz-native runtime could stop a solo agent after one turn even when
a scripted follow-up event was queued or already running. That could
turn a harness timing race into a benchmark failure.

Follow-up to #6448.

## What

- Treat solo tasks with scripted events separately from ordinary
one-turn tasks.
- Wait until turn counts and authored message IDs remain idle and
unchanged for several polls before stopping the agent.
- Remove the message-count shortcut that could stop an active turn.
- Add focused regressions for a delayed follow-up turn and an
already-running turn with a `DONE:` message.

## Testing

- `pytest -q tests/test_container_runtime.py -k "solo_turn_end or
scripted_events"` — 3 passed
- `ruff check` on the three touched Python files
- `ruff format --check` on the three touched Python files
- Local `buzz-native-solo-luna.yaml` Harbor run, one attempt per
affected task:
- `cross-thread-requests` — reward 1.0; all 6 task-specific checks
passed
- `interleaved-agent-reports` — reward 1.0; all 7 task-specific checks
passed
  - 2 completed trials, 0 exceptions, 56 seconds total

## Risk

Low. This only changes completion detection for solo benchmark tasks
with scripted events. Ordinary solo tasks retain immediate completion
after their first turn ends.

---
**Update Aug 21, 13:07 EDT:** Replaced the fixed settle delay with exact
delivery receipts after [review
feedback](https://github.com/block/buzz/pull/6487#discussion_r3832050452).

- ACP now records the event IDs delivered by each completed turn; the
runtime also recognizes existing successful-steer receipts.
- Scripted trials stop only after every expected event ID is
acknowledged and no turn is active. Missing receipts wait for the trial
budget instead of producing partial evidence.
- TDD regression delayed delivery beyond the old five-poll boundary and
failed before the implementation; the receipt parser and Rust receipt
format are pinned independently.
- Focused verification: 31 container-runtime tests passed, 6 ACP
delivery tests passed, plus Ruff, rustfmt, and Clippy.
- Receipt-gated Luna run: 2 completed trials and 0 exceptions.
`interleaved-agent-reports` scored 1.0. `cross-thread-requests`
completed both calculations and thread isolation but Luna omitted the
user mention on ALPHA, so that model-output dimension scored 0.
- Risk remains low: the ACP production change adds delivery receipt
logging and restores a missing turn-end log on a
completed-before-control race; it does not change queue dispatch
behavior.

Generated with Codex

---------

Signed-off-by: Salman Mohammed <smohammed@squareup.com>

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>) -> 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<String>,
+) {
+    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.