test(orchestration): cover mission guard corner cases

agent/tests/test_client_root_task_spec.py

@@ -0,0 +1,44 @@
+import pytest
+
+import agent.client as client_module
+
+
+@pytest.mark.asyncio
+async def test_invoke_agent_forwards_root_task_spec_only_to_local_runner(monkeypatch):
+    captured = {}
+
+    async def fake_local(**kwargs):
+        captured.update(kwargs)
+        return {"mode": "local", "status": "completed"}
+
+    monkeypatch.setattr(client_module, "_run_local_agent", fake_local)
+    root_task_spec = {
+        "objective": "Complete the mission",
+        "acceptance_criteria": [
+            {"criterion_id": "done", "description": "Mission is complete"}
+        ],
+    }
+
+    result = await client_module.invoke_agent(
+        agent_type="planner",
+        task="start",
+        project_root="/project",
+        root_task_spec=root_task_spec,
+    )
+
+    assert result["status"] == "completed"
+    assert captured["root_task_spec"] is root_task_spec
+
+
+@pytest.mark.asyncio
+async def test_invoke_agent_rejects_remote_root_task_spec():
+    result = await client_module.invoke_agent(
+        agent_type="remote_planner",
+        task="start",
+        root_task_spec={"objective": "must stay local"},
+    )
+
+    assert result["mode"] == "remote"
+    assert result["status"] == "failed"
+    assert "root_task_spec" in result["error"]
+    assert "本地" in result["error"]

agent/tests/test_mission_root.py

