mirror of
https://github.com/shareAI-lab/analysis_claude_code.git
synced 2026-09-20 12:13:38 +08:00
fix(s15): harden background task lifecycle
This commit is contained in:
@@ -2114,9 +2114,17 @@ def start_background_task(block, handlers: dict) -> str:
|
|||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
result = f"Error: {type(exc).__name__}: {exc}"
|
result = f"Error: {type(exc).__name__}: {exc}"
|
||||||
status = "failed"
|
status = "failed"
|
||||||
|
try:
|
||||||
trigger_hooks("PostToolUse", block, result)
|
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:
|
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)
|
background_results[bg_id] = str(result)
|
||||||
|
|
||||||
with background_lock:
|
with background_lock:
|
||||||
@@ -2128,7 +2136,14 @@ def start_background_task(block, handlers: dict) -> str:
|
|||||||
"status": "running",
|
"status": "running",
|
||||||
"cwd": str(cwd) if cwd else None,
|
"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")
|
print(f" \033[33m[background] {bg_id}: {str(command)[:60]}\033[0m")
|
||||||
return bg_id
|
return bg_id
|
||||||
|
|
||||||
@@ -2989,9 +3004,13 @@ def agent_loop(messages: list, context: dict, active_request: str):
|
|||||||
continue
|
continue
|
||||||
|
|
||||||
if should_run_background(block.name, block.input):
|
if should_run_background(block.name, block.input):
|
||||||
|
try:
|
||||||
bg_id = start_background_task(block, handlers)
|
bg_id = start_background_task(block, handlers)
|
||||||
output = (f"[Background task {bg_id} started] "
|
output = (f"[Background task {bg_id} started] "
|
||||||
"Result will arrive as a task_notification.")
|
"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",
|
results.append({"type": "tool_result",
|
||||||
"tool_use_id": block.id,
|
"tool_use_id": block.id,
|
||||||
"content": output})
|
"content": output})
|
||||||
|
|||||||
@@ -1208,6 +1208,161 @@ class AgentTeamsRuntimeTests(unittest.TestCase):
|
|||||||
self.assertFalse(lesson.background_tasks)
|
self.assertFalse(lesson.background_tasks)
|
||||||
self.assertFalse(lesson.background_results)
|
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):
|
def test_teammate_survives_stale_worktree_assignment(self):
|
||||||
with tempfile.TemporaryDirectory() as tmp:
|
with tempfile.TemporaryDirectory() as tmp:
|
||||||
root = Path(tmp)
|
root = Path(tmp)
|
||||||
|
|||||||
File diff suppressed because one or more lines are too long
Reference in New Issue
Block a user