from types import SimpleNamespace import pytest from agent.orchestration.config import OrchestrationConfig from agent.orchestration.coordinator import OrchestrationError, TaskConflict from agent.orchestration.executor import LocalAgentExecutor from agent.orchestration.models import ( AcceptanceCriterion, CriterionResult, DecisionAction, TaskRecord, TaskSpec, ValidationVerdict, ) from test_coordinator_integration import FakeExecutor, create_task, make_coordinator def test_orchestration_config_rejects_invalid_limits(): with pytest.raises(ValueError, match="at least"): OrchestrationConfig(max_parallel_tasks=0) with pytest.raises(ValueError, match="negative"): OrchestrationConfig(max_repair_continuations=-1) @pytest.mark.asyncio async def test_local_executor_rejects_remote_v1_and_normalizes_runner_errors(): class BrokenRunner: async def run_result(self, **kwargs): raise RuntimeError("boom") executor = LocalAgentExecutor(BrokenRunner()) worker = await executor.run_worker({"worker_preset": "remote_worker", "worker_trace_id": "w"}) validator = await executor.run_validator({"validator_preset": "remote_validator", "validator_trace_id": "v"}) assert worker.status == "failed" and "local" in worker.error.lower() assert validator.status == "failed" and "local" in validator.error.lower() worker = await executor.run_worker({ "worker_preset": "worker", "worker_trace_id": "w", "root_trace_id": "root", "task_id": "task", "spec_version": 1, "attempt_id": "attempt", "task_spec": {}, "continue_trace_id": None, }) validator = await executor.run_validator({ "validator_preset": "validator", "validator_trace_id": "v", "root_trace_id": "root", "task_id": "task", "spec_version": 1, "attempt_id": "attempt", "snapshot_id": "snapshot", "validation_id": "validation", "task_spec": {}, "artifact_snapshot": {}, }) assert worker.error == "boom" assert validator.error == "boom" @pytest.mark.asyncio async def test_local_executor_repair_continuation_is_coordinator_validated(): class Coordinator: called = None async def validate_continue_from(self, root, task, attempt): self.called = (root, task, attempt) return "worker-trace" class Store: updated = None async def get_trace(self, trace_id): return SimpleNamespace(context={"old": True}) async def update_trace(self, trace_id, **updates): self.updated = (trace_id, updates) class Runner: def __init__(self): self.task_coordinator = Coordinator() self.trace_store = Store() async def run_result(self, **kwargs): return {"trace_id": "worker-trace", "status": "completed", "summary": "repaired"} runner = Runner() result = await LocalAgentExecutor(runner).run_worker({ "worker_preset": "worker", "worker_trace_id": "worker-trace", "root_trace_id": "root", "task_id": "task", "spec_version": 1, "attempt_id": "new-attempt", "prior_attempt_id": "old-attempt", "task_spec": {}, "continue_trace_id": "worker-trace", }) assert result.status == "completed" assert runner.task_coordinator.called == ("root", "task", "old-attempt") assert runner.trace_store.updated[1]["context"]["attempt_id"] == "new-attempt" @pytest.mark.asyncio async def test_planning_and_dispatch_input_guards(tmp_path): executor = FakeExecutor([]) coordinator, _, _ = await make_coordinator(tmp_path, executor) with pytest.raises(ValueError, match="At least"): await coordinator.create_tasks("root", []) with pytest.raises(ValueError, match="objective"): await coordinator.create_tasks("root", [{"objective": ""}]) with pytest.raises(ValueError, match="Parent"): await coordinator.create_tasks("root", [{"objective": "x"}], parent_task_id="missing") assert await coordinator.dispatch_tasks("root", []) == [] task_id = await create_task(coordinator) with pytest.raises(TaskConflict, match="twice"): await coordinator.dispatch_tasks("root", [task_id, task_id]) with pytest.raises(ValueError, match="worker_presets"): await coordinator.dispatch_tasks("root", [task_id], worker_presets=["a", "b"]) focused = await coordinator.focus_task("root", task_id) assert focused["task_id"] == task_id coordinator.executor = None with pytest.raises(RuntimeError, match="Executor"): await coordinator.dispatch_tasks("root", [task_id]) @pytest.mark.asyncio async def test_decision_and_revalidation_guards(tmp_path): executor = FakeExecutor([ValidationVerdict.PASSED]) coordinator, store, _ = await make_coordinator(tmp_path, executor) task_id = await create_task(coordinator) with pytest.raises(ValueError, match="reason"): await coordinator.decide_task("root", task_id, None, DecisionAction.BLOCK, {}, "missing-reason") with pytest.raises(TaskConflict, match="blocked"): await coordinator.decide_task("root", task_id, None, DecisionAction.UNBLOCK, {}, "not-blocked") with pytest.raises(OrchestrationError, match="snapshot"): await coordinator.revalidate_attempt("root", task_id, "missing") cycle = (await coordinator.dispatch_tasks("root", [task_id]))[0] with pytest.raises(TaskConflict, match="(?i)passed"): await coordinator.decide_task( "root", task_id, cycle.validation_id, DecisionAction.REPAIR, {"reason": "should accept"}, "bad-repair", ) await coordinator.decide_task( "root", task_id, cycle.validation_id, DecisionAction.ACCEPT, {"reason": "passed"}, "accept", ) with pytest.raises(TaskConflict, match="(?i)terminal"): await coordinator.decide_task( "root", task_id, None, DecisionAction.BLOCK, {"reason": "too late"}, "late-block", ) with pytest.raises(TaskConflict, match="already terminal"): await coordinator.decide_task( "root", task_id, None, DecisionAction.CANCEL, {"reason": "too late"}, "late-cancel", ) def test_validation_report_requires_every_hard_criterion(): task = TaskRecord( task_id="t", parent_task_id=None, display_path="1", specs=[TaskSpec( version=1, objective="o", acceptance_criteria=[AcceptanceCriterion("c1", "required", hard=True)], )], ) from agent.orchestration.coordinator import TaskCoordinator with pytest.raises(ValueError, match="missing"): TaskCoordinator._validate_report(task, ValidationVerdict.PASSED, []) with pytest.raises(ValueError, match="conflicts"): TaskCoordinator._validate_report( task, ValidationVerdict.PASSED, [CriterionResult("c1", ValidationVerdict.FAILED, "no")], )