@@ -0,0 +1,427 @@
+import json
+
+import pytest
+
+from agent.orchestration.coordinator import TaskConflict, TaskCoordinator
+from agent.orchestration.models import DecisionAction, TaskLedger, TaskStatus, ValidationVerdict
+from agent.orchestration.store import (
+    FileSystemArtifactStore,
+    FileSystemTaskStore,
+    TaskStoreError,
+)
+from agent.trace.store import FileSystemTraceStore
+
+from test_coordinator_integration import (
+    FakeExecutor,
+    ROOT_TASK_SPEC,
+    create_task,
+    make_coordinator,
+)
+
+
+@pytest.mark.asyncio
+async def test_new_ledger_has_one_stable_root_and_inspect_exposes_it(tmp_path):
+    coordinator, store, _ = await make_coordinator(tmp_path, FakeExecutor([]))
+
+    ledger = await store.load("root")
+    root_id = ledger.root_task_id
+    assert root_id is not None
+    assert ledger.focused_task_id == root_id
+    assert ledger.tasks[root_id].parent_task_id is None
+    assert ledger.tasks[root_id].display_path == "0"
+    assert ledger.tasks[root_id].status == TaskStatus.NEEDS_REPLAN
+    assert [task.task_id for task in ledger.tasks.values() if task.parent_task_id is None] == [root_id]
+
+    same = await coordinator.ensure_ledger("root", None)
+    assert same.root_task_id == root_id
+    inspected = await coordinator.inspect_tasks("root")
+    assert inspected["root_task_id"] == root_id
+    assert inspected["tasks"][0]["current_spec"]["acceptance_criteria"][0][
+        "criterion_id"
+    ] == "mission-done"
+
+
+@pytest.mark.asyncio
+async def test_new_ledger_requires_root_spec_and_rootless_policy_is_explicit(tmp_path):
+    store = FileSystemTaskStore(str(tmp_path))
+    coordinator = TaskCoordinator(
+        store,
+        FileSystemArtifactStore(str(tmp_path)),
+        FileSystemTraceStore(str(tmp_path)),
+    )
+    with pytest.raises(ValueError, match="root_task_spec is required"):
+        await coordinator.ensure_ledger("missing")
+    with pytest.raises(TaskConflict, match="has no Root Task ledger"):
+        await coordinator.ensure_ledger(
+            "missing-resume", ROOT_TASK_SPEC, allow_create=False
+        )
+
+    legacy = TaskLedger(root_trace_id="legacy", mission="legacy")
+    committed = await store.commit(legacy, expected_revision=-1)
+    with pytest.raises(TaskConflict, match="Rootless"):
+        await coordinator.ensure_ledger("legacy", ROOT_TASK_SPEC)
+    unchanged = await store.load("legacy")
+    assert unchanged.root_task_id is None
+    assert unchanged.revision == committed.ledger.revision
+
+    invalid_path = store.ledger_path("legacy-invalid")
+    invalid_path.parent.mkdir(parents=True, exist_ok=True)
+    invalid_path.write_text(json.dumps({
+        "root_trace_id": "legacy-invalid",
+        "mission": "legacy invalid",
+        "tasks": {
+            "old-task": {
+                "task_id": "old-task",
+                "parent_task_id": None,
+                "display_path": "1",
+                "specs": [{
+                    "version": 1,
+                    "objective": "old task",
+                    "acceptance_criteria": [],
+                }],
+            }
+        },
+    }), encoding="utf-8")
+    with pytest.raises(TaskStoreError, match="at least one criterion"):
+        await store.load("legacy-invalid")
+
+
+@pytest.mark.asyncio
+async def test_parent_waits_until_every_child_is_terminal(tmp_path):
+    coordinator, store, _ = await make_coordinator(tmp_path, FakeExecutor([]))
+    first = await create_task(coordinator, "first")
+    second = await create_task(coordinator, "second")
+    third = await create_task(coordinator, "third")
+    root_id = (await store.load("root")).root_task_id
+
+    await coordinator.decide_task(
+        "root", first, None, DecisionAction.CANCEL, {"reason": "drop first"}, "cancel-first"
+    )
+    assert (await store.load("root")).tasks[root_id].status == TaskStatus.WAITING_CHILDREN
+
+    superseded = await coordinator.decide_task(
+        "root",
+        second,
+        None,
+        DecisionAction.SUPERSEDE,
+        {
+            "reason": "replace second",
+            "replacement": {
+                "objective": "second replacement",
+                "acceptance_criteria": [
+                    {"criterion_id": "replacement", "description": "replacement passes"}
+                ],
+            },
+        },
+        "supersede-second",
+    )
+    replacement = superseded["payload"]["replacement_task_id"]
+    assert (await store.load("root")).tasks[root_id].status == TaskStatus.WAITING_CHILDREN
+
+    await coordinator.decide_task(
+        "root", replacement, None, DecisionAction.CANCEL, {"reason": "drop replacement"}, "cancel-replacement"
+    )
+    assert (await store.load("root")).tasks[root_id].status == TaskStatus.WAITING_CHILDREN
+    await coordinator.decide_task(
+        "root", third, None, DecisionAction.CANCEL, {"reason": "drop third"}, "cancel-third"
+    )
+    assert (await store.load("root")).tasks[root_id].status == TaskStatus.NEEDS_REPLAN
+
+
+@pytest.mark.asyncio
+async def test_root_executes_after_children_and_receives_only_accepted_results(tmp_path):
+    class CapturingExecutor(FakeExecutor):
+        def __init__(self):
+            super().__init__([ValidationVerdict.PASSED, ValidationVerdict.PASSED])
+            self.worker_contexts = []
+
+        async def run_worker(self, context):
+            self.worker_contexts.append(context)
+            return await super().run_worker(context)
+
+    executor = CapturingExecutor()
+    coordinator, store, _ = await make_coordinator(tmp_path, executor)
+    child_id = await create_task(coordinator, "child")
+    root_id = (await store.load("root")).root_task_id
+
+    blocked_dispatch = (await coordinator.dispatch_tasks("root", [root_id]))[0]
+    assert "cannot dispatch" in blocked_dispatch.error
+
+    child_cycle = (await coordinator.dispatch_tasks("root", [child_id]))[0]
+    await coordinator.decide_task(
+        "root",
+        child_id,
+        child_cycle.validation_id,
+        DecisionAction.ACCEPT,
+        {"reason": "child accepted"},
+        "accept-child",
+    )
+    accepted_ledger = await store.load("root")
+    assert accepted_ledger.tasks[root_id].status == TaskStatus.NEEDS_REPLAN
+    assert [
+        item["task_id"]
+        for item in coordinator._accepted_child_results(
+            accepted_ledger, accepted_ledger.tasks[root_id]
+        )
+    ] == [child_id]
+
+    root_cycle = (await coordinator.dispatch_tasks("root", [root_id]))[0]
+    assert root_cycle.error is None
+    assert len(executor.worker_contexts) == 2
+    child_results = executor.worker_contexts[-1]["accepted_child_results"]
+    assert [item["task_id"] for item in child_results] == [child_id]
+    assert child_results[0]["attempt_id"] == child_cycle.attempt_id
+    assert child_results[0]["validation"]["validation_id"] == child_cycle.validation_id
+
+    await coordinator.decide_task(
+        "root",
+        root_id,
+        root_cycle.validation_id,
+        DecisionAction.ACCEPT,
+        {"reason": "root accepted"},
+        "accept-root",
+    )
+    completion = await coordinator.mission_completion("root")
+    assert completion["status"] == TaskStatus.COMPLETED.value
+    assert completion["result_summary"] == "done"
+
+
+@pytest.mark.asyncio
+async def test_root_identity_is_not_cancellable_or_supersedable(tmp_path):
+    coordinator, store, _ = await make_coordinator(tmp_path, FakeExecutor([]))
+    root_id = (await store.load("root")).root_task_id
+
+    with pytest.raises(TaskConflict, match="root task cannot be cancelled"):
+        await coordinator.decide_task(
+            "root", root_id, None, DecisionAction.CANCEL, {"reason": "invalid"}, "cancel-root"
+        )
+    with pytest.raises(TaskConflict, match="root task cannot be superseded"):
+        await coordinator.decide_task(
+            "root",
+            root_id,
+            None,
+            DecisionAction.SUPERSEDE,
+            {
+                "reason": "invalid",
+                "replacement": {
+                    "objective": "replacement",
+                    "acceptance_criteria": [
+                        {"criterion_id": "replacement", "description": "replacement passes"}
+                    ],
+                },
+            },
+            "supersede-root",
+        )
+
+
+@pytest.mark.asyncio
+async def test_parent_with_active_descendant_cannot_be_terminated(tmp_path):
+    coordinator, store, _ = await make_coordinator(tmp_path, FakeExecutor([]))
+    parent_id = await create_task(coordinator, "parent")
+    descendant_id = await create_task(
+        coordinator, "descendant", parent_task_id=parent_id
+    )
+    root_id = (await store.load("root")).root_task_id
+
+    with pytest.raises(TaskConflict, match="non-terminal descendants"):
+        await coordinator.decide_task(
+            "root",
+            parent_id,
+            None,
+            DecisionAction.CANCEL,
+            {"reason": "invalid while descendant is active"},
+            "cancel-active-parent",
+        )
+    with pytest.raises(TaskConflict, match="non-terminal descendants"):
+        await coordinator.decide_task(
+            "root",
+            parent_id,
+            None,
+            DecisionAction.SUPERSEDE,
+            {
+                "reason": "invalid while descendant is active",
+                "replacement": {
+                    "objective": "replacement",
+                    "acceptance_criteria": [
+                        {"criterion_id": "replacement", "description": "replacement passes"}
+                    ],
+                },
+            },
+            "supersede-active-parent",
+        )
+    root_dispatch = (await coordinator.dispatch_tasks("root", [root_id]))[0]
+    assert "cannot dispatch" in root_dispatch.error
+
+    await coordinator.decide_task(
+        "root",
+        descendant_id,
+        None,
+        DecisionAction.CANCEL,
+        {"reason": "finish descendant"},
+        "cancel-descendant",
+    )
+    await coordinator.decide_task(
+        "root",
+        parent_id,
+        None,
+        DecisionAction.CANCEL,
+        {"reason": "parent can now terminate"},
+        "cancel-parent",
+    )
+    assert (await store.load("root")).tasks[root_id].status == TaskStatus.NEEDS_REPLAN
+
+
+@pytest.mark.asyncio
+async def test_blocked_descendants_allow_mission_to_report_blocked(tmp_path):
+    coordinator, store, _ = await make_coordinator(tmp_path, FakeExecutor([]))
+    parent_id = await create_task(coordinator, "parent")
+    child_id = await create_task(coordinator, "child", parent_task_id=parent_id)
+    root_id = (await store.load("root")).root_task_id
+
+    await coordinator.decide_task(
+        "root",
+        child_id,
+        None,
+        DecisionAction.BLOCK,
+        {"reason": "child external dependency"},
+        "block-child",
+    )
+    await coordinator.decide_task(
+        "root",
+        parent_id,
+        None,
+        DecisionAction.BLOCK,
+        {"reason": "parent waits on blocked child"},
+        "block-parent",
+    )
+    await coordinator.decide_task(
+        "root",
+        root_id,
+        None,
+        DecisionAction.BLOCK,
+        {"reason": "mission external dependency"},
+        "block-root",
+    )
+
+    completion = await coordinator.mission_completion("root")
+    assert completion["status"] == TaskStatus.BLOCKED.value
+    assert completion["blocked_reason"] == "mission external dependency"
+
+    await coordinator.decide_task(
+        "root",
+        root_id,
+        None,
+        DecisionAction.UNBLOCK,
+        {},
+        "unblock-root",
+    )
+    assert (await store.load("root")).tasks[root_id].status == TaskStatus.NEEDS_REPLAN
+
+
+@pytest.mark.asyncio
+async def test_parent_cannot_block_while_descendant_attempt_is_active(tmp_path):
+    coordinator, _, _ = await make_coordinator(tmp_path, FakeExecutor([]))
+    parent_id = await create_task(coordinator, "parent")
+    child_id = await create_task(coordinator, "child", parent_task_id=parent_id)
+    await coordinator._create_attempt("root", child_id, "worker", None)
+
+    with pytest.raises(TaskConflict, match="active descendants"):
+        await coordinator.decide_task(
+            "root",
+            parent_id,
+            None,
+            DecisionAction.BLOCK,
+            {"reason": "must not hide active work"},
+            "block-active-parent",
+        )
+
+
+@pytest.mark.asyncio
+async def test_inserting_a_sibling_rebases_its_existing_subtree(tmp_path):
+    coordinator, store, _ = await make_coordinator(tmp_path, FakeExecutor([]))
+    first_id = await create_task(coordinator, "first")
+    moved_id = await create_task(coordinator, "moved")
+    descendant_id = await create_task(
+        coordinator, "moved descendant", parent_task_id=moved_id
+    )
+
+    await coordinator.create_tasks(
+        "root",
+        [{
+            "objective": "inserted",
+            "acceptance_criteria": [
+                {"criterion_id": "inserted", "description": "inserted passes"}
+            ],
+        }],
+        placement={"after_task_id": first_id},
+    )
+    ledger = await store.load("root")
+    assert ledger.tasks[moved_id].display_path == "0.3"
+    assert ledger.tasks[descendant_id].display_path == "0.3.1"
+
+
+@pytest.mark.asyncio
+async def test_invalid_batch_and_revision_are_atomic(tmp_path):
+    executor = FakeExecutor([ValidationVerdict.FAILED])
+    coordinator, store, _ = await make_coordinator(tmp_path, executor)
+    root_id = (await store.load("root")).root_task_id
+
+    with pytest.raises(ValueError, match="at least one criterion"):
+        await coordinator.create_tasks(
+            "root",
+            [
+                {
+                    "objective": "valid first draft",
+                    "acceptance_criteria": [
+                        {"criterion_id": "valid", "description": "valid passes"}
+                    ],
+                },
+                {"objective": "invalid second draft"},
+            ],
+        )
+    ledger = await store.load("root")
+    assert list(ledger.tasks) == [root_id]
+    assert ledger.tasks[root_id].status == TaskStatus.NEEDS_REPLAN
+
+    with pytest.raises(ValueError, match="context_refs"):
+        await coordinator.create_tasks(
+            "root",
+            [{
+                "objective": "invalid context refs",
+                "acceptance_criteria": [
+                    {"criterion_id": "context", "description": "context passes"}
+                ],
+                "context_refs": "memory://not-a-list",
+            }],
+        )
+
+    task_id = await create_task(coordinator, "revise atomically")
+    cycle = (await coordinator.dispatch_tasks("root", [task_id]))[0]
+    for index, criteria in enumerate(([], None, "criterion")):
+        with pytest.raises(ValueError, match="acceptance_criteria|at least one criterion"):
+            await coordinator.decide_task(
+                "root",
+                task_id,
+                cycle.validation_id,
+                DecisionAction.REVISE,
+                {"reason": "invalid revision", "acceptance_criteria": criteria},
+                f"invalid-revision-{index}",
+            )
+    task = (await store.load("root")).tasks[task_id]
+    assert task.current_spec_version == 1
+    assert len(task.specs) == 1
+    assert task.status == TaskStatus.AWAITING_DECISION
+
+    for index, objective in enumerate((123, None, "", "   ")):
+        with pytest.raises(ValueError, match="objective"):
+            await coordinator.decide_task(
+                "root",
+                task_id,
+                cycle.validation_id,
+                DecisionAction.REVISE,
+                {"reason": "invalid objective", "objective": objective},
+                f"invalid-objective-{index}",
+            )
+        unchanged = (await store.load("root")).tasks[task_id]
+        assert unchanged.current_spec_version == 1
+        assert unchanged.status == TaskStatus.AWAITING_DECISION

