script-build-and-framework-tmp/agent/tests/test_orchestration_edges.py

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", goal_id=None, 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")],
        )