fix: build task dependencies in two phases

tests/test_agent_teams_runtime.py

@@ -123,6 +123,13 @@ def claim_in_child(lesson_path: str, root: str, task_id: str, owner: str,
     results.put(lesson.claim_task(task_id, owner=owner))
 
 
+def update_in_child(lesson_path: str, root: str, task_id: str,
+                    dependency_id: str, barrier, results):
+    lesson = load_lesson(Path(root), Path(lesson_path))
+    barrier.wait()
+    results.put(lesson.run_update_task(task_id, [dependency_id]))
+
+
 class AgentTeamsRuntimeTests(unittest.TestCase):
     def test_downstream_lessons_execute_the_merged_runtime_contract(self):
         for lesson_path in RUNTIME_LESSONS:
@@ -140,6 +147,54 @@ class AgentTeamsRuntimeTests(unittest.TestCase):
                     self.assertTrue(lesson.release_completed_assignment("alice"))
                     self.assertNotIn("alice", lesson.teammate_assignments)
 
+    def test_task_dependencies_use_runtime_ids_and_are_lead_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
+                    tools = {tool["name"]: tool for tool in tool_defs}
+
+                    self.assertIn("update_task", tools)
+                    self.assertNotIn(
+                        "blockedBy",
+                        tools["create_task"]["input_schema"]["properties"],
+                    )
+                    self.assertIn(
+                        "runtime-generated IDs",
+                        lesson.PROMPT_SECTIONS["tasks"],
+                    )
+
+                    dependency = lesson.create_task("Create schema")
+                    target = lesson.create_task("Write API")
+                    self.assertIn(
+                        "Updated",
+                        lesson.run_update_task(target.id, [dependency.id]),
+                    )
+                    self.assertEqual(
+                        lesson.load_task(target.id).blockedBy, [dependency.id]
+                    )
+
+                    captured_tools = []
+
+                    def stop_after_capture(**kwargs):
+                        captured_tools.extend(
+                            tool["name"] for tool in kwargs["tools"]
+                        )
+                        raise RuntimeError("stop after capturing teammate tools")
+
+                    lesson.client.messages.create = stop_after_capture
+                    lesson.spawn_teammate_thread(
+                        "tool-auditor", "reviewer", "Inspect the task board."
+                    )
+                    self.assertTrue(wait_until(
+                        lambda: "tool-auditor" not in lesson.active_teammates
+                    ))
+                    self.assertIn("claim_task", captured_tools)
+                    self.assertNotIn("update_task", captured_tools)
+
     def test_inbox_delivery_is_runtime_owned(self):
         with tempfile.TemporaryDirectory() as tmp:
             lesson = load_lesson(Path(tmp))
@@ -669,6 +724,9 @@ class AgentTeamsRuntimeTests(unittest.TestCase):
                             with self.subTest(tool=tool_name, task_id=task_id):
                                 result = getattr(lesson, tool_name)(task_id)
                                 self.assertIn("Error:", result)
+                        self.assertIn(
+                            "Error:", lesson.run_update_task(task_id, [])
+                        )
 
     def test_plan_gate_blocks_mutating_tools_until_approval(self):
         with tempfile.TemporaryDirectory() as tmp:
@@ -1621,6 +1679,55 @@ class AgentTeamsRuntimeTests(unittest.TestCase):
                     self.assertEqual(persisted.status, "in_progress")
                     self.assertIn(persisted.owner, {"alice", "bob"})
 
+    def test_dependency_updates_are_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)
+                    first = lesson.create_task("First")
+                    second = lesson.create_task("Second")
+                    barrier = context.Barrier(3)
+                    results = context.Queue()
+                    workers = [
+                        context.Process(
+                            target=update_in_child,
+                            args=(
+                                str(lesson_path), tmp, task_id, dependency_id,
+                                barrier, results,
+                            ),
+                        )
+                        for task_id, dependency_id in (
+                            (first.id, second.id),
+                            (second.id, first.id),
+                        )
+                    ]
+
+                    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("Updated ") for outcome in outcomes),
+                        1,
+                    )
+                    self.assertEqual(
+                        sum("Dependency cycle detected" in outcome
+                            for outcome in outcomes),
+                        1,
+                    )
+                    persisted = {
+                        first.id: lesson.load_task(first.id).blockedBy,
+                        second.id: lesson.load_task(second.id).blockedBy,
+                    }
+                    self.assertEqual(
+                        sum(bool(value) for value in persisted.values()), 1
+                    )
+
     def test_plan_approval_cannot_cross_assignment_boundary(self):
         for lesson_path in RUNTIME_LESSONS:
             with self.subTest(lesson=lesson_path.parent.name):

tests/test_task_system.py