agent/tests/test_runner_completion_gate.py

@@ -0,0 +1,303 @@
+import pytest
+
+from agent.core.presets import AgentPreset, register_preset
+from agent.core.runner import AgentRunner, RunConfig
+from agent.orchestration.models import AgentRole, CompletionPolicy
+from agent.trace.models import Trace
+from agent.trace.store import FileSystemTraceStore
+
+
+class StubCoordinator:
+    def __init__(self, snapshots):
+        self.snapshots = list(snapshots)
+        self.ensure_calls = []
+        self.completion_calls = 0
+
+    async def ensure_ledger(self, trace_id, mission, *, allow_create=True):
+        self.ensure_calls.append((trace_id, mission))
+
+    async def mission_completion(self, root_trace_id):
+        self.completion_calls += 1
+        if len(self.snapshots) > 1:
+            return self.snapshots.pop(0)
+        return self.snapshots[0]
+
+
+def _knowledge_off():
+    from agent.tools.builtin.knowledge import KnowledgeConfig
+
+    return KnowledgeConfig(
+        enable_extraction=False,
+        enable_completion_extraction=False,
+        enable_injection=False,
+    )
+
+
+def _register_role_preset(name, role, max_iterations):
+    register_preset(
+        name,
+        AgentPreset(
+            role=role,
+            allowed_tools=[],
+            max_iterations=max_iterations,
+            skills=[],
+        ),
+    )
+
+
+def _explicit_config(name, max_iterations):
+    root_task_spec = {
+        "objective": "complete the mission",
+        "acceptance_criteria": [{"description": "mission result is validated"}],
+    }
+    return RunConfig(
+        agent_type=name,
+        completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
+        max_iterations=max_iterations,
+        tools=[],
+        tool_groups=[],
+        root_task_spec=root_task_spec,
+        knowledge=_knowledge_off(),
+    )
+
+
+@pytest.mark.asyncio
+async def test_planner_no_tool_exit_is_persisted_and_loop_continues(tmp_path):
+    preset = "test_completion_gate_planner"
+    _register_role_preset(preset, AgentRole.PLANNER, 3)
+    coordinator = StubCoordinator([
+        {
+            "status": "needs_replan",
+            "root_task_id": "root-task",
+            "pending_tasks": [{"task_id": "child", "status": "pending"}],
+        },
+        {
+            "status": "completed",
+            "root_task_id": "root-task",
+            "pending_tasks": [],
+            "result_summary": "accepted root result",
+        },
+    ])
+    calls = []
+
+    async def llm_call(**kwargs):
+        calls.append(kwargs["messages"])
+        return {"content": "", "tool_calls": None, "finish_reason": "stop"}
+
+    store = FileSystemTraceStore(str(tmp_path))
+    runner = AgentRunner(
+        trace_store=store,
+        llm_call=llm_call,
+        task_coordinator=coordinator,
+    )
+    events = [
+        event
+        async for event in runner.run(
+            [{"role": "user", "content": "run mission"}],
+            _explicit_config(preset, 3),
+        )
+    ]
+
+    final = events[-1]
+    assert isinstance(final, Trace)
+    assert final.status == "completed"
+    assert final.result_summary == "accepted root result"
+    assert coordinator.ensure_calls[0][1]["objective"] == "complete the mission"
+    assert len(calls) == 2
+    assert any(
+        message.get("role") == "user"
+        and "[FRAMEWORK_MISSION_INCOMPLETE]" in message.get("content", "")
+        for message in calls[1]
+    )
+    assert calls[1][0]["role"] == "system"
+    assert "Framework role contract: Planner" in calls[1][0]["content"]
+    persisted = await store.get_trace_messages(final.trace_id)
+    controls = [
+        message for message in persisted
+        if message.role == "user"
+        and "[FRAMEWORK_MISSION_INCOMPLETE]" in str(message.content)
+    ]
+    assert len(controls) == 1
+
+
+@pytest.mark.asyncio
+async def test_incomplete_planner_resume_restores_explicit_identity(tmp_path):
+    preset = "test_completion_gate_resume"
+    _register_role_preset(preset, AgentRole.PLANNER, 1)
+    coordinator = StubCoordinator([{
+        "status": "needs_replan",
+        "root_task_id": "root-task",
+        "pending_tasks": [{"task_id": "root-task", "status": "needs_replan"}],
+    }])
+
+    async def llm_call(**_kwargs):
+        return {"content": "", "tool_calls": None, "finish_reason": "stop"}
+
+    store = FileSystemTraceStore(str(tmp_path))
+    runner = AgentRunner(
+        trace_store=store,
+        llm_call=llm_call,
+        task_coordinator=coordinator,
+    )
+    first = await runner.run_result(
+        [{"role": "user", "content": "run mission"}],
+        _explicit_config(preset, 1),
+    )
+    assert first["status"] == "incomplete"
+
+    resume_config = RunConfig(
+        trace_id=first["trace_id"],
+        knowledge=_knowledge_off(),
+    )
+    resumed = await runner.run_result([], resume_config)
+
+    assert resumed["status"] == "incomplete"
+    assert resumed["error"] == "iteration_limit_reached"
+    assert resume_config.agent_type == preset
+    assert resume_config.completion_policy == CompletionPolicy.EXPLICIT_VALIDATION
+
+
+@pytest.mark.asyncio
+async def test_explicit_planner_rewind_and_wrong_role_resume_are_rejected_before_mutation(
+    tmp_path,
+):
+    planner_preset = "test_completion_gate_rewind_planner"
+    worker_preset = "test_completion_gate_wrong_role_worker"
+    _register_role_preset(planner_preset, AgentRole.PLANNER, 2)
+    _register_role_preset(worker_preset, AgentRole.WORKER, 2)
+    store = FileSystemTraceStore(str(tmp_path))
+    planner_trace = Trace(
+        trace_id="planner-trace",
+        mode="agent",
+        task="mission",
+        agent_type=planner_preset,
+        agent_role=AgentRole.PLANNER.value,
+        status="completed",
+        head_sequence=0,
+        last_sequence=10,
+    )
+    worker_trace = Trace(
+        trace_id="worker-trace",
+        mode="agent",
+        task="worker",
+        agent_type=worker_preset,
+        agent_role=AgentRole.WORKER.value,
+        status="completed",
+    )
+    await store.create_trace(planner_trace)
+    await store.create_trace(worker_trace)
+    coordinator = StubCoordinator([{"status": "completed"}])
+    runner = AgentRunner(
+        trace_store=store,
+        llm_call=lambda **_kwargs: None,
+        task_coordinator=coordinator,
+    )
+
+    with pytest.raises(ValueError, match="do not support rewind"):
+        await runner.run_result(
+            [],
+            RunConfig(
+                trace_id=planner_trace.trace_id,
+                agent_type=planner_preset,
+                completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
+                after_sequence=5,
+            ),
+        )
+    with pytest.raises(ValueError, match="Trace role mismatch"):
+        await runner.run_result(
+            [],
+            RunConfig(
+                trace_id=worker_trace.trace_id,
+                agent_type=planner_preset,
+                completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
+            ),
+        )
+
+    assert (await store.get_trace(planner_trace.trace_id)).status == "completed"
+    assert (await store.get_trace(worker_trace.trace_id)).status == "completed"
+    assert coordinator.ensure_calls == []
+
+
+@pytest.mark.asyncio
+async def test_planner_iteration_exhaustion_is_incomplete(tmp_path):
+    preset = "test_completion_gate_exhaustion"
+    _register_role_preset(preset, AgentRole.PLANNER, 2)
+    coordinator = StubCoordinator([{
+        "status": "waiting_children",
+        "root_task_id": "root-task",
+        "pending_tasks": ["child"],
+    }])
+
+    async def llm_call(**_kwargs):
+        return {"content": "", "tool_calls": None, "finish_reason": "stop"}
+
+    runner = AgentRunner(
+        trace_store=FileSystemTraceStore(str(tmp_path)),
+        llm_call=llm_call,
+        task_coordinator=coordinator,
+    )
+    result = await runner.run_result(
+        [{"role": "user", "content": "run mission"}],
+        _explicit_config(preset, 2),
+    )
+
+    assert result["status"] == "incomplete"
+    assert result["error"] == "iteration_limit_reached"
+    assert result["summary"] is None
+
+
+@pytest.mark.asyncio
+async def test_blocked_root_stops_planner_as_incomplete(tmp_path):
+    preset = "test_completion_gate_blocked"
+    _register_role_preset(preset, AgentRole.PLANNER, 3)
+    coordinator = StubCoordinator([{
+        "status": "blocked",
+        "root_task_id": "root-task",
+        "blocked_reason": "external dependency",
+        "pending_tasks": [],
+    }])
+    calls = 0
+
+    async def llm_call(**_kwargs):
+        nonlocal calls
+        calls += 1
+        return {"content": "blocked", "tool_calls": None, "finish_reason": "stop"}
+
+    runner = AgentRunner(
+        trace_store=FileSystemTraceStore(str(tmp_path)),
+        llm_call=llm_call,
+        task_coordinator=coordinator,
+    )
+    result = await runner.run_result(
+        [{"role": "user", "content": "run mission"}],
+        _explicit_config(preset, 3),
+    )
+
+    assert calls == 1
+    assert result["status"] == "incomplete"
+    assert result["error"] == "mission_blocked"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("role", [AgentRole.WORKER, AgentRole.VALIDATOR])
+async def test_explicit_subtrace_without_terminal_submit_is_protocol_failure(
+    tmp_path, role
+):
+    preset = "test_completion_gate_" + role.value
+    _register_role_preset(preset, role, 2)
+
+    async def llm_call(**_kwargs):
+        return {"content": "done", "tool_calls": None, "finish_reason": "stop"}
+
+    runner = AgentRunner(
+        trace_store=FileSystemTraceStore(str(tmp_path / role.value)),
+        llm_call=llm_call,
+        task_coordinator=StubCoordinator([{"status": "completed"}]),
+    )
+    result = await runner.run_result(
+        [{"role": "user", "content": "sub task"}],
+        _explicit_config(preset, 2),
+    )
+
+    assert result["status"] == "failed"
+    assert result["error"] == "protocol_violation"

