Merge main into fix/micro-compact-latest-batch

s15_integrated_harness/code.py

@@ -2109,8 +2109,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()
 
@@ -2127,19 +2125,36 @@ def start_background_task(block, handlers: dict) -> str:
         except Exception as exc:
             result = f"Error: {type(exc).__name__}: {exc}"
             status = "failed"
+        try:
             trigger_hooks("PostToolUse", block, result)
+        except Exception as exc:
+            result = (f"Error: PostToolUse hook failed: "
+                      f"{type(exc).__name__}: {exc}\n{result}")
+            status = "failed"
         with background_lock:
-            background_tasks[bg_id]["status"] = status
+            task = background_tasks.get(bg_id)
+            if task is None:
+                return
+            task["status"] = status
             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,
             "status": "running",
             "cwd": str(cwd) if cwd else None,
         }
-    threading.Thread(target=worker, daemon=True).start()
+    thread = threading.Thread(target=worker, daemon=True)
+    try:
+        thread.start()
+    except Exception:
+        with background_lock:
+            background_tasks.pop(bg_id, None)
+            background_results.pop(bg_id, None)
+        raise
     print(f"  \033[33m[background] {bg_id}: {str(command)[:60]}\033[0m")
     return bg_id
 
@@ -2148,11 +2163,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:
+    for bg_id, task, output in completed:
-        with background_lock:
-            task = background_tasks.pop(bg_id)
-            output = background_results.pop(bg_id, "")
         summary = output[:200] if len(output) > 200 else output
         notifications.append(
             f"<task_notification>\n"
@@ -2998,9 +3015,13 @@ def agent_loop(messages: list, context: dict, active_request: str):
                 continue
 
             if should_run_background(block.name, block.input):
+                try:
                     bg_id = start_background_task(block, handlers)
                     output = (f"[Background task {bg_id} started] "
                               "Result will arrive as a task_notification.")
+                except Exception as exc:
+                    output = (f"Error: Failed to start background task: "
+                              f"{type(exc).__name__}: {exc}")
                 results.append({"type": "tool_result",
                                 "tool_use_id": block.id,
                                 "content": output})

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,202 @@ 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_s15_background_ids_are_allocated_atomically(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            lesson = load_lesson(
+                Path(tmp), ROOT / "s15_integrated_harness" / "code.py"
+            )
+
+            class TrackingLock:
+                def __init__(self):
+                    self.lock = threading.Lock()
+                    self.owner = None
+
+                def __enter__(self):
+                    self.lock.acquire()
+                    self.owner = threading.get_ident()
+                    return self
+
+                def __exit__(self, exc_type, exc_value, traceback):
+                    self.owner = None
+                    self.lock.release()
+
+                def held_by_current_thread(self):
+                    return self.owner == threading.get_ident()
+
+            class RaceAwareCounter:
+                def __init__(self, lock):
+                    self.lock = lock
+                    self.barrier = threading.Barrier(2)
+
+                def __add__(self, value):
+                    if not self.lock.held_by_current_thread():
+                        self.barrier.wait(timeout=2.0)
+                    return 1
+
+            blocks = [
+                types.SimpleNamespace(
+                    id=f"tool-{index}",
+                    name="bash",
+                    input={"command": f"printf {index}",
+                           "run_in_background": True},
+                )
+                for index in (1, 2)
+            ]
+            lock = TrackingLock()
+            lesson.background_lock = lock
+            lesson._bg_counter = RaceAwareCounter(lock)
+            lesson._run_bash_process = lambda *args, **kwargs: ("ok", 0)
+            with ThreadPoolExecutor(max_workers=2) as executor:
+                task_ids = list(executor.map(
+                    lambda block: lesson.start_background_task(block, {}),
+                    blocks,
+                ))
+
+            self.assertCountEqual(task_ids, ["bg_0001", "bg_0002"])
+            self.assertEqual(set(lesson.background_tasks),
+                             {"bg_0001", "bg_0002"})
+
+    def test_s15_background_hook_failure_reaches_terminal_state(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            lesson = load_lesson(
+                Path(tmp), ROOT / "s15_integrated_harness" / "code.py"
+            )
+            block = types.SimpleNamespace(
+                id="tool-hook",
+                name="bash",
+                input={"command": "printf ok", "run_in_background": True},
+            )
+            lesson._run_bash_process = lambda *args, **kwargs: ("ok", 0)
+
+            def fail_post_hook(event, *args):
+                if event == "PostToolUse":
+                    raise RuntimeError("hook failed")
+
+            lesson.trigger_hooks = fail_post_hook
+            bg_id = lesson.start_background_task(block, {})
+
+            self.assertTrue(wait_until(
+                lambda: lesson.background_tasks[bg_id]["status"] != "running"
+            ))
+            self.assertEqual(lesson.background_tasks[bg_id]["status"], "failed")
+            self.assertIn("PostToolUse hook failed",
+                          lesson.background_results[bg_id])
+            notification = lesson.collect_background_results()[0]
+            self.assertIn("<status>failed</status>", notification)
+            self.assertIn("PostToolUse hook failed", notification)
+
+    def test_s15_background_thread_start_failure_rolls_back_task(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            lesson = load_lesson(
+                Path(tmp), ROOT / "s15_integrated_harness" / "code.py"
+            )
+            block = types.SimpleNamespace(
+                id="tool-start",
+                name="bash",
+                input={"command": "printf ok", "run_in_background": True},
+            )
+
+            class FailingThread:
+                def __init__(self, *args, **kwargs):
+                    pass
+
+                def start(self):
+                    raise RuntimeError("cannot start thread")
+
+            with patch.object(lesson.threading, "Thread", FailingThread):
+                with self.assertRaisesRegex(RuntimeError, "cannot start thread"):
+                    lesson.start_background_task(block, {})
+
+            self.assertFalse(lesson.background_tasks)
+            self.assertFalse(lesson.background_results)
+
+    def test_s15_background_start_failure_becomes_tool_result(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            lesson = load_lesson(
+                Path(tmp), ROOT / "s15_integrated_harness" / "code.py"
+            )
+            block = types.SimpleNamespace(
+                type="tool_use",
+                id="tool-start",
+                name="bash",
+                input={"command": "printf ok", "run_in_background": True},
+            )
+            responses = iter([
+                types.SimpleNamespace(
+                    stop_reason="tool_use",
+                    content=[block],
+                ),
+                types.SimpleNamespace(
+                    stop_reason="end_turn",
+                    content=[types.SimpleNamespace(type="text", text="done")],
+                ),
+            ])
+            lesson.call_llm = lambda *args, **kwargs: next(responses)
+            lesson.trigger_hooks = lambda *args, **kwargs: None
+            lesson.remember_after_turn = lambda messages: None
+
+            def fail_start(*args, **kwargs):
+                raise RuntimeError("cannot start thread")
+
+            lesson.start_background_task = fail_start
+            messages = []
+            lesson.agent_loop(messages, {}, "run in background")
+
+            tool_results = [
+                item
+                for message in messages
+                if message.get("role") == "user"
+                and isinstance(message.get("content"), list)
+                for item in message["content"]
+                if item.get("type") == "tool_result"
+            ]
+            self.assertEqual(len(tool_results), 1)
+            self.assertIn("Failed to start background task",
+                          tool_results[0]["content"])
+            self.assertIn("cannot start thread", tool_results[0]["content"])
+
     def test_teammate_survives_stale_worktree_assignment(self):
         with tempfile.TemporaryDirectory() as tmp:
             root = Path(tmp)