@@ -73,6 +73,7 @@ def test_s10_keeps_the_s04_kernel_and_adds_task_tools() -> None:
             "edit_file",
             "glob",
             "create_task",
+            "update_task",
             "list_tasks",
             "get_task",
             "claim_task",
@@ -83,6 +84,13 @@ def test_s10_keeps_the_s04_kernel_and_adds_task_tools() -> None:
         assert not hasattr(lesson, "MEMORY_DIR")
         assert not (workdir / ".tasks").exists()
 
+        tools = {tool["name"]: tool for tool in lesson.TOOLS}
+        create_schema = tools["create_task"]["input_schema"]
+        update_schema = tools["update_task"]["input_schema"]
+        assert "blockedBy" not in create_schema["properties"]
+        assert create_schema["additionalProperties"] is False
+        assert update_schema["required"] == ["task_id", "addBlockedBy"]
+
 
 def test_dependencies_gate_claim_and_completion_checks_owner() -> None:
     with tempfile.TemporaryDirectory() as tmp:
@@ -90,7 +98,8 @@ def test_dependencies_gate_claim_and_completion_checks_owner() -> None:
         lesson = load_lesson(workdir)
 
         schema = lesson.create_task("create schema")
-        api = lesson.create_task("write API", blockedBy=[schema.id])
+        api = lesson.create_task("write API")
+        lesson.update_task(api.id, [schema.id])
 
         assert lesson.claim_task(api.id) == f"Blocked by: ['{schema.id}']"
         assert "Claimed" in lesson.claim_task(schema.id)
@@ -103,6 +112,41 @@ def test_dependencies_gate_claim_and_completion_checks_owner() -> None:
         assert lesson.load_task(api.id).status == "completed"
 
 
+def test_dependencies_are_added_after_create_returns_runtime_ids() -> None:
+    with tempfile.TemporaryDirectory() as tmp:
+        lesson = load_lesson(Path(tmp))
+
+        create_results = [
+            lesson.execute_tool(tool_call("create_task", subject=subject))
+            for subject in (
+                "create schema",
+                "write API",
+                "write tests",
+                "write docs",
+            )
+        ]
+        task_ids = [result.split()[1].rstrip(":") for result in create_results]
+        schema_id, api_id, tests_id, docs_id = task_ids
+
+        update_results = [
+            lesson.execute_tool(tool_call(
+                "update_task", task_id=api_id, addBlockedBy=[schema_id]
+            )),
+            lesson.execute_tool(tool_call(
+                "update_task", task_id=tests_id, addBlockedBy=[api_id]
+            )),
+            lesson.execute_tool(tool_call(
+                "update_task", task_id=docs_id, addBlockedBy=[schema_id]
+            )),
+        ]
+
+        assert all(not result.startswith("Error:") for result in update_results)
+        assert lesson.load_task(schema_id).blockedBy == []
+        assert lesson.load_task(api_id).blockedBy == [schema_id]
+        assert lesson.load_task(tests_id).blockedBy == [api_id]
+        assert lesson.load_task(docs_id).blockedBy == [schema_id]
+
+
 def test_invalid_and_missing_task_ids_become_tool_results() -> None:
     with tempfile.TemporaryDirectory() as tmp:
         lesson = load_lesson(Path(tmp))
