mirror of
https://github.com/shareAI-lab/learn-claude-code.git
synced 2026-08-26 08:42:12 +00:00
1786 lines
77 KiB
Python
1786 lines
77 KiB
Python
import importlib.util
|
|
import multiprocessing
|
|
import os
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import types
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
LESSON = ROOT / "s13_agent_teams" / "code.py"
|
|
DOWNSTREAM_LESSONS = (
|
|
ROOT / "s15_integrated_harness" / "code.py",
|
|
)
|
|
RUNTIME_LESSONS = (LESSON, *DOWNSTREAM_LESSONS)
|
|
MCP_LESSONS = (
|
|
ROOT / "s14_mcp_plugin" / "code.py",
|
|
ROOT / "s15_integrated_harness" / "code.py",
|
|
)
|
|
BACKGROUND_LESSONS = tuple(
|
|
ROOT / name / "code.py" for name in (
|
|
"s11_background_tasks",
|
|
"s15_integrated_harness",
|
|
)
|
|
)
|
|
CRON_LESSONS = tuple(
|
|
ROOT / name / "code.py" for name in (
|
|
"s12_cron_scheduler",
|
|
"s15_integrated_harness",
|
|
)
|
|
)
|
|
|
|
|
|
def load_lesson(temp_cwd: Path, lesson_path: Path = LESSON):
|
|
fake_anthropic = types.ModuleType("anthropic")
|
|
fake_yaml = types.ModuleType("yaml")
|
|
|
|
class FakeAnthropic:
|
|
def __init__(self, *args, **kwargs):
|
|
self.messages = types.SimpleNamespace(create=None)
|
|
|
|
fake_dotenv = types.ModuleType("dotenv")
|
|
setattr(fake_anthropic, "Anthropic", FakeAnthropic)
|
|
setattr(fake_dotenv, "load_dotenv", lambda override=True: None)
|
|
setattr(fake_yaml, "safe_load", lambda value: {})
|
|
setattr(fake_yaml, "YAMLError", ValueError)
|
|
|
|
previous_modules = {
|
|
"anthropic": sys.modules.get("anthropic"),
|
|
"dotenv": sys.modules.get("dotenv"),
|
|
"yaml": sys.modules.get("yaml"),
|
|
}
|
|
previous_cwd = Path.cwd()
|
|
previous_model = os.environ.get("MODEL_ID")
|
|
|
|
name = f"agent_teams_test_{lesson_path.parent.name}_{time.time_ns()}"
|
|
spec = importlib.util.spec_from_file_location(name, lesson_path)
|
|
if spec is None or spec.loader is None:
|
|
raise RuntimeError(f"Unable to load {lesson_path}")
|
|
module = importlib.util.module_from_spec(spec)
|
|
|
|
sys.modules["anthropic"] = fake_anthropic
|
|
sys.modules["dotenv"] = fake_dotenv
|
|
sys.modules["yaml"] = fake_yaml
|
|
sys.modules[name] = module
|
|
try:
|
|
os.chdir(temp_cwd)
|
|
os.environ["MODEL_ID"] = "test-model"
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
finally:
|
|
os.chdir(previous_cwd)
|
|
if previous_model is None:
|
|
os.environ.pop("MODEL_ID", None)
|
|
else:
|
|
os.environ["MODEL_ID"] = previous_model
|
|
for module_name, previous in previous_modules.items():
|
|
if previous is None:
|
|
sys.modules.pop(module_name, None)
|
|
else:
|
|
sys.modules[module_name] = previous
|
|
|
|
|
|
def wait_until(predicate, timeout=2.0):
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
if predicate():
|
|
return True
|
|
time.sleep(0.01)
|
|
return False
|
|
|
|
|
|
def init_git_repo(root: Path):
|
|
subprocess.run(
|
|
["git", "init", "-q", "-b", "main"], cwd=root, check=True
|
|
)
|
|
subprocess.run(
|
|
["git", "config", "user.email", "tests@example.com"],
|
|
cwd=root, check=True,
|
|
)
|
|
subprocess.run(
|
|
["git", "config", "user.name", "Runtime Tests"],
|
|
cwd=root, check=True,
|
|
)
|
|
(root / "tracked.txt").write_text("initial\n")
|
|
subprocess.run(["git", "add", "tracked.txt"], cwd=root, check=True)
|
|
subprocess.run(
|
|
["git", "commit", "-q", "-m", "initial"], cwd=root, check=True
|
|
)
|
|
|
|
|
|
def claim_in_child(lesson_path: str, root: str, task_id: str, owner: str,
|
|
barrier, results):
|
|
lesson = load_lesson(Path(root), Path(lesson_path))
|
|
barrier.wait()
|
|
results.put(lesson.claim_task(task_id, owner=owner))
|
|
|
|
|
|
class AgentTeamsRuntimeTests(unittest.TestCase):
|
|
def test_downstream_lessons_execute_the_merged_runtime_contract(self):
|
|
for lesson_path in RUNTIME_LESSONS:
|
|
with self.subTest(lesson=lesson_path.parent.name):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), lesson_path)
|
|
task = lesson.create_task("Runtime contract")
|
|
self.assertIn(
|
|
"Claimed", lesson.claim_task(task.id, owner="alice")
|
|
)
|
|
self.assertIn(
|
|
"Completed", lesson.complete_task(task.id, owner="alice")
|
|
)
|
|
self.assertIn("alice", lesson.teammate_assignments)
|
|
self.assertTrue(lesson.release_completed_assignment("alice"))
|
|
self.assertNotIn("alice", lesson.teammate_assignments)
|
|
|
|
def test_inbox_delivery_is_runtime_owned(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp))
|
|
|
|
tool_names = {tool["name"] for tool in lesson.TOOLS}
|
|
self.assertNotIn("check_inbox", tool_names)
|
|
self.assertIn("create_worktree", tool_names)
|
|
self.assertNotIn("remove_worktree", tool_names)
|
|
self.assertNotIn("keep_worktree", tool_names)
|
|
worktree_tools = {
|
|
tool["name"]: tool["input_schema"] for tool in lesson.TOOLS
|
|
if tool["name"] == "create_worktree"
|
|
}
|
|
for schema in worktree_tools.values():
|
|
self.assertFalse(schema["additionalProperties"])
|
|
self.assertEqual(schema["properties"]["name"]["maxLength"], 64)
|
|
self.assertIn("wait for the user's confirmation",
|
|
lesson.PROMPT_SECTIONS["teams"])
|
|
self.assertIn("creating a Task", lesson.PROMPT_SECTIONS["teams"])
|
|
self.assertIn("not a sandbox", lesson.PROMPT_SECTIONS["teams"])
|
|
|
|
lesson.BUS.send("alice", "lead", "done", "result")
|
|
events = lesson.consume_lead_inbox()
|
|
|
|
self.assertEqual([event["type"] for event in events], ["result"])
|
|
self.assertIn("[result] alice: done",
|
|
lesson.format_team_events(events))
|
|
|
|
def test_spawn_claims_the_initial_task_before_starting_the_thread(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp))
|
|
task = lesson.create_task("Review authentication")
|
|
schema = next(
|
|
tool["input_schema"] for tool in lesson.TOOLS
|
|
if tool["name"] == "spawn_teammate"
|
|
)
|
|
self.assertIn("task_id", schema["properties"])
|
|
|
|
with patch.object(
|
|
lesson.threading.Thread, "start", lambda _thread: None
|
|
):
|
|
result = lesson.spawn_teammate_thread(
|
|
"alice", "reviewer", "Review the assigned Task.", task.id
|
|
)
|
|
|
|
self.assertIn(task.id, result)
|
|
claimed = lesson.load_task(task.id)
|
|
self.assertEqual(claimed.status, "in_progress")
|
|
self.assertEqual(claimed.owner, "alice")
|
|
self.assertEqual(
|
|
lesson.teammate_assignments["alice"]["task_id"], task.id
|
|
)
|
|
|
|
def test_spawn_allows_an_idle_teammate_without_an_initial_task(self):
|
|
for lesson_path in RUNTIME_LESSONS:
|
|
with self.subTest(lesson=lesson_path.parent.name):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), lesson_path)
|
|
tool_defs = getattr(lesson, "TOOLS", None)
|
|
if tool_defs is None:
|
|
tool_defs = lesson.BUILTIN_TOOLS
|
|
schema = next(
|
|
tool["input_schema"] for tool in tool_defs
|
|
if tool["name"] == "spawn_teammate"
|
|
)
|
|
self.assertNotIn("task_id", schema["required"])
|
|
|
|
with patch.object(
|
|
lesson.threading.Thread, "start", lambda _thread: None
|
|
):
|
|
result = lesson.run_spawn_teammate(
|
|
"alice", "reviewer", "Wait for a ready Task."
|
|
)
|
|
|
|
self.assertIn("without an initial Task", result)
|
|
self.assertNotIn("alice", lesson.teammate_assignments)
|
|
|
|
def test_teammate_workspace_tools_require_a_claimed_task(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
lesson = load_lesson(root)
|
|
runtime = lesson.TeammateRuntime(
|
|
"alice", "reviewer", "Inspect the project.", None, False
|
|
)
|
|
|
|
result = runtime.write("unassigned.txt", "must not be written")
|
|
|
|
self.assertIn("Claim a Task", result)
|
|
self.assertFalse((root / "unassigned.txt").exists())
|
|
|
|
def test_plain_message_does_not_change_assignment_or_plan_version(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp))
|
|
lesson.active_teammates["alice"] = "working"
|
|
lesson.plan_gates["alice"] = "approved"
|
|
lesson.assignment_versions["alice"] = 3
|
|
|
|
self.assertIn("Sent", lesson.run_send_message("alice", "Continue."))
|
|
|
|
self.assertEqual(lesson.plan_gates["alice"], "approved")
|
|
self.assertEqual(lesson.assignment_versions["alice"], 3)
|
|
|
|
def test_worktree_removal_is_host_only(self):
|
|
for lesson_path in RUNTIME_LESSONS:
|
|
with self.subTest(lesson=lesson_path.parent.name):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), lesson_path)
|
|
tool_defs = getattr(lesson, "TOOLS", None)
|
|
if tool_defs is None:
|
|
tool_defs = lesson.BUILTIN_TOOLS
|
|
self.assertNotIn(
|
|
"remove_worktree",
|
|
{tool["name"] for tool in tool_defs},
|
|
)
|
|
self.assertTrue(callable(lesson.remove_worktree))
|
|
self.assertFalse(hasattr(lesson, "run_remove_worktree"))
|
|
|
|
def test_agent_teams_builds_on_tasks_not_background_or_cron(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
lesson = load_lesson(root)
|
|
tool_names = {tool["name"] for tool in lesson.TOOLS}
|
|
|
|
self.assertTrue({
|
|
"bash", "read_file", "write_file", "edit_file", "glob",
|
|
"create_task", "list_tasks", "get_task", "claim_task",
|
|
"complete_task", "spawn_teammate", "list_teammates",
|
|
"send_message", "request_shutdown", "request_plan",
|
|
"review_plan", "create_worktree",
|
|
}.issubset(tool_names))
|
|
self.assertTrue({
|
|
"schedule_cron", "list_crons", "cancel_cron",
|
|
}.isdisjoint(tool_names))
|
|
self.assertNotIn("run_in_background", next(
|
|
tool["input_schema"] for tool in lesson.TOOLS
|
|
if tool["name"] == "bash"
|
|
)["properties"])
|
|
self.assertFalse((root / ".tasks").exists())
|
|
self.assertFalse((root / ".mailboxes").exists())
|
|
self.assertFalse((root / ".worktrees").exists())
|
|
|
|
def test_integrated_harness_reuses_memory_recall_and_extraction(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), DOWNSTREAM_LESSONS[0])
|
|
calls = []
|
|
lesson.MEMORY_RUNTIME = types.SimpleNamespace(
|
|
read_memory_index=lambda: "- [Style](style.md) - Project style",
|
|
load_memories=lambda messages: (
|
|
calls.append(("recall", list(messages)))
|
|
or '[{"source":"style.md","content":"Use black."}]'
|
|
),
|
|
extract_memories=lambda messages: (
|
|
calls.append(("extract", list(messages))) or 1
|
|
),
|
|
consolidate_memories=lambda: calls.append(("consolidate", None)),
|
|
)
|
|
messages = [{"role": "user", "content": "Format this file."}]
|
|
|
|
context = lesson.update_context({}, messages)
|
|
system = lesson.assemble_system_prompt(context)
|
|
lesson.remember_after_turn(messages)
|
|
|
|
self.assertIn("Memory catalog", system)
|
|
self.assertIn("Relevant memory records", system)
|
|
self.assertEqual(
|
|
[name for name, _payload in calls],
|
|
["recall", "extract", "consolidate"],
|
|
)
|
|
|
|
def test_mcp_lesson_builds_on_the_base_kernel(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(
|
|
Path(tmp), ROOT / "s14_mcp_plugin" / "code.py"
|
|
)
|
|
tools_before, handlers_before = lesson.assemble_tool_pool()
|
|
self.assertEqual(
|
|
{tool["name"] for tool in tools_before},
|
|
{"bash", "read_file", "write_file", "edit_file", "glob",
|
|
"connect_mcp"},
|
|
)
|
|
self.assertNotIn("mcp__docs__search", handlers_before)
|
|
|
|
self.assertIn(
|
|
"Connected to MCP server 'docs'", lesson.connect_mcp("docs")
|
|
)
|
|
tools_after, handlers_after = lesson.assemble_tool_pool()
|
|
self.assertIn(
|
|
"mcp__docs__search",
|
|
{tool["name"] for tool in tools_after},
|
|
)
|
|
self.assertEqual(
|
|
handlers_after["mcp__docs__search"](query="hooks"),
|
|
"[docs] Found 3 results for 'hooks'",
|
|
)
|
|
|
|
def test_background_dispatch_is_bash_only_and_reports_failures(self):
|
|
for lesson_path in BACKGROUND_LESSONS:
|
|
with self.subTest(lesson=lesson_path.parent.name):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), lesson_path)
|
|
self.assertFalse(
|
|
lesson.should_run_background(
|
|
"write_file", {"run_in_background": True}
|
|
)
|
|
)
|
|
block = types.SimpleNamespace(
|
|
id="tool_fail",
|
|
name="bash",
|
|
input={"command": "exit 7", "run_in_background": True},
|
|
)
|
|
if lesson_path.parent.name in {
|
|
"s14_mcp_plugin", "s15_integrated_harness"
|
|
}:
|
|
bg_id = lesson.start_background_task(block, {})
|
|
else:
|
|
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"
|
|
)
|
|
notification = lesson.collect_background_results()[0]
|
|
self.assertIn("<status>failed</status>", notification)
|
|
self.assertIn("status 7", notification)
|
|
|
|
def test_shell_completion_terminates_children_in_the_same_process_group(self):
|
|
for lesson_path in BACKGROUND_LESSONS:
|
|
with self.subTest(lesson=lesson_path.parent.name):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), lesson_path)
|
|
marker = Path(tmp) / "late-write.txt"
|
|
command = (
|
|
"nohup sh -c "
|
|
+ shlex.quote(f"sleep 0.3; printf late > {marker}")
|
|
+ " >/dev/null 2>&1 &"
|
|
)
|
|
|
|
_, exit_code = lesson._run_bash_process(command)
|
|
time.sleep(0.5)
|
|
|
|
self.assertEqual(exit_code, 0)
|
|
self.assertFalse(marker.exists())
|
|
|
|
def test_sigterm_stops_active_shell_process_groups(self):
|
|
for lesson_path in BACKGROUND_LESSONS:
|
|
with self.subTest(lesson=lesson_path.parent.name):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
started = root / "started.txt"
|
|
late = root / "late.txt"
|
|
command = (
|
|
f"printf started > {shlex.quote(str(started))}; "
|
|
f"sleep 0.8; printf late > {shlex.quote(str(late))}"
|
|
)
|
|
script = (
|
|
"import importlib.util, os, sys, time, types\n"
|
|
"fake_anthropic = types.ModuleType('anthropic')\n"
|
|
"fake_anthropic.Anthropic = lambda *a, **k: "
|
|
"types.SimpleNamespace(messages=types.SimpleNamespace(create=None))\n"
|
|
"fake_dotenv = types.ModuleType('dotenv')\n"
|
|
"fake_dotenv.load_dotenv = lambda **k: None\n"
|
|
"fake_yaml = types.ModuleType('yaml')\n"
|
|
"fake_yaml.safe_load = lambda value: {}\n"
|
|
"fake_yaml.YAMLError = ValueError\n"
|
|
"sys.modules.update({'anthropic': fake_anthropic, "
|
|
"'dotenv': fake_dotenv, 'yaml': fake_yaml})\n"
|
|
f"os.environ['MODEL_ID'] = 'test-model'\n"
|
|
f"os.environ['ANTHROPIC_API_KEY'] = 'test-key'\n"
|
|
f"spec = importlib.util.spec_from_file_location('lesson', {str(lesson_path)!r})\n"
|
|
"lesson = importlib.util.module_from_spec(spec)\n"
|
|
"spec.loader.exec_module(lesson)\n"
|
|
f"lesson.run_bash({command!r}, run_in_background=True)\n"
|
|
"time.sleep(10)\n"
|
|
)
|
|
process = subprocess.Popen(
|
|
[sys.executable, "-c", script],
|
|
cwd=root,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
try:
|
|
self.assertTrue(wait_until(started.exists))
|
|
process.terminate()
|
|
process.wait(timeout=2)
|
|
time.sleep(1)
|
|
self.assertFalse(late.exists())
|
|
finally:
|
|
if process.poll() is None:
|
|
process.kill()
|
|
process.wait(timeout=2)
|
|
|
|
def test_durable_one_shot_is_acknowledged_after_model_acceptance(self):
|
|
for lesson_path in CRON_LESSONS:
|
|
with self.subTest(lesson=lesson_path.parent.name):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), lesson_path)
|
|
job = lesson.CronJob(
|
|
id="cron_test",
|
|
cron="* * * * *",
|
|
prompt="resume the report",
|
|
recurring=False,
|
|
durable=True,
|
|
pending_delivery=True,
|
|
)
|
|
lesson.scheduled_jobs[job.id] = job
|
|
lesson.cron_queue.append(job)
|
|
lesson.save_durable_jobs()
|
|
|
|
self.assertIn(job.id, lesson.scheduled_jobs)
|
|
persisted = lesson.DURABLE_PATH.read_text()
|
|
self.assertIn('"pending_delivery": true', persisted)
|
|
lesson.client.messages.create = lambda **_: types.SimpleNamespace(
|
|
content=[], stop_reason="end_turn"
|
|
)
|
|
messages = []
|
|
if lesson_path.parent.name == "s15_integrated_harness":
|
|
lesson.agent_loop(messages, {}, "scheduled delivery")
|
|
else:
|
|
lesson.agent_loop(messages, {})
|
|
|
|
self.assertTrue(any(
|
|
message.get("content") == "[Scheduled] resume the report"
|
|
for message in messages
|
|
))
|
|
self.assertNotIn(job.id, lesson.scheduled_jobs)
|
|
self.assertNotIn("cron_test", lesson.DURABLE_PATH.read_text())
|
|
|
|
def test_failed_model_call_restores_unacknowledged_cron_delivery(self):
|
|
for lesson_path in CRON_LESSONS:
|
|
with self.subTest(lesson=lesson_path.parent.name):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), lesson_path)
|
|
job = lesson.CronJob(
|
|
id="cron_retry",
|
|
cron="* * * * *",
|
|
prompt="retry me",
|
|
recurring=False,
|
|
durable=True,
|
|
pending_delivery=True,
|
|
)
|
|
lesson.scheduled_jobs[job.id] = job
|
|
lesson.cron_queue.append(job)
|
|
lesson.save_durable_jobs()
|
|
lesson.client.messages.create = (
|
|
lambda **_: (_ for _ in ()).throw(RuntimeError("offline"))
|
|
)
|
|
|
|
messages = []
|
|
if lesson_path.parent.name == "s15_integrated_harness":
|
|
lesson.agent_loop(messages, {}, "scheduled retry")
|
|
else:
|
|
lesson.agent_loop(messages, {})
|
|
|
|
self.assertIn(job.id, lesson.scheduled_jobs)
|
|
self.assertEqual(
|
|
[queued.id for queued in lesson.cron_queue], [job.id]
|
|
)
|
|
self.assertIn(job.id, lesson.DURABLE_PATH.read_text())
|
|
|
|
def test_failed_cron_persistence_retries_before_queueing(self):
|
|
for lesson_path in CRON_LESSONS:
|
|
with self.subTest(lesson=lesson_path.parent.name):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), lesson_path)
|
|
job = lesson.CronJob(
|
|
id="cron_persist_retry",
|
|
cron="* * * * *",
|
|
prompt="persist before delivery",
|
|
recurring=False,
|
|
durable=True,
|
|
)
|
|
lesson.scheduled_jobs[job.id] = job
|
|
original_save = lesson.save_durable_jobs
|
|
attempts = 0
|
|
|
|
def flaky_save():
|
|
nonlocal attempts
|
|
attempts += 1
|
|
if attempts == 1:
|
|
raise OSError("disk unavailable")
|
|
original_save()
|
|
|
|
lesson.save_durable_jobs = flaky_save
|
|
with self.assertRaisesRegex(OSError, "disk unavailable"):
|
|
with lesson.cron_lock:
|
|
lesson._enqueue_due_job(job)
|
|
|
|
self.assertFalse(job.pending_delivery)
|
|
self.assertEqual(lesson.cron_queue, [])
|
|
|
|
with lesson.cron_lock:
|
|
lesson._enqueue_due_job(job)
|
|
|
|
self.assertTrue(job.pending_delivery)
|
|
self.assertEqual([queued.id for queued in lesson.cron_queue], [job.id])
|
|
self.assertIn(
|
|
'"pending_delivery": true',
|
|
lesson.DURABLE_PATH.read_text(),
|
|
)
|
|
|
|
def test_cancelled_cron_is_removed_from_pending_queue(self):
|
|
for lesson_path in CRON_LESSONS:
|
|
with self.subTest(lesson=lesson_path.parent.name):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), lesson_path)
|
|
job = lesson.CronJob(
|
|
id="cron_cancel",
|
|
cron="* * * * *",
|
|
prompt="do not run",
|
|
recurring=True,
|
|
durable=True,
|
|
)
|
|
lesson.scheduled_jobs[job.id] = job
|
|
lesson.cron_queue.append(job)
|
|
|
|
self.assertIn("Cancelled", lesson.cancel_job(job.id))
|
|
self.assertEqual(lesson.consume_cron_queue(), [])
|
|
|
|
def test_mcp_permission_uses_host_policy(self):
|
|
for lesson_path in MCP_LESSONS:
|
|
with self.subTest(lesson=lesson_path.parent.name):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), lesson_path)
|
|
lesson.connect_mcp("deploy")
|
|
lesson.assemble_tool_pool()
|
|
status = types.SimpleNamespace(
|
|
name="mcp__deploy__status", input={"service": "web"}
|
|
)
|
|
trigger = types.SimpleNamespace(
|
|
name="mcp__deploy__trigger", input={"service": "web"}
|
|
)
|
|
|
|
self.assertIsNone(lesson.permission_hook(status))
|
|
with patch("builtins.input", return_value="no"):
|
|
self.assertEqual(
|
|
lesson.permission_hook(trigger),
|
|
"Permission denied by user",
|
|
)
|
|
|
|
spoofed = types.SimpleNamespace(
|
|
name="mcp__third_party__erase",
|
|
input={"description": "Erase records. (readOnly)"},
|
|
)
|
|
with patch("builtins.input", return_value="no"):
|
|
self.assertEqual(
|
|
lesson.permission_hook(spoofed),
|
|
"Permission denied by user",
|
|
)
|
|
|
|
def test_integrated_permission_requires_approval_for_every_shell_command(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
lesson = load_lesson(
|
|
root, ROOT / "s15_integrated_harness" / "code.py"
|
|
)
|
|
outside = root.parent / f"outside-{time.time_ns()}.txt"
|
|
block = types.SimpleNamespace(
|
|
name="bash",
|
|
input={"command": f"printf overwritten > {outside}"},
|
|
)
|
|
try:
|
|
with patch("builtins.input", return_value="no"):
|
|
self.assertEqual(
|
|
lesson.permission_hook(block),
|
|
"Permission denied by user",
|
|
)
|
|
self.assertFalse(outside.exists())
|
|
finally:
|
|
outside.unlink(missing_ok=True)
|
|
|
|
def test_message_bus_rejects_unregistered_or_unsafe_recipients(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
lesson = load_lesson(root)
|
|
lesson.active_teammates["alice"] = "idle"
|
|
|
|
with self.assertRaises(ValueError):
|
|
lesson.BUS.send("alice", "../escape", "bad")
|
|
self.assertFalse((root / "escape.jsonl").exists())
|
|
|
|
result = lesson._teammate_send_message(
|
|
"alice", "ghost", "Are you there?"
|
|
)
|
|
self.assertIn("not active", result)
|
|
self.assertFalse((lesson.MAILBOX_DIR / "ghost.jsonl").exists())
|
|
|
|
def test_reserved_teammate_names_do_not_shadow_runtime_identities(self):
|
|
for lesson_path in RUNTIME_LESSONS:
|
|
with self.subTest(lesson=lesson_path.parent.name):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), lesson_path)
|
|
|
|
for name in ("lead", "agent", "Lead", "Agent"):
|
|
rejected = lesson.spawn_teammate_thread(
|
|
name, "backend", "Inspect auth."
|
|
)
|
|
self.assertIn("reserved", rejected.lower())
|
|
self.assertNotIn(name, lesson.active_teammates)
|
|
|
|
lesson.BUS.send("alice", "lead", "still routable")
|
|
self.assertEqual(
|
|
lesson.BUS.read_inbox("lead")[0]["content"],
|
|
"still routable",
|
|
)
|
|
|
|
lesson.active_teammates["Alice"] = "idle"
|
|
duplicate = lesson.spawn_teammate_thread(
|
|
"alice", "backend", "Inspect auth."
|
|
)
|
|
self.assertIn("already exists", duplicate)
|
|
|
|
def test_public_task_tools_return_errors_for_bad_ids(self):
|
|
for lesson_path in RUNTIME_LESSONS:
|
|
with self.subTest(lesson=lesson_path.parent.name):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), lesson_path)
|
|
for task_id in ("../escape", "task_missing"):
|
|
for tool_name in (
|
|
"run_get_task",
|
|
"run_claim_task",
|
|
"run_complete_task",
|
|
):
|
|
with self.subTest(tool=tool_name, task_id=task_id):
|
|
result = getattr(lesson, tool_name)(task_id)
|
|
self.assertIn("Error:", result)
|
|
|
|
def test_plan_gate_blocks_mutating_tools_until_approval(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp))
|
|
cases = {
|
|
"write_file": {"path": "config.py", "content": "VALUE = 1"},
|
|
"edit_file": {
|
|
"path": "config.py", "old_text": "0", "new_text": "1"
|
|
},
|
|
}
|
|
for tool_name, tool_input in cases.items():
|
|
with self.subTest(tool=tool_name):
|
|
calls = []
|
|
block = types.SimpleNamespace(
|
|
name=tool_name, input=tool_input
|
|
)
|
|
handlers = {
|
|
tool_name: lambda **kwargs: calls.append(kwargs) or "done"
|
|
}
|
|
|
|
lesson.plan_gates["alice"] = "pending"
|
|
blocked = lesson._run_teammate_tool(
|
|
"alice", block, handlers
|
|
)
|
|
self.assertIn("Blocked", blocked)
|
|
self.assertEqual(calls, [])
|
|
|
|
lesson.plan_gates["alice"] = "approved"
|
|
allowed = lesson._run_teammate_tool(
|
|
"alice", block, handlers
|
|
)
|
|
self.assertEqual(allowed, "done")
|
|
self.assertEqual(len(calls), 1)
|
|
|
|
def test_teammate_tool_errors_become_tool_results(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp))
|
|
lesson.plan_gates["alice"] = "not_required"
|
|
block = types.SimpleNamespace(
|
|
name="write_file", input={"path": "config.py"}
|
|
)
|
|
|
|
result = lesson._run_teammate_tool(
|
|
"alice", block,
|
|
{"write_file": lambda path, content: "wrote"},
|
|
)
|
|
|
|
self.assertIn("TypeError", result)
|
|
|
|
def test_teammate_keeps_complete_tool_history(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp))
|
|
lesson.IDLE_SCAN_INTERVAL = 5.0
|
|
calls = 0
|
|
|
|
def respond(**kwargs):
|
|
nonlocal calls
|
|
calls += 1
|
|
self.assertEqual(
|
|
kwargs["messages"][0]["content"], "Inspect the project."
|
|
)
|
|
if calls <= 11:
|
|
return types.SimpleNamespace(
|
|
stop_reason="tool_use",
|
|
content=[types.SimpleNamespace(
|
|
type="tool_use", name="list_tasks",
|
|
id=f"list-{calls}", input={},
|
|
)],
|
|
)
|
|
return types.SimpleNamespace(
|
|
stop_reason="end_turn",
|
|
content=[types.SimpleNamespace(
|
|
type="text", text="Inspection complete."
|
|
)],
|
|
)
|
|
|
|
lesson.client.messages.create = respond
|
|
lesson.spawn_teammate_thread(
|
|
"alice", "reviewer", "Inspect the project."
|
|
)
|
|
self.assertTrue(wait_until(
|
|
lambda: lesson.BUS.peek("lead"), timeout=3.0
|
|
))
|
|
self.assertEqual(calls, 12)
|
|
lesson.run_request_shutdown("alice")
|
|
self.assertTrue(wait_until(
|
|
lambda: "alice" not in lesson.active_teammates
|
|
))
|
|
|
|
def test_s15_teammate_dispatch_runs_permission_and_post_hooks(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(
|
|
Path(tmp), ROOT / "s15_integrated_harness" / "code.py"
|
|
)
|
|
block = types.SimpleNamespace(
|
|
name="write_file",
|
|
input={"path": "config.py", "content": "VALUE = 1"},
|
|
)
|
|
calls = []
|
|
handlers = {
|
|
"write_file": lambda **kwargs: calls.append(
|
|
("handler", kwargs)
|
|
) or "wrote"
|
|
}
|
|
lesson.plan_gates["alice"] = "approved"
|
|
lesson.HOOKS["PreToolUse"] = [
|
|
lambda seen: calls.append(("pre", seen.name)) or "denied"
|
|
]
|
|
lesson.HOOKS["PostToolUse"] = [
|
|
lambda seen, output: calls.append(
|
|
("post", seen.name, output)
|
|
)
|
|
]
|
|
|
|
denied = lesson._run_teammate_tool("alice", block, handlers)
|
|
self.assertEqual(denied, "denied")
|
|
self.assertEqual(calls, [("pre", "write_file")])
|
|
|
|
calls.clear()
|
|
lesson.HOOKS["PreToolUse"] = [
|
|
lambda seen: calls.append(("pre", seen.name))
|
|
]
|
|
allowed = lesson._run_teammate_tool("alice", block, handlers)
|
|
|
|
self.assertEqual(allowed, "wrote")
|
|
self.assertEqual(
|
|
calls,
|
|
[
|
|
("pre", "write_file"),
|
|
("handler", block.input),
|
|
("post", "write_file", "wrote"),
|
|
],
|
|
)
|
|
|
|
def test_s15_teammate_reads_shutdown_between_tool_rounds(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(
|
|
Path(tmp), ROOT / "s15_integrated_harness" / "code.py"
|
|
)
|
|
entered = threading.Event()
|
|
release = threading.Event()
|
|
calls = []
|
|
|
|
def create(**_kwargs):
|
|
calls.append("llm")
|
|
entered.set()
|
|
release.wait(timeout=2)
|
|
block = types.SimpleNamespace(
|
|
type="tool_use", id="tool_1", name="list_tasks", input={}
|
|
)
|
|
return types.SimpleNamespace(
|
|
stop_reason="tool_use", content=[block]
|
|
)
|
|
|
|
lesson.client.messages.create = create
|
|
lesson.spawn_teammate_thread("alice", "reviewer", "Inspect tasks")
|
|
self.assertTrue(entered.wait(timeout=2))
|
|
lesson.run_request_shutdown("alice")
|
|
release.set()
|
|
|
|
self.assertTrue(
|
|
wait_until(lambda: "alice" not in lesson.active_teammates)
|
|
)
|
|
self.assertEqual(calls, ["llm"])
|
|
|
|
def test_normalized_mcp_tool_name_collisions_are_rejected(self):
|
|
for lesson_path in MCP_LESSONS:
|
|
with self.subTest(lesson=lesson_path.parent.name):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), lesson_path)
|
|
first = lesson.MCPClient("docs.one")
|
|
first.register(
|
|
[{"name": "get.version", "inputSchema": {}}],
|
|
{"get.version": lambda: "one"},
|
|
)
|
|
second = lesson.MCPClient("docs_one")
|
|
second.register(
|
|
[{"name": "get_version", "inputSchema": {}}],
|
|
{"get_version": lambda: "two"},
|
|
)
|
|
lesson.mcp_clients.clear()
|
|
lesson.mcp_clients.update({
|
|
"docs.one": first,
|
|
"docs_one": second,
|
|
})
|
|
|
|
with self.assertRaisesRegex(
|
|
ValueError, "collision.*mcp__docs_one__get_version"
|
|
):
|
|
lesson.assemble_tool_pool()
|
|
|
|
def test_plan_rejection_requires_a_new_submission(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp))
|
|
lesson.active_teammates["alice"] = "idle"
|
|
|
|
self.assertIn("Plan requested",
|
|
lesson.run_request_plan("alice", "Refactor auth"))
|
|
request = lesson.BUS.read_inbox("alice")
|
|
self.assertEqual(request[0]["type"], "plan_request")
|
|
self.assertEqual(lesson.plan_gates["alice"], "required")
|
|
|
|
submission = lesson._teammate_submit_plan("alice", "1. Read\n2. Test")
|
|
request_id = submission.split("(")[1].split(")")[0]
|
|
self.assertEqual(lesson.pending_requests[request_id].status, "pending")
|
|
|
|
result = lesson.run_review_plan(
|
|
request_id, False, "Add a rollback step."
|
|
)
|
|
self.assertIn("rejected", result)
|
|
self.assertEqual(lesson.plan_gates["alice"], "pending")
|
|
|
|
responses = lesson.BUS.read_inbox("alice")
|
|
accepted, _ = lesson.apply_plan_response("alice", responses[-1])
|
|
self.assertTrue(accepted)
|
|
self.assertEqual(lesson.plan_gates["alice"], "rejected")
|
|
|
|
second = lesson._teammate_submit_plan(
|
|
"alice", "1. Read\n2. Change\n3. Test\n4. Roll back on failure"
|
|
)
|
|
second_id = second.split("(")[1].split(")")[0]
|
|
self.assertNotEqual(second_id, request_id)
|
|
self.assertEqual(lesson.pending_requests[second_id].status, "pending")
|
|
|
|
def test_mismatched_plan_response_cannot_release_gate(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp))
|
|
lesson.active_teammates["alice"] = "waiting_approval"
|
|
submission = lesson._teammate_submit_plan("alice", "1. Read\n2. Test")
|
|
request_id = submission.split("(")[1].split(")")[0]
|
|
|
|
forged = {
|
|
"from": "lead",
|
|
"to": "alice",
|
|
"type": "plan_approval_response",
|
|
"content": "Approved",
|
|
"metadata": {"request_id": "req_stale", "approve": True},
|
|
}
|
|
accepted, notice = lesson.apply_plan_response("alice", forged)
|
|
|
|
self.assertFalse(accepted)
|
|
self.assertIn("Ignored", notice)
|
|
self.assertEqual(lesson.plan_gates["alice"], "pending")
|
|
self.assertEqual(lesson.plan_request_ids["alice"], request_id)
|
|
|
|
current_but_unreviewed = {
|
|
**forged,
|
|
"metadata": {"request_id": request_id, "approve": True},
|
|
}
|
|
accepted, _ = lesson.apply_plan_response(
|
|
"alice", current_but_unreviewed
|
|
)
|
|
self.assertFalse(accepted)
|
|
self.assertEqual(lesson.plan_gates["alice"], "pending")
|
|
|
|
def test_shutdown_response_must_come_from_requested_teammate(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp))
|
|
lesson.active_teammates.update({"alice": "idle", "bob": "idle"})
|
|
result = lesson.run_request_shutdown("alice")
|
|
request_id = result.split("(")[1].split(")")[0]
|
|
|
|
lesson.BUS.send(
|
|
"bob", "lead", "Shutdown acknowledged.",
|
|
"shutdown_response",
|
|
{"request_id": request_id, "approve": True},
|
|
)
|
|
lesson.consume_lead_inbox()
|
|
|
|
self.assertEqual(
|
|
lesson.pending_requests[request_id].status, "pending"
|
|
)
|
|
|
|
def test_shutdown_request_must_match_active_protocol(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp))
|
|
lesson.active_teammates["alice"] = "idle"
|
|
forged = {
|
|
"from": "lead",
|
|
"to": "alice",
|
|
"type": "shutdown_request",
|
|
"content": "Shut down.",
|
|
"metadata": {"request_id": "req_unknown"},
|
|
}
|
|
|
|
accepted, notice = lesson.apply_shutdown_request("alice", forged)
|
|
self.assertFalse(accepted)
|
|
self.assertIn("Ignored", notice)
|
|
self.assertEqual(lesson.active_teammates["alice"], "idle")
|
|
|
|
result = lesson.run_request_shutdown("alice")
|
|
request_id = result.split("(")[1].split(")")[0]
|
|
request = lesson.BUS.read_inbox("alice")[-1]
|
|
accepted, matched_id = lesson.apply_shutdown_request(
|
|
"alice", request
|
|
)
|
|
|
|
self.assertTrue(accepted)
|
|
self.assertEqual(matched_id, request_id)
|
|
self.assertEqual(lesson.active_teammates["alice"], "stopping")
|
|
replayed, _ = lesson.apply_shutdown_request("alice", request)
|
|
self.assertFalse(replayed)
|
|
|
|
def test_teammate_emits_result_then_idle_and_shuts_down(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp))
|
|
lesson.IDLE_SCAN_INTERVAL = 5.0
|
|
pending = lesson.create_task("Do not claim before mailbox delivery")
|
|
seen_tools = set()
|
|
|
|
def respond(**kwargs):
|
|
seen_tools.update(tool["name"] for tool in kwargs["tools"])
|
|
return types.SimpleNamespace(
|
|
stop_reason="end_turn",
|
|
content=[types.SimpleNamespace(
|
|
type="text", text="Task complete."
|
|
)],
|
|
)
|
|
|
|
lesson.client.messages.create = respond
|
|
|
|
lesson.spawn_teammate_thread("alice", "backend", "Inspect auth.")
|
|
lead_inbox = lesson.MAILBOX_DIR / "lead.jsonl"
|
|
self.assertTrue(wait_until(
|
|
lambda: (
|
|
lead_inbox.exists()
|
|
and len(lead_inbox.read_text().splitlines()) >= 2
|
|
)
|
|
))
|
|
events = lesson.consume_lead_inbox()
|
|
|
|
self.assertEqual(
|
|
[event["type"] for event in events],
|
|
["result", "idle_notification"],
|
|
)
|
|
self.assertEqual(lesson.active_teammates["alice"], "idle")
|
|
self.assertTrue(
|
|
{"list_tasks", "claim_task", "complete_task"}
|
|
.issubset(seen_tools)
|
|
)
|
|
self.assertTrue(
|
|
{"create_worktree", "remove_worktree", "keep_worktree"}
|
|
.isdisjoint(seen_tools)
|
|
)
|
|
|
|
lesson.run_request_shutdown("alice")
|
|
self.assertTrue(
|
|
wait_until(lambda: "alice" not in lesson.active_teammates)
|
|
)
|
|
shutdown_events = lesson.consume_lead_inbox()
|
|
self.assertEqual(shutdown_events[-1]["type"], "shutdown_response")
|
|
request_id = shutdown_events[-1]["metadata"]["request_id"]
|
|
self.assertEqual(
|
|
lesson.pending_requests[request_id].status, "approved"
|
|
)
|
|
self.assertEqual(lesson.load_task(pending.id).status, "pending")
|
|
|
|
def test_downstream_teammates_continue_past_ten_tool_rounds(self):
|
|
for lesson_path in DOWNSTREAM_LESSONS:
|
|
with self.subTest(lesson=lesson_path.parent.name):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), lesson_path)
|
|
lesson.IDLE_SCAN_INTERVAL = 5.0
|
|
calls = 0
|
|
|
|
def respond(**kwargs):
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls <= 11:
|
|
return types.SimpleNamespace(
|
|
stop_reason="tool_use",
|
|
content=[types.SimpleNamespace(
|
|
type="tool_use",
|
|
name="list_tasks",
|
|
id=f"list-{calls}",
|
|
input={},
|
|
)],
|
|
)
|
|
return types.SimpleNamespace(
|
|
stop_reason="end_turn",
|
|
content=[types.SimpleNamespace(
|
|
type="text", text="Long task complete."
|
|
)],
|
|
)
|
|
|
|
lesson.client.messages.create = respond
|
|
lesson.spawn_teammate_thread(
|
|
"alice", "backend", "Use more than ten tool rounds."
|
|
)
|
|
lead_inbox = lesson.MAILBOX_DIR / "lead.jsonl"
|
|
self.assertTrue(wait_until(
|
|
lambda: (
|
|
lead_inbox.exists()
|
|
and len(lead_inbox.read_text().splitlines()) >= 2
|
|
),
|
|
timeout=3.0,
|
|
))
|
|
events = lesson.consume_lead_inbox()
|
|
|
|
self.assertEqual(calls, 12)
|
|
self.assertEqual(
|
|
[event["type"] for event in events],
|
|
["result", "idle_notification"],
|
|
)
|
|
self.assertEqual(
|
|
lesson.active_teammates.get("alice"), "idle"
|
|
)
|
|
lesson.run_request_shutdown("alice")
|
|
self.assertTrue(wait_until(
|
|
lambda: "alice" not in lesson.active_teammates
|
|
))
|
|
|
|
def test_teammate_exception_releases_runtime_and_task_ownership(self):
|
|
for lesson_path in RUNTIME_LESSONS:
|
|
with self.subTest(lesson=lesson_path.parent.name):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), lesson_path)
|
|
task = lesson.create_task("Implement auth")
|
|
calls = 0
|
|
|
|
def respond(**kwargs):
|
|
nonlocal calls
|
|
calls += 1
|
|
tool_name = "claim_task" if calls == 1 else "list_tasks"
|
|
tool_input = {"task_id": task.id} if calls == 1 else {}
|
|
return types.SimpleNamespace(
|
|
stop_reason="tool_use",
|
|
content=[types.SimpleNamespace(
|
|
type="tool_use", name=tool_name,
|
|
id=f"tool-{calls}", input=tool_input,
|
|
)],
|
|
)
|
|
|
|
original_dispatch = lesson._run_teammate_tool
|
|
|
|
def crash_after_claim(name, block, handlers):
|
|
if block.name == "list_tasks":
|
|
raise RuntimeError("simulated dispatch failure")
|
|
return original_dispatch(name, block, handlers)
|
|
|
|
lesson.client.messages.create = respond
|
|
lesson._run_teammate_tool = crash_after_claim
|
|
lesson.spawn_teammate_thread(
|
|
"alice", "backend", "Claim and begin work."
|
|
)
|
|
|
|
self.assertTrue(wait_until(
|
|
lambda: "alice" not in lesson.active_teammates
|
|
))
|
|
self.assertNotIn("alice", lesson.teammate_assignments)
|
|
recovered = lesson.load_task(task.id)
|
|
self.assertEqual(recovered.status, "pending")
|
|
self.assertIsNone(recovered.owner)
|
|
events = lesson.consume_lead_inbox()
|
|
self.assertEqual([event["type"] for event in events], ["error"])
|
|
self.assertIn("simulated dispatch failure", events[0]["content"])
|
|
|
|
def test_s15_completed_background_task_wakes_the_agent_once(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(
|
|
Path(tmp), ROOT / "s15_integrated_harness" / "code.py"
|
|
)
|
|
seen_messages = []
|
|
|
|
def respond(messages, context, tools, state, max_tokens):
|
|
seen_messages.append(list(messages))
|
|
return types.SimpleNamespace(
|
|
stop_reason="end_turn",
|
|
content=[types.SimpleNamespace(
|
|
type="text", text="Background result handled."
|
|
)],
|
|
)
|
|
|
|
lesson.call_llm = respond
|
|
lesson.background_tasks["bg_0001"] = {
|
|
"tool_use_id": "tool-1",
|
|
"command": "pytest",
|
|
"status": "completed",
|
|
}
|
|
lesson.background_results["bg_0001"] = "all tests passed"
|
|
history = []
|
|
context = {}
|
|
session_state = {"active_user_request": "Run tests"}
|
|
threading.Thread(
|
|
target=lesson.async_event_loop,
|
|
args=(history, context, session_state),
|
|
daemon=True,
|
|
).start()
|
|
|
|
self.assertTrue(wait_until(lambda: bool(seen_messages), timeout=3.0))
|
|
delivered = str(seen_messages[0])
|
|
self.assertIn("<task_notification>", delivered)
|
|
self.assertIn("all tests passed", delivered)
|
|
self.assertFalse(lesson.has_pending_background())
|
|
calls_after_delivery = len(seen_messages)
|
|
time.sleep(1.2)
|
|
self.assertEqual(len(seen_messages), calls_after_delivery)
|
|
|
|
def test_teammate_survives_stale_worktree_assignment(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
init_git_repo(root)
|
|
lesson = load_lesson(root)
|
|
lesson.IDLE_SCAN_INTERVAL = 5.0
|
|
task = lesson.create_task("Implement auth")
|
|
lesson.create_worktree("auth", task.id)
|
|
worktree = lesson.WORKTREES_DIR / "auth"
|
|
calls = 0
|
|
bash_result = []
|
|
|
|
def respond(**kwargs):
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls == 1:
|
|
return types.SimpleNamespace(
|
|
stop_reason="tool_use",
|
|
content=[types.SimpleNamespace(
|
|
type="tool_use", name="claim_task", id="claim-1",
|
|
input={"task_id": task.id},
|
|
)],
|
|
)
|
|
if calls == 2:
|
|
subprocess.run(
|
|
["git", "worktree", "remove", "--force",
|
|
str(worktree)], cwd=root, check=True,
|
|
)
|
|
return types.SimpleNamespace(
|
|
stop_reason="tool_use",
|
|
content=[types.SimpleNamespace(
|
|
type="tool_use", name="bash", id="bash-1",
|
|
input={"command": "pwd"},
|
|
)],
|
|
)
|
|
bash_result.append(
|
|
kwargs["messages"][-1]["content"][0]["content"]
|
|
)
|
|
return types.SimpleNamespace(
|
|
stop_reason="end_turn",
|
|
content=[types.SimpleNamespace(
|
|
type="text", text="Handled stale assignment."
|
|
)],
|
|
)
|
|
|
|
lesson.client.messages.create = respond
|
|
lesson.spawn_teammate_thread("alice", "backend", "Claim the task.")
|
|
|
|
self.assertTrue(wait_until(lambda: bool(bash_result)))
|
|
self.assertIn("Invalid task assignment", bash_result[0])
|
|
self.assertIn("alice", lesson.active_teammates)
|
|
lesson.run_request_shutdown("alice")
|
|
self.assertTrue(
|
|
wait_until(lambda: "alice" not in lesson.active_teammates)
|
|
)
|
|
|
|
def test_idle_claim_is_atomic_across_teammates(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp))
|
|
task = lesson.create_task("Refactor auth")
|
|
|
|
barrier = threading.Barrier(3)
|
|
claimed = {}
|
|
|
|
def claim(name):
|
|
barrier.wait()
|
|
claimed[name] = lesson.claim_next_task(name)
|
|
|
|
threads = [
|
|
threading.Thread(target=claim, args=(name,))
|
|
for name in ("alice", "bob")
|
|
]
|
|
for thread in threads:
|
|
thread.start()
|
|
barrier.wait()
|
|
for thread in threads:
|
|
thread.join(timeout=2)
|
|
|
|
self.assertTrue(all(not thread.is_alive() for thread in threads))
|
|
self.assertEqual(
|
|
len([result for result in claimed.values() if result is not None]),
|
|
1,
|
|
)
|
|
winner = next(result.owner for result in claimed.values()
|
|
if result is not None)
|
|
self.assertEqual(lesson.load_task(task.id).owner, winner)
|
|
|
|
def test_assignment_enforces_one_task_and_owner_only_completion(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp))
|
|
first = lesson.create_task("Refactor auth")
|
|
second = lesson.create_task("Refactor login")
|
|
|
|
self.assertIn("Claimed", lesson.claim_task(first.id, owner="alice"))
|
|
denied = lesson.claim_task(second.id, owner="alice")
|
|
self.assertIn("must finish", denied)
|
|
self.assertEqual(lesson.load_task(second.id).status, "pending")
|
|
|
|
denied = lesson.complete_task(first.id, owner="bob")
|
|
self.assertIn("not bob", denied)
|
|
self.assertEqual(lesson.load_task(first.id).status, "in_progress")
|
|
|
|
self.assertIn(
|
|
"Completed", lesson.complete_task(first.id, owner="alice")
|
|
)
|
|
self.assertIn("alice", lesson.teammate_assignments)
|
|
denied = lesson.claim_task(second.id, owner="alice")
|
|
self.assertIn("must finish", denied)
|
|
self.assertTrue(lesson.release_completed_assignment("alice"))
|
|
self.assertIn("Claimed", lesson.claim_task(second.id, owner="alice"))
|
|
|
|
def test_completed_assignment_keeps_lead_in_worktree_until_turn_boundary(self):
|
|
for lesson_path in RUNTIME_LESSONS:
|
|
with self.subTest(lesson=lesson_path.parent.name):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
init_git_repo(root)
|
|
lesson = load_lesson(root, lesson_path)
|
|
first = lesson.create_task("Implement auth")
|
|
second = lesson.create_task("Update docs")
|
|
lesson.create_worktree("auth", first.id)
|
|
|
|
self.assertIn(
|
|
"Claimed", lesson.claim_task(first.id, owner="agent")
|
|
)
|
|
self.assertIn(
|
|
"Completed", lesson.complete_task(first.id, owner="agent")
|
|
)
|
|
self.assertIn(
|
|
"Wrote",
|
|
lesson.run_agent_write("after-complete.txt", "done"),
|
|
)
|
|
self.assertTrue(
|
|
(lesson.WORKTREES_DIR / "auth" / "after-complete.txt").exists()
|
|
)
|
|
self.assertFalse((root / "after-complete.txt").exists())
|
|
self.assertIn(
|
|
"must finish",
|
|
lesson.claim_task(second.id, owner="agent"),
|
|
)
|
|
|
|
self.assertTrue(lesson.release_completed_assignment("agent"))
|
|
self.assertIn(
|
|
"Claimed", lesson.claim_task(second.id, owner="agent")
|
|
)
|
|
|
|
def test_in_progress_assignment_rehydrates_after_runtime_restart(self):
|
|
for lesson_path in RUNTIME_LESSONS:
|
|
with self.subTest(lesson=lesson_path.parent.name):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
init_git_repo(root)
|
|
lesson = load_lesson(root, lesson_path)
|
|
task = lesson.create_task("Implement auth")
|
|
lesson.create_worktree("auth", task.id)
|
|
lesson.claim_task(task.id, owner="alice")
|
|
|
|
lesson.teammate_assignments.clear()
|
|
recovered = lesson.assignment_cwd("alice")
|
|
|
|
self.assertEqual(
|
|
recovered.resolve(),
|
|
(lesson.WORKTREES_DIR / "auth").resolve(),
|
|
)
|
|
self.assertEqual(
|
|
lesson.teammate_assignments["alice"]["task_id"], task.id
|
|
)
|
|
|
|
def test_completion_rehydrates_cwd_lease_before_status_change(self):
|
|
for lesson_path in RUNTIME_LESSONS:
|
|
with self.subTest(lesson=lesson_path.parent.name):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
init_git_repo(root)
|
|
lesson = load_lesson(root, lesson_path)
|
|
task = lesson.create_task("Implement auth")
|
|
lesson.create_worktree("auth", task.id)
|
|
lesson.claim_task(task.id, owner="agent")
|
|
lesson.teammate_assignments.clear()
|
|
|
|
self.assertIn(
|
|
"Completed", lesson.complete_task(task.id, owner="agent")
|
|
)
|
|
self.assertIn(
|
|
"Wrote", lesson.run_agent_write("after.txt", "done")
|
|
)
|
|
self.assertTrue(
|
|
(lesson.WORKTREES_DIR / "auth" / "after.txt").exists()
|
|
)
|
|
self.assertFalse((root / "after.txt").exists())
|
|
|
|
def test_completion_replaces_a_stale_cross_runtime_cwd_lease(self):
|
|
for lesson_path in RUNTIME_LESSONS:
|
|
with self.subTest(lesson=lesson_path.parent.name):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
init_git_repo(root)
|
|
first = load_lesson(root, lesson_path)
|
|
old_task = first.create_task("Old assignment")
|
|
first.create_worktree("old", old_task.id)
|
|
first.claim_task(old_task.id, owner="agent")
|
|
first.complete_task(old_task.id, owner="agent")
|
|
|
|
second = load_lesson(root, lesson_path)
|
|
new_task = second.create_task("New assignment")
|
|
second.create_worktree("new", new_task.id)
|
|
second.claim_task(new_task.id, owner="agent")
|
|
|
|
self.assertIn(
|
|
"Completed",
|
|
first.complete_task(new_task.id, owner="agent"),
|
|
)
|
|
self.assertIn(
|
|
"Wrote", first.run_agent_write("after.txt", "done")
|
|
)
|
|
self.assertTrue(
|
|
(first.WORKTREES_DIR / "new" / "after.txt").exists()
|
|
)
|
|
self.assertFalse(
|
|
(first.WORKTREES_DIR / "old" / "after.txt").exists()
|
|
)
|
|
|
|
def test_task_claim_is_atomic_across_processes(self):
|
|
context = multiprocessing.get_context("spawn")
|
|
for lesson_path in RUNTIME_LESSONS:
|
|
with self.subTest(lesson=lesson_path.parent.name):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), lesson_path)
|
|
task = lesson.create_task("Only once")
|
|
barrier = context.Barrier(3)
|
|
results = context.Queue()
|
|
|
|
workers = [
|
|
context.Process(
|
|
target=claim_in_child,
|
|
args=(
|
|
str(lesson_path), tmp, task.id, owner,
|
|
barrier, results,
|
|
),
|
|
)
|
|
for owner in ("alice", "bob")
|
|
]
|
|
for worker in workers:
|
|
worker.start()
|
|
barrier.wait()
|
|
for worker in workers:
|
|
worker.join(5)
|
|
self.assertEqual(worker.exitcode, 0)
|
|
outcomes = [results.get(timeout=1) for _ in workers]
|
|
|
|
self.assertEqual(
|
|
sum(outcome.startswith("Claimed ") for outcome in outcomes),
|
|
1,
|
|
)
|
|
persisted = lesson.load_task(task.id)
|
|
self.assertEqual(persisted.status, "in_progress")
|
|
self.assertIn(persisted.owner, {"alice", "bob"})
|
|
|
|
def test_plan_approval_cannot_cross_assignment_boundary(self):
|
|
for lesson_path in RUNTIME_LESSONS:
|
|
with self.subTest(lesson=lesson_path.parent.name):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), lesson_path)
|
|
lesson.active_teammates["alice"] = "working"
|
|
lesson.plan_gates["alice"] = "required"
|
|
lesson.assignment_versions["alice"] = 1
|
|
lesson._teammate_submit_plan("alice", "Inspect, edit, test")
|
|
request_id = next(iter(lesson.pending_requests))
|
|
|
|
lesson.advance_assignment_version("alice")
|
|
result = lesson.run_review_plan(request_id, True)
|
|
|
|
self.assertIn("earlier assignment", result)
|
|
self.assertNotEqual(lesson.plan_gates["alice"], "approved")
|
|
|
|
def test_required_plan_is_active_before_teammate_thread_starts(self):
|
|
for lesson_path in RUNTIME_LESSONS:
|
|
with self.subTest(lesson=lesson_path.parent.name):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), lesson_path)
|
|
tool_defs = getattr(lesson, "TOOLS", None)
|
|
if tool_defs is None:
|
|
tool_defs = lesson.BUILTIN_TOOLS
|
|
spawn_schema = next(
|
|
tool["input_schema"] for tool in tool_defs
|
|
if tool["name"] == "spawn_teammate"
|
|
)
|
|
self.assertIn("require_plan", spawn_schema["properties"])
|
|
with patch.object(
|
|
lesson.threading.Thread, "start", lambda _thread: None
|
|
):
|
|
lesson.spawn_teammate_thread(
|
|
"alice", "backend", "Claim and edit.",
|
|
require_plan=True,
|
|
)
|
|
task = lesson.create_task("Edit auth")
|
|
self.assertIn("Claimed", lesson.claim_task(task.id, "alice"))
|
|
self.assertEqual(lesson.plan_gates["alice"], "required")
|
|
calls = []
|
|
block = types.SimpleNamespace(
|
|
name="write_file",
|
|
input={"path": "auth.py", "content": "changed"},
|
|
)
|
|
denied = lesson._run_teammate_tool(
|
|
"alice", block,
|
|
{"write_file": lambda **kw: calls.append(kw)},
|
|
)
|
|
self.assertIn("Blocked", denied)
|
|
self.assertEqual(calls, [])
|
|
|
|
def test_worktree_registry_parsing_does_not_use_display_truncation(self):
|
|
for lesson_path in RUNTIME_LESSONS:
|
|
with self.subTest(lesson=lesson_path.parent.name):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
lesson = load_lesson(Path(tmp), lesson_path)
|
|
entries = []
|
|
for index in range(80):
|
|
path = Path(tmp) / ".worktrees" / (f"work-{index}-" + "x" * 80)
|
|
entries.append(
|
|
f"worktree {path}\nHEAD {'0' * 40}\n"
|
|
f"branch refs/heads/wt/work-{index}\n"
|
|
)
|
|
porcelain = "\n".join(entries)
|
|
self.assertGreater(len(porcelain), 5000)
|
|
lesson._run_git = lambda args, cwd=None: (True, porcelain)
|
|
|
|
registered, error = lesson._registered_worktrees()
|
|
|
|
self.assertIsNone(error)
|
|
self.assertEqual(len(registered), 80)
|
|
|
|
def test_task_worktree_sets_assignment_cwd_and_contains_file_tools(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
init_git_repo(root)
|
|
lesson = load_lesson(root)
|
|
task = lesson.create_task("Implement auth")
|
|
|
|
created = lesson.create_worktree("auth", task.id)
|
|
self.assertIn("created", created)
|
|
worktree = lesson.WORKTREES_DIR / "auth"
|
|
self.assertEqual(lesson.load_task(task.id).worktree, "auth")
|
|
|
|
self.assertIn("Claimed", lesson.claim_task(task.id, owner="alice"))
|
|
assignment = lesson.teammate_assignments["alice"]
|
|
self.assertEqual(assignment["task_id"], task.id)
|
|
self.assertEqual(assignment["cwd"], worktree)
|
|
self.assertIn(
|
|
"Wrote", lesson.run_write(
|
|
"nested/result.txt", "done", cwd=lesson.assignment_cwd("alice")
|
|
)
|
|
)
|
|
self.assertEqual((worktree / "nested" / "result.txt").read_text(),
|
|
"done")
|
|
escaped = lesson.run_write(
|
|
"../outside.txt", "bad", cwd=lesson.assignment_cwd("alice")
|
|
)
|
|
self.assertIn("escapes workspace", escaped)
|
|
self.assertFalse((lesson.WORKTREES_DIR / "outside.txt").exists())
|
|
missing_cwd = lesson.run_bash("pwd", cwd=worktree / "missing")
|
|
self.assertIn("FileNotFoundError", missing_cwd)
|
|
|
|
def test_invalid_or_unregistered_worktree_never_becomes_claimable(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
init_git_repo(root)
|
|
lesson = load_lesson(root)
|
|
task = lesson.create_task("Implement auth")
|
|
|
|
invalid = lesson.create_worktree("../escape", task.id)
|
|
self.assertIn("Error", invalid)
|
|
self.assertIsNone(lesson.load_task(task.id).worktree)
|
|
self.assertFalse((root / "escape").exists())
|
|
|
|
missing = lesson.create_worktree("auth", "../missing")
|
|
self.assertIn("Error", missing)
|
|
self.assertFalse((lesson.WORKTREES_DIR / "auth").exists())
|
|
|
|
bound = lesson.load_task(task.id)
|
|
bound.worktree = "ghost"
|
|
lesson.save_task(bound)
|
|
denied = lesson.claim_task(task.id, owner="alice")
|
|
self.assertIn("not registered", denied)
|
|
self.assertEqual(lesson.load_task(task.id).status, "pending")
|
|
self.assertEqual(lesson.scan_unclaimed_tasks(), [])
|
|
|
|
def test_create_validates_branch_and_binds_only_after_git_add(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
init_git_repo(root)
|
|
lesson = load_lesson(root)
|
|
task = lesson.create_task("Implement auth")
|
|
subprocess.run(
|
|
["git", "branch", "wt/auth"], cwd=root, check=True
|
|
)
|
|
|
|
collision = lesson.create_worktree("auth", task.id)
|
|
self.assertIn("already exists", collision)
|
|
self.assertIsNone(lesson.load_task(task.id).worktree)
|
|
self.assertFalse((lesson.WORKTREES_DIR / "auth").exists())
|
|
|
|
original_run_git = lesson.run_git
|
|
|
|
def fail_add(args, cwd=None):
|
|
if args[:2] == ["worktree", "add"]:
|
|
return False, "simulated add failure"
|
|
return original_run_git(args, cwd=cwd)
|
|
|
|
lesson.run_git = fail_add
|
|
failed = lesson.create_worktree("login", task.id)
|
|
self.assertIn("simulated add failure", failed)
|
|
self.assertIsNone(lesson.load_task(task.id).worktree)
|
|
self.assertFalse((lesson.WORKTREES_DIR / "login").exists())
|
|
|
|
def test_failed_git_add_reports_and_preserves_partial_artifacts(self):
|
|
for lesson_path in RUNTIME_LESSONS:
|
|
with self.subTest(lesson=lesson_path.parent.name):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
init_git_repo(root)
|
|
lesson = load_lesson(root, lesson_path)
|
|
task = lesson.create_task("Implement auth")
|
|
original_run_git = lesson.run_git
|
|
|
|
def fail_after_add(args, cwd=None):
|
|
if args[:2] == ["worktree", "add"]:
|
|
ok, output = original_run_git(args, cwd=cwd)
|
|
self.assertTrue(ok, output)
|
|
return False, "simulated late add failure"
|
|
return original_run_git(args, cwd=cwd)
|
|
|
|
lesson.run_git = fail_after_add
|
|
result = lesson.create_worktree("auth", task.id)
|
|
|
|
self.assertIn("Partial operation", result)
|
|
self.assertIn("simulated late add failure", result)
|
|
self.assertIn("remains unbound", result)
|
|
self.assertIn("git worktree list", result)
|
|
self.assertTrue((lesson.WORKTREES_DIR / "auth").is_dir())
|
|
self.assertIsNone(lesson.load_task(task.id).worktree)
|
|
branch = subprocess.run(
|
|
["git", "show-ref", "--verify", "--quiet",
|
|
"refs/heads/wt/auth"],
|
|
cwd=root,
|
|
)
|
|
self.assertEqual(branch.returncode, 0)
|
|
|
|
def test_binding_failure_retains_created_git_data_for_recovery(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
init_git_repo(root)
|
|
lesson = load_lesson(root)
|
|
task = lesson.create_task("Implement auth")
|
|
original_save_task = lesson.save_task
|
|
|
|
def fail_binding(candidate):
|
|
if candidate.worktree == "auth":
|
|
raise OSError("simulated task persistence failure")
|
|
original_save_task(candidate)
|
|
|
|
lesson.save_task = fail_binding
|
|
result = lesson.create_worktree("auth", task.id)
|
|
|
|
self.assertIn("Partial success", result)
|
|
self.assertIn("manual recovery", result)
|
|
self.assertTrue((lesson.WORKTREES_DIR / "auth").is_dir())
|
|
self.assertIsNone(lesson.load_task(task.id).worktree)
|
|
branch = subprocess.run(
|
|
["git", "show-ref", "--verify", "--quiet",
|
|
"refs/heads/wt/auth"], cwd=root,
|
|
)
|
|
self.assertEqual(branch.returncode, 0)
|
|
|
|
def test_remove_worktree_refuses_dirty_checkout_by_default(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
init_git_repo(root)
|
|
lesson = load_lesson(root)
|
|
task = lesson.create_task("Implement auth")
|
|
lesson.create_worktree("auth", task.id)
|
|
lesson.claim_task(task.id, owner="alice")
|
|
lesson.complete_task(task.id, owner="alice")
|
|
lesson.release_completed_assignment("alice")
|
|
worktree = lesson.WORKTREES_DIR / "auth"
|
|
(worktree / "dirty.txt").write_text("unsaved\n")
|
|
|
|
denied = lesson.remove_worktree("auth")
|
|
|
|
self.assertIn("uncommitted", denied)
|
|
self.assertTrue(worktree.exists())
|
|
self.assertEqual(lesson.load_task(task.id).worktree, "auth")
|
|
|
|
def test_remove_worktree_treats_ignored_files_as_uncommitted_data(self):
|
|
for lesson_path in RUNTIME_LESSONS:
|
|
with self.subTest(lesson=lesson_path.parent.name):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
init_git_repo(root)
|
|
(root / ".gitignore").write_text("ignored.log\n")
|
|
subprocess.run(
|
|
["git", "add", ".gitignore"], cwd=root, check=True
|
|
)
|
|
subprocess.run(
|
|
["git", "commit", "-q", "-m", "ignore runtime log"],
|
|
cwd=root,
|
|
check=True,
|
|
)
|
|
lesson = load_lesson(root, lesson_path)
|
|
task = lesson.create_task("Implement auth")
|
|
lesson.create_worktree("auth", task.id)
|
|
lesson.claim_task(task.id, owner="alice")
|
|
lesson.complete_task(task.id, owner="alice")
|
|
lesson.release_completed_assignment("alice")
|
|
worktree = lesson.WORKTREES_DIR / "auth"
|
|
(worktree / "ignored.log").write_text("valuable output\n")
|
|
|
|
denied = lesson.remove_worktree("auth")
|
|
|
|
self.assertIn("uncommitted", denied)
|
|
self.assertTrue(worktree.exists())
|
|
self.assertEqual(lesson.load_task(task.id).worktree, "auth")
|
|
removed = lesson.remove_worktree(
|
|
"auth", discard_changes=True
|
|
)
|
|
self.assertIn("branch 'wt/auth' retained", removed)
|
|
self.assertFalse(worktree.exists())
|
|
|
|
def test_discard_removes_checkout_but_retains_branch(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
init_git_repo(root)
|
|
lesson = load_lesson(root)
|
|
task = lesson.create_task("Implement auth")
|
|
lesson.create_worktree("auth", task.id)
|
|
lesson.claim_task(task.id, owner="alice")
|
|
lesson.complete_task(task.id, owner="alice")
|
|
lesson.release_completed_assignment("alice")
|
|
worktree = lesson.WORKTREES_DIR / "auth"
|
|
(worktree / "dirty.txt").write_text("discard me\n")
|
|
|
|
removed = lesson.remove_worktree("auth", discard_changes=True)
|
|
|
|
self.assertIn("branch 'wt/auth' retained", removed)
|
|
self.assertFalse(worktree.exists())
|
|
self.assertIsNone(lesson.load_task(task.id).worktree)
|
|
branch = subprocess.run(
|
|
["git", "show-ref", "--verify", "--quiet",
|
|
"refs/heads/wt/auth"], cwd=root,
|
|
)
|
|
self.assertEqual(branch.returncode, 0)
|
|
|
|
def test_clean_local_commit_survives_non_force_checkout_removal(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
init_git_repo(root)
|
|
lesson = load_lesson(root)
|
|
task = lesson.create_task("Implement auth")
|
|
lesson.create_worktree("auth", task.id)
|
|
lesson.claim_task(task.id, owner="alice")
|
|
lesson.complete_task(task.id, owner="alice")
|
|
lesson.release_completed_assignment("alice")
|
|
worktree = lesson.WORKTREES_DIR / "auth"
|
|
(worktree / "feature.txt").write_text("committed work\n")
|
|
subprocess.run(
|
|
["git", "add", "feature.txt"], cwd=worktree, check=True
|
|
)
|
|
subprocess.run(
|
|
["git", "commit", "-q", "-m", "feature"],
|
|
cwd=worktree, check=True,
|
|
)
|
|
commit = subprocess.check_output(
|
|
["git", "rev-parse", "HEAD"], cwd=worktree, text=True
|
|
).strip()
|
|
upstream = subprocess.check_output(
|
|
["git", "for-each-ref", "--format=%(upstream)",
|
|
"refs/heads/wt/auth"], cwd=root, text=True,
|
|
).strip()
|
|
self.assertEqual(upstream, "")
|
|
|
|
removed = lesson.remove_worktree("auth")
|
|
|
|
self.assertIn("branch 'wt/auth' retained", removed)
|
|
self.assertFalse(worktree.exists())
|
|
retained = subprocess.check_output(
|
|
["git", "rev-parse", "wt/auth"], cwd=root, text=True
|
|
).strip()
|
|
self.assertEqual(retained, commit)
|
|
|
|
def test_active_task_blocks_normal_and_discard_removal(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
init_git_repo(root)
|
|
lesson = load_lesson(root)
|
|
task = lesson.create_task("Implement auth")
|
|
lesson.create_worktree("auth", task.id)
|
|
worktree = lesson.WORKTREES_DIR / "auth"
|
|
|
|
pending_normal = lesson.remove_worktree("auth")
|
|
pending_discard = lesson.remove_worktree(
|
|
"auth", discard_changes=True
|
|
)
|
|
self.assertIn("active task", pending_normal)
|
|
self.assertIn("active task", pending_discard)
|
|
|
|
lesson.claim_task(task.id, owner="alice")
|
|
progress_normal = lesson.remove_worktree("auth")
|
|
progress_discard = lesson.remove_worktree(
|
|
"auth", discard_changes=True
|
|
)
|
|
|
|
self.assertIn("active task", progress_normal)
|
|
self.assertIn("active task", progress_discard)
|
|
self.assertTrue(worktree.exists())
|
|
self.assertEqual(lesson.load_task(task.id).status, "in_progress")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|