agent/tests/test_task_spec_invariants.py

@@ -0,0 +1,164 @@
+from dataclasses import asdict
+import json
+
+import pytest
+
+from agent.orchestration.models import (
+    AcceptanceCriterion,
+    TaskSpec,
+    json_values,
+)
+from agent.orchestration.wire import TaskSpecView
+
+
+def criterion(criterion_id="criterion-1", description="The result is correct", hard=True):
+    return AcceptanceCriterion(criterion_id, description, hard)
+
+
+def test_acceptance_criterion_normalizes_text_and_requires_boolean_hard():
+    value = AcceptanceCriterion("  criterion-1  ", "  The result is correct  ")
+
+    assert value.criterion_id == "criterion-1"
+    assert value.description == "The result is correct"
+    assert value.hard is True
+
+    for criterion_id in ("", "   ", None, 1):
+        with pytest.raises(ValueError, match="criterion_id"):
+            AcceptanceCriterion(criterion_id, "description")
+    for description in ("", "   ", None, 1):
+        with pytest.raises(ValueError, match="description"):
+            AcceptanceCriterion("criterion-1", description)
+    for hard in (0, 1, "true", None):
+        with pytest.raises(ValueError, match="hard"):
+            AcceptanceCriterion("criterion-1", "description", hard)
+
+
+def test_acceptance_criterion_from_dict_generates_only_missing_id():
+    generated = AcceptanceCriterion.from_dict({"description": "description"})
+
+    assert generated.criterion_id
+    assert AcceptanceCriterion.from_dict(
+        {"id": " legacy-id ", "description": "description"}
+    ).criterion_id == "legacy-id"
+    for explicit_id in ("", "   ", None):
+        with pytest.raises(ValueError, match="criterion_id"):
+            AcceptanceCriterion.from_dict(
+                {"criterion_id": explicit_id, "description": "description"}
+            )
+
+
+@pytest.mark.parametrize("version", [True, False, 0, -1, 1.0, "1", None])
+def test_task_spec_requires_a_positive_integer_version(version):
+    with pytest.raises(ValueError, match="positive integer"):
+        TaskSpec(version, "objective", [criterion()])
+
+
+@pytest.mark.parametrize("objective", ["", "   ", None, 1])
+def test_task_spec_requires_a_non_empty_string_objective(objective):
+    with pytest.raises(ValueError, match="objective"):
+        TaskSpec(1, objective, [criterion()])
+
+
+def test_task_spec_requires_criteria_with_unique_normalized_ids():
+    with pytest.raises(ValueError, match="at least one"):
+        TaskSpec(1, "objective")
+    with pytest.raises(ValueError, match="unique"):
+        TaskSpec(
+            1,
+            "objective",
+            [criterion(" duplicate "), criterion("duplicate", "another check")],
+        )
+
+    value = TaskSpec(
+        1,
+        "objective",
+        [criterion("case"), criterion("CASE", "another check")],
+    )
+    assert [item.criterion_id for item in value.acceptance_criteria] == ["case", "CASE"]
+
+
+def test_task_spec_freezes_collections_and_does_not_track_source_mutation():
+    criteria = [criterion()]
+    context_refs = ["context-1"]
+    value = TaskSpec(1, "  objective  ", criteria, context_refs)
+
+    criteria.append(criterion("criterion-2", "another check"))
+    context_refs.append("context-2")
+
+    assert value.objective == "objective"
+    assert isinstance(value.acceptance_criteria, tuple)
+    assert isinstance(value.context_refs, tuple)
+    assert len(value.acceptance_criteria) == 1
+    assert value.context_refs == ("context-1",)
+
+
+@pytest.mark.parametrize(
+    "context_refs",
+    ["context", {"context": True}, None, [""], ["   "], [1], [[]]],
+)
+def test_task_spec_requires_immutable_string_context_refs(context_refs):
+    with pytest.raises(ValueError, match="context_refs|collections"):
+        TaskSpec(1, "objective", [criterion()], context_refs)
+    with pytest.raises(ValueError, match="context_refs|collections"):
+        TaskSpec.from_dict({
+            "objective": "objective",
+            "acceptance_criteria": [{"description": "description"}],
+            "context_refs": context_refs,
+        })
+
+    value = TaskSpec(1, "objective", [criterion()], ["  context-1  "])
+    assert value.context_refs == ("context-1",)
+
+
+def test_task_spec_from_dict_is_strict_and_generates_stable_id_for_instance():
+    value = TaskSpec.from_dict(
+        {
+            "objective": "objective",
+            "acceptance_criteria": [{"description": "description"}],
+        }
+    )
+
+    assert value.version == 1
+    assert value.acceptance_criteria[0].criterion_id
+    restored = TaskSpec.from_dict(json.loads(json.dumps(json_values(asdict(value)))))
+    assert restored.acceptance_criteria[0].criterion_id == value.acceptance_criteria[0].criterion_id
+    with pytest.raises(ValueError, match="positive integer"):
+        TaskSpec.from_dict(
+            {
+                "version": "1",
+                "objective": "objective",
+                "acceptance_criteria": [{"description": "description"}],
+            }
+        )
+
+
+@pytest.mark.parametrize("criteria", [None, "criterion", {"description": "description"}])
+def test_task_spec_from_dict_rejects_invalid_criteria_collections(criteria):
+    with pytest.raises(ValueError, match="acceptance_criteria"):
+        TaskSpec.from_dict({
+            "objective": "objective",
+            "acceptance_criteria": criteria,
+        })
+
+
+def test_task_spec_wire_shape_keeps_json_arrays():
+    value = TaskSpec(
+        1,
+        "objective",
+        [criterion()],
+        ["context-1"],
+        created_at="2026-01-01T00:00:00+00:00",
+    )
+
+    payload = json_values(asdict(value))
+    wire = TaskSpecView.model_validate(payload).model_dump(mode="json")
+
+    assert wire["acceptance_criteria"] == [
+        {
+            "criterion_id": "criterion-1",
+            "description": "The result is correct",
+            "hard": True,
+        }
+    ]
+    assert wire["context_refs"] == ["context-1"]
+    assert isinstance(json.loads(json.dumps(payload))["acceptance_criteria"], list)