diff --git a/s15_integrated_harness/code.py b/s15_integrated_harness/code.py index bfefff58..ba315983 100644 --- a/s15_integrated_harness/code.py +++ b/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"\n" diff --git a/tests/test_agent_teams_runtime.py b/tests/test_agent_teams_runtime.py index 49d7fcbb..89f785ad 100644 --- a/tests/test_agent_teams_runtime.py +++ b/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)