fix(s15): atomically deliver background results

s15_integrated_harness/code.py

@@ -2098,8 +2098,6 @@ def should_run_background(tool_name: str, tool_input: dict) -> bool:
 
 def start_background_task(block, handlers: dict) -> str:
     global _bg_counter
-    _bg_counter += 1
-    bg_id = f"bg_{_bg_counter:04d}"
     command = block.input.get("command", block.name)
     cwd, cwd_error = _agent_cwd()
 
@@ -2122,6 +2120,8 @@ def start_background_task(block, handlers: dict) -> str:
             background_results[bg_id] = str(result)
 
     with background_lock:
+        _bg_counter += 1
+        bg_id = f"bg_{_bg_counter:04d}"
         background_tasks[bg_id] = {
             "tool_use_id": block.id,
             "command": command,
@@ -2137,11 +2137,13 @@ def collect_background_results() -> list[str]:
     with background_lock:
         ready = [bg_id for bg_id, task in background_tasks.items()
                  if task["status"] in {"completed", "failed"}]
+        completed = [
+            (bg_id, background_tasks.pop(bg_id),
+             background_results.pop(bg_id, ""))
+            for bg_id in ready
+        ]
     notifications = []
-    for bg_id in ready:
-        with background_lock:
-            task = background_tasks.pop(bg_id)
-            output = background_results.pop(bg_id, "")
+    for bg_id, task, output in completed:
         summary = output[:200] if len(output) > 200 else output
         notifications.append(
             f"<task_notification>\n"

tests/test_agent_teams_runtime.py

@@ -9,6 +9,7 @@ import threading
 import time
 import types
 import unittest
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 from unittest.mock import patch
 
@@ -1166,6 +1167,47 @@ class AgentTeamsRuntimeTests(unittest.TestCase):
             time.sleep(1.2)
             self.assertEqual(len(seen_messages), calls_after_delivery)
 
+    def test_s15_background_results_have_one_atomic_consumer(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            lesson = load_lesson(
+                Path(tmp), ROOT / "s15_integrated_harness" / "code.py"
+            )
+
+            class CoordinatedLock:
+                def __init__(self):
+                    self.lock = threading.Lock()
+                    self.barrier = threading.Barrier(2)
+                    self.local = threading.local()
+
+                def __enter__(self):
+                    self.lock.acquire()
+                    return self
+
+                def __exit__(self, exc_type, exc_value, traceback):
+                    self.lock.release()
+                    if not getattr(self.local, "coordinated", False):
+                        self.local.coordinated = True
+                        self.barrier.wait(timeout=2.0)
+
+            lesson.background_tasks["bg_0001"] = {
+                "tool_use_id": "tool-1",
+                "command": "pytest",
+                "status": "completed",
+            }
+            lesson.background_results["bg_0001"] = "all tests passed"
+            lesson.background_lock = CoordinatedLock()
+
+            with ThreadPoolExecutor(max_workers=2) as executor:
+                results = list(executor.map(
+                    lambda _: lesson.collect_background_results(), range(2)
+                ))
+
+            notifications = [note for batch in results for note in batch]
+            self.assertEqual(len(notifications), 1)
+            self.assertIn("all tests passed", notifications[0])
+            self.assertFalse(lesson.background_tasks)
+            self.assertFalse(lesson.background_results)
+
     def test_teammate_survives_stale_worktree_assignment(self):
         with tempfile.TemporaryDirectory() as tmp:
             root = Path(tmp)