@@ -132,17 +176,49 @@ def test_create_retries_instead_of_overwriting_an_existing_id(
         assert [task.subject for task in lesson.list_tasks()] == ["second", "first"]
 
 
-def test_create_rejects_unknown_dependencies() -> None:
+def test_update_rejects_invalid_graph_changes_without_partial_mutation() -> None:
     with tempfile.TemporaryDirectory() as tmp:
         lesson = load_lesson(Path(tmp))
+        dependency = lesson.create_task("create schema")
+        target = lesson.create_task("write API")
 
-        output = lesson.execute_tool(tool_call(
-            "create_task",
-            subject="write API",
-            blockedBy=["task_00000000"],
+        missing = lesson.execute_tool(tool_call(
+            "update_task",
+            task_id=target.id,
+            addBlockedBy=[dependency.id, "task_00000000"],
+        ))
+        self_dependency = lesson.execute_tool(tool_call(
+            "update_task", task_id=target.id, addBlockedBy=[target.id]
         ))
 
-        assert output == "Error: Dependency not found: task_00000000"
+        assert missing == "Error: Dependency not found: task_00000000"
+        assert self_dependency == "Error: Task cannot depend on itself"
+        assert lesson.load_task(target.id).blockedBy == []
+
+
+def test_update_is_idempotent_and_rejects_cycles_or_started_tasks() -> None:
+    with tempfile.TemporaryDirectory() as tmp:
+        lesson = load_lesson(Path(tmp))
+        first = lesson.create_task("first")
+        second = lesson.create_task("second")
+        third = lesson.create_task("third")
+
+        lesson.update_task(second.id, [first.id, first.id])
+        lesson.update_task(second.id, [first.id])
+        lesson.update_task(third.id, [second.id])
+
+        cycle = lesson.execute_tool(tool_call(
+            "update_task", task_id=first.id, addBlockedBy=[third.id]
+        ))
+        assert cycle.startswith("Error: Dependency cycle detected")
+        assert lesson.load_task(first.id).blockedBy == []
+        assert lesson.load_task(second.id).blockedBy == [first.id]
+
+        assert "Claimed" in lesson.claim_task(first.id)
+        started = lesson.execute_tool(tool_call(
+            "update_task", task_id=first.id, addBlockedBy=[second.id]
+        ))
+        assert "only be updated while pending and unowned" in started
 
 
 def test_task_store_rejects_a_symlink_outside_the_workspace() -> None:

tests/test_web_scenarios.py

@@ -25,6 +25,47 @@ def load_lesson(name: str, script: Path):
     return module
 
 
+def test_s10_scenario_builds_the_task_graph_in_two_phases() -> None:
+    steps = load_scenario("s10")["steps"]
+    create_calls = [
+        (index, json.loads(step["content"]))
+        for index, step in enumerate(steps)
+        if step.get("toolName") == "create_task"
+        and step["type"] == "tool_call"
+    ]
+    create_results = [
+        (index, step["content"])
+        for index, step in enumerate(steps)
+        if step.get("toolName") == "create_task"
+        and step["type"] == "tool_result"
+    ]
+    update_index = next(
+        index for index, step in enumerate(steps)
+        if step.get("toolName") == "update_task"
+        and step["type"] == "tool_call"
+    )
+    update = json.loads(steps[update_index]["content"])
+    task_ids = [
+        re.fullmatch(r"Created (task_[0-9a-f]{8}): .+", content).group(1)
+        for _, content in create_results
+    ]
+
+    assert len(create_calls) == len(create_results) == 2
+    assert all("blockedBy" not in content for _, content in create_calls)
+    assert max(index for index, _ in create_results) < update_index
+    assert update == {
+        "task_id": task_ids[1],
+        "addBlockedBy": [task_ids[0]],
+    }
+    claim_inputs = [
+        json.loads(step["content"])
+        for step in steps
+        if step.get("toolName") == "claim_task"
+        and step["type"] == "tool_call"
+    ]
+    assert all(set(claim_input) == {"task_id"} for claim_input in claim_inputs)
+
+
 def test_s13_scenario_uses_the_real_plan_protocol() -> None:
     steps = load_scenario("s13")["steps"]
     spawn = next(
@@ -48,6 +89,21 @@ def test_s13_scenario_uses_the_real_plan_protocol() -> None:
         index for index, step in enumerate(steps)
         if "plan_approval_response" in step.get("content", "")
     )
+    create_indices = [
+        index for index, step in enumerate(steps)
+        if step.get("toolName") == "create_task"
+        and step["type"] == "tool_call"
+    ]
+    update_index = next(
+        index for index, step in enumerate(steps)
+        if step.get("toolName") == "update_task"
+        and step["type"] == "tool_call"
+    )
+    first_spawn_index = next(
+        index for index, step in enumerate(steps)
+        if step.get("toolName") == "spawn_teammate"
+        and step["type"] == "tool_call"
+    )
 
     review = json.loads(steps[review_index]["content"])
     spawn_input = json.loads(spawn["content"])
@@ -58,6 +114,15 @@ def test_s13_scenario_uses_the_real_plan_protocol() -> None:
     assert re.fullmatch(r"req_\d{6}", review["request_id"])
     assert review["approve"] is True
     assert "approved" not in review
+    assert all(
+        "blockedBy" not in json.loads(steps[index]["content"])
+        for index in create_indices
+    )
+    assert max(create_indices) < update_index < first_spawn_index
+    assert json.loads(steps[update_index]["content"]) == {
+        "task_id": "task_5e6f7a8b",
+        "addBlockedBy": ["task_1a2b3c4d"],
+    }
 
 
 def test_s15_scenario_calls_the_discovered_mcp_tool() -> None:

tests/test_workflow_goal_lessons.py

@@ -473,8 +473,9 @@ def test_workflow_default_entry_extends_the_real_s15_host(
     tools, handlers = host.assemble_tool_pool()
     names = [tool["name"] for tool in tools]
 
-    assert len(host.BUILTIN_TOOLS) == 25
+    assert len(host.BUILTIN_TOOLS) == 26
     assert names[:-1] == [tool["name"] for tool in host.BUILTIN_TOOLS]
+    assert "update_task" in names
     assert names[-1] == "Workflow"
     assert handlers["Workflow"] is workflow.run_workflow_sync
     assert handlers["Workflow"](name="missing") == (