feat(api): add orchestration v2 api and sdk

agent/agent/__init__.py

@@ -33,7 +33,17 @@ from agent.orchestration import (
     AgentRole,
     TaskStatus,
     ValidationVerdict,
+    ExecutionStats,
+    FailureCode,
+    BackgroundOperation,
+    OperationKind,
+    OperationStatus,
+    OrchestrationAPIError,
+    OrchestrationClient,
+    ValidationMode,
+    ValidationPlan,
     TaskCoordinator,
+    AgentExecutor,
     TaskStore,
     FileSystemTaskStore,
     ArtifactStore,
@@ -83,7 +93,17 @@ __all__ = [
     "AgentRole",
     "TaskStatus",
     "ValidationVerdict",
+    "ExecutionStats",
+    "FailureCode",
+    "BackgroundOperation",
+    "OperationKind",
+    "OperationStatus",
+    "OrchestrationAPIError",
+    "OrchestrationClient",
+    "ValidationMode",
+    "ValidationPlan",
     "TaskCoordinator",
+    "AgentExecutor",
     "TaskStore",
     "FileSystemTaskStore",
     "ArtifactStore",

agent/agent/orchestration/__init__.py

@@ -1,6 +1,21 @@
-"""Explicit task execution and independent validation."""
+"""Explicit task execution, validation, recovery, and V2 client ports."""
 
+from .api_v2 import create_orchestration_router
+from .client_v2 import OrchestrationAPIError, OrchestrationClient
 from .config import OrchestrationConfig
+from .coordinator import TaskCoordinator
+from .errors import OrchestrationError, TaskConflict
+from .evidence import (
+    EvidenceBudgetExceeded,
+    EvidenceError,
+    EvidenceOwnershipError,
+    EvidencePort,
+    EvidenceProvider,
+    EvidenceProviderError,
+    EvidenceProviderResult,
+    EvidenceQuery,
+    EvidenceResponse,
+)
 from .models import (
     AcceptanceCriterion,
     AgentRole,
@@ -8,9 +23,17 @@ from .models import (
     ArtifactSnapshot,
     AttemptStatus,
     AttemptSubmission,
+    BackgroundOperation,
+    CommandRecord,
     CompletionPolicy,
     CriterionResult,
     DecisionAction,
+    EventPage,
+    ExecutionStats,
+    FailureCode,
+    OperationKind,
+    OperationStatus,
+    OrchestrationEvent,
     PlannerDecision,
     TaskAttempt,
     TaskCycleResult,
@@ -18,21 +41,99 @@ from .models import (
     TaskRecord,
     TaskSpec,
     TaskStatus,
+    ValidationMode,
+    ValidationPlan,
     ValidationReport,
     ValidationRunStatus,
     ValidationVerdict,
 )
-from .protocols import ArtifactStore, TaskStore
+from .protocols import AgentExecutor, ArtifactStore, EventSink, TaskStore
 from .state_machine import InvalidTaskTransition
-from .coordinator import TaskCoordinator
-from .store import FileSystemTaskStore, FileSystemArtifactStore
+from .store import (
+    ArtifactConflict,
+    FileSystemArtifactStore,
+    FileSystemTaskStore,
+    RevisionConflict,
+    TaskStoreError,
+    TaskStoreNotFound,
+)
+from .validation_policy import (
+    DefaultValidationPolicy,
+    DeterministicRule,
+    DeterministicRuleResult,
+    DeterministicValidationResult,
+    DeterministicValidator,
+    RuleBasedDeterministicValidator,
+    ValidationContext,
+    ValidationPolicy,
+)
+from .wire import OperationView, ProblemDetails
 
 __all__ = [
-    "OrchestrationConfig", "TaskStore", "ArtifactStore", "InvalidTaskTransition",
-    "TaskCoordinator", "FileSystemTaskStore", "FileSystemArtifactStore",
-    "CompletionPolicy", "AgentRole", "TaskStatus", "AttemptStatus",
-    "ValidationRunStatus", "ValidationVerdict", "DecisionAction",
-    "AcceptanceCriterion", "TaskSpec", "TaskRecord", "TaskAttempt",
-    "ArtifactRef", "ArtifactSnapshot", "AttemptSubmission", "CriterionResult",
-    "ValidationReport", "PlannerDecision", "TaskLedger", "TaskCycleResult",
+    "AcceptanceCriterion",
+    "AgentExecutor",
+    "AgentRole",
+    "ArtifactConflict",
+    "ArtifactRef",
+    "ArtifactSnapshot",
+    "ArtifactStore",
+    "AttemptStatus",
+    "AttemptSubmission",
+    "BackgroundOperation",
+    "CommandRecord",
+    "CompletionPolicy",
+    "CriterionResult",
+    "DecisionAction",
+    "DefaultValidationPolicy",
+    "DeterministicRule",
+    "DeterministicRuleResult",
+    "DeterministicValidationResult",
+    "DeterministicValidator",
+    "EventPage",
+    "EventSink",
+    "EvidenceBudgetExceeded",
+    "EvidenceError",
+    "EvidenceOwnershipError",
+    "EvidencePort",
+    "EvidenceProvider",
+    "EvidenceProviderError",
+    "EvidenceProviderResult",
+    "EvidenceQuery",
+    "EvidenceResponse",
+    "ExecutionStats",
+    "FailureCode",
+    "FileSystemArtifactStore",
+    "FileSystemTaskStore",
+    "InvalidTaskTransition",
+    "OperationKind",
+    "OperationStatus",
+    "OperationView",
+    "OrchestrationAPIError",
+    "OrchestrationClient",
+    "OrchestrationConfig",
+    "OrchestrationError",
+    "OrchestrationEvent",
+    "PlannerDecision",
+    "ProblemDetails",
+    "RevisionConflict",
+    "RuleBasedDeterministicValidator",
+    "TaskAttempt",
+    "TaskConflict",
+    "TaskCoordinator",
+    "TaskCycleResult",
+    "TaskLedger",
+    "TaskRecord",
+    "TaskSpec",
+    "TaskStatus",
+    "TaskStore",
+    "TaskStoreError",
+    "TaskStoreNotFound",
+    "ValidationContext",
+    "ValidationMode",
+    "ValidationPlan",
+    "ValidationPolicy",
+    "ValidationReport",
+    "ValidationRunStatus",
+    "ValidationVerdict",
+    "create_orchestration_router",
 ]

agent/agent/orchestration/api_v2.py

@@ -0,0 +1,396 @@
+"""Optional FastAPI adapter for orchestration V2.
+
+FastAPI is imported only when :func:`create_orchestration_router` is called,
+so importing the core orchestration package does not require server extras.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any, Awaitable, Callable, Optional
+
+from pydantic import ValidationError
+
+from .coordinator import OrchestrationError, TaskConflict, TaskCoordinator
+from .store import (
+    ArtifactConflict,
+    RevisionConflict,
+    TaskStoreError,
+    TaskStoreNotFound,
+)
+from .wire import (
+    AttemptList,
+    AttemptView,
+    DispatchOperationRequest,
+    EventPageView,
+    EventView,
+    OperationRequest,
+    OperationView,
+    ProblemDetails,
+    TaskList,
+    TaskView,
+    ValidationList,
+    ValidationView,
+)
+
+
+_ERROR_RESPONSES = {
+    status: {"model": ProblemDetails}
+    for status in (404, 409, 422, 503)
+}
+
+
+class _ResourceNotFound(LookupError):
+    pass
+
+
+def create_orchestration_router(coordinator: TaskCoordinator) -> Any:
+    """Build an isolated router backed by the supplied coordinator instance."""
+
+    try:
+        from fastapi import APIRouter, Header
+        from fastapi.exceptions import RequestValidationError
+        from fastapi.responses import JSONResponse
+        from fastapi.routing import APIRoute
+    except ImportError as exc:  # pragma: no cover - depends on optional extra
+        raise RuntimeError(
+            "FastAPI is optional; install cyber-agent[server] to create the V2 router"
+        ) from exc
+
+    class ProblemRoute(APIRoute):
+        def get_route_handler(self) -> Callable[..., Awaitable[Any]]:
+            original = super().get_route_handler()
+
+            async def route_handler(request: Any) -> Any:
+                try:
+                    return await original(request)
+                except RequestValidationError as exc:
+                    value = ProblemDetails(
+                        title="Invalid Request",
+                        status=422,
+                        detail=str(exc),
+                        code="invalid_request",
+                        instance=request.url.path,
+                    )
+                    return JSONResponse(
+                        status_code=422, content=value.model_dump(mode="json")
+                    )
+
+            return route_handler
+
+    router = APIRouter(
+        prefix="/api/v2", tags=["orchestration-v2"], route_class=ProblemRoute
+    )
+
+    def problem(exc: Exception, instance: str) -> JSONResponse:
+        status, title, code, detail = _problem_fields(exc)
+        value = ProblemDetails(
+            title=title,
+            status=status,
+            detail=detail,
+            code=code,
+            instance=instance,
+        )
+        return JSONResponse(status_code=status, content=value.model_dump(mode="json"))
+
+    async def call(
+        action: Callable[[], Awaitable[Any]],
+        instance: str,
+    ) -> Any:
+        try:
+            return await action()
+        except Exception as exc:  # endpoint boundary owns stable error mapping
+            return problem(exc, instance)
+
+    @router.post(
+        "/roots/{root_trace_id}/operations",
+        status_code=202,
+        response_model=OperationView,
+        responses=_ERROR_RESPONSES,
+    )
+    async def start_operation(
+        root_trace_id: str,
+        body: OperationRequest,
+        idempotency_key: Optional[str] = Header(
+            None,
+            alias="Idempotency-Key",
+            min_length=1,
+            max_length=200,
+            pattern=r"^[\x21-\x7e]+$",
+        ),
+    ) -> Any:
+        path = f"/api/v2/roots/{root_trace_id}/operations"
+
+        async def action() -> OperationView:
+            if isinstance(body, DispatchOperationRequest):
+                request = body
+                operation = await coordinator.start_operation(
+                    root_trace_id,
+                    request.kind,
+                    task_ids=request.task_ids,
+                    worker_presets=request.worker_presets,
+                    deadline_at=request.deadline_at,
+                    idempotency_key=idempotency_key,
+                )
+            else:
+                request = body
+                operation = await coordinator.start_operation(
+                    root_trace_id,
+                    request.kind,
+                    task_id=request.task_id,
+                    attempt_id=request.attempt_id,
+                    deadline_at=request.deadline_at,
+                    idempotency_key=idempotency_key,
+                )
+            return OperationView.from_domain(operation)
+
+        return await call(action, path)
+
+    @router.get(
+        "/roots/{root_trace_id}/operations/{operation_id}",
+        response_model=OperationView,
+        responses=_ERROR_RESPONSES,
+    )
+    async def get_operation(root_trace_id: str, operation_id: str) -> Any:
+        path = f"/api/v2/roots/{root_trace_id}/operations/{operation_id}"
+
+        async def action() -> OperationView:
+            operation = await coordinator.get_operation(root_trace_id, operation_id)
+            return OperationView.from_domain(operation)
+
+        return await call(action, path)
+
+    @router.post(
+        "/roots/{root_trace_id}/operations/{operation_id}/stop",
+        response_model=OperationView,
+        responses=_ERROR_RESPONSES,
+    )
+    async def stop_operation(
+        root_trace_id: str,
+        operation_id: str,
+        idempotency_key: Optional[str] = Header(
+            None,
+            alias="Idempotency-Key",
+            min_length=1,
+            max_length=200,
+            pattern=r"^[\x21-\x7e]+$",
+        ),
+    ) -> Any:
+        path = f"/api/v2/roots/{root_trace_id}/operations/{operation_id}/stop"
+
+        async def action() -> OperationView:
+            operation = await coordinator.stop_operation(
+                root_trace_id, operation_id, idempotency_key=idempotency_key
+            )
+            return OperationView.from_domain(operation)
+
+        return await call(action, path)
+
+    @router.post(
+        "/roots/{root_trace_id}/operations/{operation_id}/resume",
+        response_model=OperationView,
+        responses=_ERROR_RESPONSES,
+    )
+    async def resume_operation(
+        root_trace_id: str,
+        operation_id: str,
+        idempotency_key: Optional[str] = Header(
+            None,
+            alias="Idempotency-Key",
+            min_length=1,
+            max_length=200,
+            pattern=r"^[\x21-\x7e]+$",
+        ),
+    ) -> Any:
+        path = f"/api/v2/roots/{root_trace_id}/operations/{operation_id}/resume"
+
+        async def action() -> OperationView:
+            operation = await coordinator.resume_operation(
+                root_trace_id, operation_id, idempotency_key=idempotency_key
+            )
+            return OperationView.from_domain(operation)
+
+        return await call(action, path)
+
+    @router.get(
+        "/roots/{root_trace_id}/tasks",
+        response_model=TaskList,
+        responses=_ERROR_RESPONSES,
+    )
+    async def list_tasks(root_trace_id: str) -> Any:
+        path = f"/api/v2/roots/{root_trace_id}/tasks"
+
+        async def action() -> TaskList:
+            ledger = await coordinator.task_store.load(root_trace_id)
+            return TaskList(
+                revision=ledger.revision,
+                items=[
+                    TaskView.from_domain(item)
+                    for item in sorted(
+                        ledger.tasks.values(),
+                        key=lambda value: (value.created_at, value.task_id),
+                    )
+                ],
+            )
+
+        return await call(action, path)
+
+    @router.get(
+        "/roots/{root_trace_id}/tasks/{task_id}",
+        response_model=TaskView,
+        responses=_ERROR_RESPONSES,
+    )
+    async def get_task(root_trace_id: str, task_id: str) -> Any:
+        path = f"/api/v2/roots/{root_trace_id}/tasks/{task_id}"
+
+        async def action() -> TaskView:
+            ledger = await coordinator.task_store.load(root_trace_id)
+            item = ledger.tasks.get(task_id)
+            if item is None:
+                raise _ResourceNotFound(f"Task not found: {task_id}")
+            return TaskView.from_domain(item)
+
+        return await call(action, path)
+
+    @router.get(
+        "/roots/{root_trace_id}/attempts",
+        response_model=AttemptList,
+        responses=_ERROR_RESPONSES,
+    )
+    async def list_attempts(root_trace_id: str) -> Any:
+        path = f"/api/v2/roots/{root_trace_id}/attempts"
+
+        async def action() -> AttemptList:
+            ledger = await coordinator.task_store.load(root_trace_id)
+            return AttemptList(
+                revision=ledger.revision,
+                items=[
+                    AttemptView.from_domain(item)
+                    for item in sorted(
+                        ledger.attempts.values(),
+                        key=lambda value: (value.created_at, value.attempt_id),
+                    )
+                ],
+            )
+
+        return await call(action, path)
+
+    @router.get(
+        "/roots/{root_trace_id}/attempts/{attempt_id}",
+        response_model=AttemptView,
+        responses=_ERROR_RESPONSES,
+    )
+    async def get_attempt(root_trace_id: str, attempt_id: str) -> Any:
+        path = f"/api/v2/roots/{root_trace_id}/attempts/{attempt_id}"
+
+        async def action() -> AttemptView:
+            ledger = await coordinator.task_store.load(root_trace_id)
+            item = ledger.attempts.get(attempt_id)
+            if item is None:
+                raise _ResourceNotFound(f"Attempt not found: {attempt_id}")
+            return AttemptView.from_domain(item)
+
+        return await call(action, path)
+
+    @router.get(
+        "/roots/{root_trace_id}/validations",
+        response_model=ValidationList,
+        responses=_ERROR_RESPONSES,
+    )
+    async def list_validations(root_trace_id: str) -> Any:
+        path = f"/api/v2/roots/{root_trace_id}/validations"
+
+        async def action() -> ValidationList:
+            ledger = await coordinator.task_store.load(root_trace_id)
+            return ValidationList(
+                revision=ledger.revision,
+                items=[
+                    ValidationView.from_domain(item)
+                    for item in sorted(
+                        ledger.validations.values(),
+                        key=lambda value: (value.created_at, value.validation_id),
+                    )
+                ],
+            )
+
+        return await call(action, path)
+
+    @router.get(
+        "/roots/{root_trace_id}/validations/{validation_id}",
+        response_model=ValidationView,
+        responses=_ERROR_RESPONSES,
+    )
+    async def get_validation(root_trace_id: str, validation_id: str) -> Any:
+        path = f"/api/v2/roots/{root_trace_id}/validations/{validation_id}"
+
+        async def action() -> ValidationView:
+            ledger = await coordinator.task_store.load(root_trace_id)
+            item = ledger.validations.get(validation_id)
+            if item is None:
+                raise _ResourceNotFound(f"Validation not found: {validation_id}")
+            return ValidationView.from_domain(item)
+
+        return await call(action, path)
+
+    @router.get(
+        "/roots/{root_trace_id}/events",
+        response_model=EventPageView,
+        responses=_ERROR_RESPONSES,
+    )
+    async def list_events(
+        root_trace_id: str,
+        cursor: Optional[str] = None,
+        limit: str = "100",
+        wait_seconds: str = "0",
+    ) -> Any:
+        path = f"/api/v2/roots/{root_trace_id}/events"
+
+        async def action() -> EventPageView:
+            try:
+                parsed_limit = int(limit)
+                parsed_wait = float(wait_seconds)
+            except ValueError as exc:
+                raise ValueError("limit and wait_seconds must be numeric") from exc
+            if not 1 <= parsed_limit <= 1000:
+                raise ValueError("limit must be between 1 and 1000")
+            if not 0 <= parsed_wait <= 30:
+                raise ValueError("wait_seconds must be between 0 and 30")
+            deadline = asyncio.get_running_loop().time() + parsed_wait
+            while True:
+                page = await coordinator.task_store.list_events(
+                    root_trace_id, cursor=cursor, limit=parsed_limit
+                )
+                if page.events or asyncio.get_running_loop().time() >= deadline:
+                    return EventPageView(
+                        events=[EventView.from_domain(item) for item in page.events],
+                        next_cursor=page.next_cursor,
+                        has_more=page.has_more,
+                    )
+                await asyncio.sleep(min(0.1, max(deadline - asyncio.get_running_loop().time(), 0)))
+
+        return await call(action, path)
+
+    return router
+
+
+def _problem_fields(exc: Exception) -> tuple[int, str, str, str]:
+    if isinstance(exc, (TaskStoreNotFound, _ResourceNotFound)):
+        return 404, "Not Found", "not_found", str(exc)
+    if isinstance(exc, ValueError) and "not found" in str(exc).lower():
+        return 404, "Not Found", "not_found", str(exc)
+    if isinstance(exc, (TaskConflict, RevisionConflict, ArtifactConflict)):
+        return 409, "Conflict", "conflict", str(exc)
+    if isinstance(exc, ValidationError):
+        return 422, "Invalid Request", "invalid_request", str(exc)
+    if isinstance(exc, ValueError):
+        return 422, "Invalid Request", "invalid_request", str(exc)
+    if isinstance(exc, RuntimeError) and "AgentExecutor" in str(exc):
+        return 503, "Service Unavailable", "service_unavailable", str(exc)
+    if isinstance(exc, TaskStoreError):
+        return 503, "Service Unavailable", "storage_unavailable", str(exc)
+    if isinstance(exc, OrchestrationError):
+        return 409, "Conflict", "orchestration_conflict", str(exc)
+    return 500, "Internal Server Error", "internal_error", "Internal server error"
+
+
+__all__ = ["create_orchestration_router"]

agent/agent/orchestration/client_v2.py

@@ -0,0 +1,270 @@
+"""Async Python client for the orchestration V2 HTTP API."""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any, AsyncIterator, Dict, List, Optional
+from urllib.parse import quote
+
+import httpx
+
+from .wire import (
+    AttemptList,
+    AttemptView,
+    EventPageView,
+    EventView,
+    OperationView,
+    ProblemDetails,
+    TaskList,
+    TaskView,
+    ValidationList,
+    ValidationView,
+)
+
+
+class OrchestrationAPIError(RuntimeError):
+    def __init__(self, problem: ProblemDetails) -> None:
+        super().__init__(problem.detail)
+        self.problem = problem
+        self.status_code = problem.status
+        self.code = problem.code
+
+
+class OrchestrationClient:
+    """Small async-first client with safe retry rules for mutations."""
+
+    _TRANSIENT_STATUSES = {502, 503, 504}
+
+    def __init__(
+        self,
+        base_url: str,
+        *,
+        client: Optional[httpx.AsyncClient] = None,
+        timeout: float = 30.0,
+        max_retries: int = 2,
+    ) -> None:
+        self._owns_client = client is None
+        self._client = client or httpx.AsyncClient(
+            base_url=base_url.rstrip("/"), timeout=timeout
+        )
+        self.max_retries = max(0, max_retries)
+
+    async def __aenter__(self) -> "OrchestrationClient":
+        return self
+
+    async def __aexit__(self, *_: Any) -> None:
+        await self.aclose()
+
+    async def aclose(self) -> None:
+        if self._owns_client:
+            await self._client.aclose()
+
+    async def start_dispatch(
+        self,
+        root_trace_id: str,
+        task_ids: List[str],
+        *,
+        worker_presets: Optional[List[str]] = None,
+        deadline_at: Optional[str] = None,
+        idempotency_key: Optional[str] = None,
+    ) -> OperationView:
+        body: Dict[str, Any] = {"kind": "dispatch", "task_ids": task_ids}
+        if worker_presets is not None:
+            body["worker_presets"] = worker_presets
+        if deadline_at is not None:
+            body["deadline_at"] = deadline_at
+        data = await self._request(
+            "POST",
+            f"{self._root(root_trace_id)}/operations",
+            json=body,
+            idempotency_key=idempotency_key,
+        )
+        return OperationView.model_validate(data)
+
+    async def start_revalidation(
+        self,
+        root_trace_id: str,
+        task_id: str,
+        attempt_id: str,
+        *,
+        deadline_at: Optional[str] = None,
+        idempotency_key: Optional[str] = None,
+    ) -> OperationView:
+        body: Dict[str, Any] = {
+            "kind": "revalidate",
+            "task_id": task_id,
+            "attempt_id": attempt_id,
+        }
+        if deadline_at is not None:
+            body["deadline_at"] = deadline_at
+        data = await self._request(
+            "POST",
+            f"{self._root(root_trace_id)}/operations",
+            json=body,
+            idempotency_key=idempotency_key,
+        )
+        return OperationView.model_validate(data)
+
+    async def poll(self, root_trace_id: str, operation_id: str) -> OperationView:
+        data = await self._request(
+            "GET", f"{self._root(root_trace_id)}/operations/{self._part(operation_id)}"
+        )
+        return OperationView.model_validate(data)
+
+    async def stop(
+        self,
+        root_trace_id: str,
+        operation_id: str,
+        *,
+        idempotency_key: Optional[str] = None,
+    ) -> OperationView:
+        data = await self._request(
+            "POST",
+            f"{self._root(root_trace_id)}/operations/{self._part(operation_id)}/stop",
+            idempotency_key=idempotency_key,
+        )
+        return OperationView.model_validate(data)
+
+    async def resume(
+        self,
+        root_trace_id: str,
+        operation_id: str,
+        *,
+        idempotency_key: Optional[str] = None,
+    ) -> OperationView:
+        data = await self._request(
+            "POST",
+            f"{self._root(root_trace_id)}/operations/{self._part(operation_id)}/resume",
+            idempotency_key=idempotency_key,
+        )
+        return OperationView.model_validate(data)
+
+    async def list_tasks(self, root_trace_id: str) -> TaskList:
+        return TaskList.model_validate(
+            await self._request("GET", f"{self._root(root_trace_id)}/tasks")
+        )
+
+    async def get_task(self, root_trace_id: str, task_id: str) -> TaskView:
+        return TaskView.model_validate(
+            await self._request(
+                "GET", f"{self._root(root_trace_id)}/tasks/{self._part(task_id)}"
+            )
+        )
+
+    async def list_attempts(self, root_trace_id: str) -> AttemptList:
+        return AttemptList.model_validate(
+            await self._request("GET", f"{self._root(root_trace_id)}/attempts")
+        )
+
+    async def get_attempt(self, root_trace_id: str, attempt_id: str) -> AttemptView:
+        return AttemptView.model_validate(
+            await self._request(
+                "GET", f"{self._root(root_trace_id)}/attempts/{self._part(attempt_id)}"
+            )
+        )
+
+    async def list_validations(self, root_trace_id: str) -> ValidationList:
+        return ValidationList.model_validate(
+            await self._request("GET", f"{self._root(root_trace_id)}/validations")
+        )
+
+    async def get_validation(
+        self, root_trace_id: str, validation_id: str
+    ) -> ValidationView:
+        return ValidationView.model_validate(
+            await self._request(
+                "GET",
+                f"{self._root(root_trace_id)}/validations/{self._part(validation_id)}",
+            )
+        )
+
+    async def list_events(
+        self,
+        root_trace_id: str,
+        *,
+        cursor: Optional[str] = None,
+        limit: int = 100,
+        wait_seconds: float = 0,
+    ) -> EventPageView:
+        params: Dict[str, Any] = {"limit": limit, "wait_seconds": wait_seconds}
+        if cursor is not None:
+            params["cursor"] = cursor
+        return EventPageView.model_validate(
+            await self._request(
+                "GET", f"{self._root(root_trace_id)}/events", params=params
+            )
+        )
+
+    async def watch_events(
+        self,
+        root_trace_id: str,
+        *,
+        cursor: Optional[str] = None,
+        limit: int = 100,
+        wait_seconds: float = 20,
+    ) -> AsyncIterator[EventView]:
+        next_cursor = cursor
+        while True:
+            page = await self.list_events(
+                root_trace_id,
+                cursor=next_cursor,
+                limit=limit,
+                wait_seconds=wait_seconds,
+            )
+            for event in page.events:
+                yield event
+            next_cursor = page.next_cursor or next_cursor
+            if not page.events:
+                await asyncio.sleep(0.05)
+
+    async def _request(
+        self,
+        method: str,
+        path: str,
+        *,
+        idempotency_key: Optional[str] = None,
+        **kwargs: Any,
+    ) -> Any:
+        headers = dict(kwargs.pop("headers", {}))
+        if idempotency_key:
+            headers["Idempotency-Key"] = idempotency_key
+        method = method.upper()
+        retryable = method in {"GET", "HEAD"} or bool(idempotency_key)
+        attempts = self.max_retries + 1 if retryable else 1
+        for index in range(attempts):
+            try:
+                response = await self._client.request(
+                    method, path, headers=headers, **kwargs
+                )
+            except httpx.TransportError:
+                if index + 1 == attempts:
+                    raise
+                await asyncio.sleep(0.05 * (2**index))
+                continue
+            if response.status_code in self._TRANSIENT_STATUSES and index + 1 < attempts:
+                await asyncio.sleep(0.05 * (2**index))
+                continue
+            if response.is_error:
+                try:
+                    problem = ProblemDetails.model_validate(response.json())
+                except Exception:
+                    problem = ProblemDetails(
+                        title="HTTP Error",
+                        status=response.status_code,
+                        detail=response.text or f"HTTP {response.status_code}",
+                        code="http_error",
+                    )
+                raise OrchestrationAPIError(problem)
+            return response.json()
+        raise AssertionError("unreachable")
+
+    @classmethod
+    def _root(cls, root_trace_id: str) -> str:
+        return f"/api/v2/roots/{cls._part(root_trace_id)}"
+
+    @staticmethod
+    def _part(value: str) -> str:
+        return quote(value, safe="")
+
+
+__all__ = ["OrchestrationAPIError", "OrchestrationClient"]

agent/agent/orchestration/wire.py

@@ -0,0 +1,278 @@
+"""Stable wire models shared by the V2 HTTP API and Python client."""
+
+from __future__ import annotations
+
+from dataclasses import asdict
+from typing import Annotated, Any, Dict, List, Literal, Optional, Union
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from .models import (
+    BackgroundOperation,
+    AttemptStatus,
+    FailureCode,
+    OperationKind,
+    OperationStatus,
+    OrchestrationEvent,
+    TaskAttempt,
+    TaskRecord,
+    TaskStatus,
+    ValidationReport,
+    ValidationMode,
+    ValidationRunStatus,
+    ValidationVerdict,
+    json_values,
+)
+
+
+class WireModel(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+
+class DispatchOperationRequest(WireModel):
+    kind: Literal["dispatch"] = "dispatch"
+    task_ids: List[Annotated[str, Field(min_length=1)]] = Field(min_length=1)
+    worker_presets: Optional[List[str]] = None
+    deadline_at: Optional[str] = None
+
+
+class RevalidateOperationRequest(WireModel):
+    kind: Literal["revalidate"] = "revalidate"
+    task_id: str = Field(min_length=1)
+    attempt_id: str = Field(min_length=1)
+    deadline_at: Optional[str] = None
+
+
+OperationRequest = Annotated[
+    Union[DispatchOperationRequest, RevalidateOperationRequest],
+    Field(discriminator="kind"),
+]
+
+
+class OperationView(WireModel):
+    operation_id: str
+    root_trace_id: str
+    kind: OperationKind
+    status: OperationStatus
+    task_ids: List[str]
+    attempt_ids: List[str]
+    validation_ids: List[str]
+    result_ref: Optional[Dict[str, Any]] = None
+    execution_epoch: int
+    deadline_at: Optional[str] = None
+    error: Optional[str] = None
+    started_at: Optional[str] = None
+    completed_at: Optional[str] = None
+    created_at: str
+    updated_at: str
+
+    @classmethod
+    def from_domain(cls, value: BackgroundOperation) -> "OperationView":
+        data = json_values(asdict(value))
+        data.pop("request", None)
+        data.pop("request_fingerprint", None)
+        return cls.model_validate(data)
+
+
+class AcceptanceCriterionView(WireModel):
+    criterion_id: str
+    description: str
+    hard: bool
+
+
+class TaskSpecView(WireModel):
+    version: int
+    objective: str
+    acceptance_criteria: List[AcceptanceCriterionView]
+    context_refs: List[str]
+    created_at: str
+
+
+class ArtifactRefView(WireModel):
+    uri: str
+    kind: str
+    version: Optional[str] = None
+    digest: Optional[str] = None
+    summary: Optional[str] = None
+    metadata: Dict[str, Any]
+
+
+class AttemptSubmissionView(WireModel):
+    summary: str
+    artifact_refs: List[ArtifactRefView]
+    evidence_refs: List[ArtifactRefView]
+
+
+class CriterionResultView(WireModel):
+    criterion_id: str
+    verdict: ValidationVerdict
+    reason: str
+    evidence_refs: List[ArtifactRefView]
+
+
+class ValidationPlanView(WireModel):
+    mode: ValidationMode
+    validator_preset: Optional[str] = None
+    rule_ids: List[str]
+    max_evidence_queries: int
+    max_evidence_items_per_query: int
+    evidence_timeout_seconds: float
+
+
+class ExecutionStatsView(WireModel):
+    primary_model: Optional[str] = None
+    total_tokens: Optional[int] = None
+    total_cost: Optional[float] = None
+    failure_code: Optional[FailureCode] = None
+
+
+class TaskView(WireModel):
+    task_id: str
+    goal_id: Optional[str] = None
+    parent_task_id: Optional[str] = None
+    display_path: str
+    specs: List[TaskSpecView]
+    current_spec_version: int
+    status: TaskStatus
+    child_task_ids: List[str]
+    attempt_ids: List[str]
+    validation_ids: List[str]
+    decision_ids: List[str]
+    repair_count_by_version: Dict[str, int]
+    blocked_reason: Optional[str] = None
+    superseded_by: Optional[str] = None
+    created_at: str
+    updated_at: str
+
+    @classmethod
+    def from_domain(cls, value: TaskRecord) -> "TaskView":
+        return cls.model_validate(json_values(asdict(value)))
+
+
+class AttemptView(WireModel):
+    attempt_id: str
+    task_id: str
+    spec_version: int
+    worker_trace_id: str
+    worker_preset: str
+    execution_mode: str
+    status: AttemptStatus
+    operation_id: Optional[str] = None
+    execution_epoch: int
+    continue_from_trace_id: Optional[str] = None
+    snapshot_id: Optional[str] = None
+    submission: Optional[AttemptSubmissionView] = None
+    execution_stats: Optional[ExecutionStatsView] = None
+    error: Optional[str] = None
+    started_at: Optional[str] = None
+    completed_at: Optional[str] = None
+    duration_ms: Optional[int] = None
+    created_at: str
+    updated_at: str
+
+    @classmethod
+    def from_domain(cls, value: TaskAttempt) -> "AttemptView":
+        data = json_values(asdict(value))
+        data["duration_ms"] = value.duration_ms
+        return cls.model_validate(data)
+
+
+class ValidationView(WireModel):
+    validation_id: str
+    task_id: str
+    attempt_id: str
+    spec_version: int
+    snapshot_id: str
+    validator_trace_id: str
+    validator_preset: str
+    validation_plan: ValidationPlanView
+    evidence_queries_used: int
+    status: ValidationRunStatus
+    operation_id: Optional[str] = None
+    execution_epoch: int
+    verdict: Optional[ValidationVerdict] = None
+    criterion_results: List[CriterionResultView]
+    summary: str
+    evidence_refs: List[ArtifactRefView]
+    unverified_claims: List[str]
+    risks: List[str]
+    recommendation: str
+    execution_stats: Optional[ExecutionStatsView] = None
+    error: Optional[str] = None
+    started_at: Optional[str] = None
+    completed_at: Optional[str] = None
+    duration_ms: Optional[int] = None
+    created_at: str
+    updated_at: str
+
+    @classmethod
+    def from_domain(cls, value: ValidationReport) -> "ValidationView":
+        data = json_values(asdict(value))
+        data.pop("evidence_query_results", None)
+        data["duration_ms"] = value.duration_ms
+        return cls.model_validate(data)
+
+
+class TaskList(WireModel):
+    revision: int
+    items: List[TaskView]
+
+
+class AttemptList(WireModel):
+    revision: int
+    items: List[AttemptView]
+
+
+class ValidationList(WireModel):
+    revision: int
+    items: List[ValidationView]
+
+
+class EventView(WireModel):
+    schema_version: int
+    event_id: str
+    sequence: int
+    root_trace_id: str
+    ledger_revision: int
+    event_type: str
+    occurred_at: str
+    payload: Dict[str, Any]
+    command_id: Optional[str] = None
+    operation_id: Optional[str] = None
+
+    @classmethod
+    def from_domain(cls, value: OrchestrationEvent) -> "EventView":
+        return cls.model_validate(json_values(asdict(value)))
+
+
+class EventPageView(WireModel):
+    events: List[EventView]
+    next_cursor: Optional[str] = None
+    has_more: bool = False
+
+
+class ProblemDetails(WireModel):
+    type: str = "about:blank"
+    title: str
+    status: int
+    detail: str
+    code: str
+    instance: Optional[str] = None
+
+
+__all__ = [
+    "AttemptList",
+    "AttemptView",
+    "DispatchOperationRequest",
+    "EventPageView",
+    "EventView",
+    "ExecutionStatsView",
+    "OperationRequest",
+    "OperationView",
+    "ProblemDetails",
+    "RevalidateOperationRequest",
+    "TaskList",
+    "TaskView",
+    "ValidationList",
+    "ValidationView",
+]

agent/tests/test_orchestration_v2_api.py

@@ -0,0 +1,333 @@
+from __future__ import annotations
+
+from dataclasses import replace
+
+import httpx
+import pytest
+
+from agent.orchestration.api_v2 import create_orchestration_router
+from agent.orchestration.client_v2 import OrchestrationAPIError, OrchestrationClient
+from agent.orchestration.coordinator import TaskConflict
+from agent.orchestration.models import (
+    BackgroundOperation,
+    EventPage,
+    OperationKind,
+    OperationStatus,
+    OrchestrationEvent,
+    TaskAttempt,
+    TaskLedger,
+    TaskRecord,
+    TaskSpec,
+    TaskStatus,
+    ValidationReport,
+)
+from agent.orchestration.store import TaskStoreError, TaskStoreNotFound
+
+
+FastAPI = pytest.importorskip("fastapi").FastAPI
+
+
+class StubStore:
+    def __init__(self, ledger: TaskLedger, event: OrchestrationEvent) -> None:
+        self.ledger = ledger
+        self.event = event
+
+    async def load(self, root_trace_id):
+        if root_trace_id != self.ledger.root_trace_id:
+            raise TaskStoreNotFound(f"No task ledger for root trace {root_trace_id}")
+        return self.ledger
+
+    async def list_events(self, root_trace_id, cursor=None, limit=100):
+        await self.load(root_trace_id)
+        if cursor == "bad":
+            raise ValueError("Invalid event cursor")
+        events = [] if cursor == "end" else [self.event]
+        return EventPage(events=events, next_cursor="end", has_more=False)
+
+
+class StubCoordinator:
+    def __init__(self, store, operation):
+        self.task_store = store
+        self.operation = operation
+        self.calls = []
+
+    async def start_operation(self, root_trace_id, kind, **kwargs):
+        self.calls.append(("start", root_trace_id, kind, kwargs))
+        return replace(
+            self.operation,
+            root_trace_id=root_trace_id,
+            kind=OperationKind(kind),
+        )
+
+    async def get_operation(self, root_trace_id, operation_id):
+        if operation_id == "conflict":
+            raise TaskConflict("operation conflict")
+        if operation_id == "unavailable":
+            raise TaskStoreError("storage unavailable")
+        if operation_id != self.operation.operation_id:
+            raise ValueError(f"Operation not found: {operation_id}")
+        return self.operation
+
+    async def stop_operation(self, root_trace_id, operation_id, idempotency_key=None):
+        self.calls.append(("stop", idempotency_key))
+        return replace(self.operation, status=OperationStatus.STOPPED)
+
+    async def resume_operation(self, root_trace_id, operation_id, idempotency_key=None):
+        self.calls.append(("resume", idempotency_key))
+        return replace(self.operation, status=OperationStatus.PENDING)
+
+
+@pytest.fixture
+def api_fixture():
+    task = TaskRecord(
+        task_id="task-1",
+        goal_id=None,
+        parent_task_id=None,
+        display_path="1",
+        specs=[TaskSpec(version=1, objective="test")],
+        status=TaskStatus.PENDING,
+    )
+    attempt = TaskAttempt(
+        attempt_id="attempt-1",
+        task_id=task.task_id,
+        spec_version=1,
+        worker_trace_id="worker-1",
+        worker_preset="worker",
+        execution_mode="new",
+    )
+    validation = ValidationReport(
+        validation_id="validation-1",
+        task_id=task.task_id,
+        attempt_id=attempt.attempt_id,
+        spec_version=1,
+        snapshot_id="snapshot-1",
+        validator_trace_id="validator-1",
+    )
+    operation = BackgroundOperation(
+        operation_id="operation-1",
+        root_trace_id="root",
+        kind=OperationKind.DISPATCH,
+        request={"kind": "dispatch", "task_ids": [task.task_id], "secret": "hidden"},
+        request_fingerprint="private-fingerprint",
+        task_ids=[task.task_id],
+    )
+    ledger = TaskLedger(root_trace_id="root", mission="test", revision=3)
+    ledger.tasks[task.task_id] = task
+    ledger.attempts[attempt.attempt_id] = attempt
+    ledger.validations[validation.validation_id] = validation
+    event = OrchestrationEvent(
+        schema_version=1,
+        event_id="event-1",
+        sequence=1,
+        root_trace_id="root",
+        ledger_revision=3,
+        event_type="task_created",
+        occurred_at="2026-07-18T00:00:00+00:00",
+    )
+    store = StubStore(ledger, event)
+    coordinator = StubCoordinator(store, operation)
+    app = FastAPI()
+    app.include_router(create_orchestration_router(coordinator))
+    transport = httpx.ASGITransport(app=app)
+    http = httpx.AsyncClient(transport=transport, base_url="http://test")
+    return coordinator, http
+
+
+@pytest.mark.asyncio
+async def test_router_start_queries_and_problem_details(api_fixture):
+    coordinator, http = api_fixture
+    response = await http.post(
+        "/api/v2/roots/root/operations",
+        headers={"Idempotency-Key": "dispatch-once"},
+        json={"kind": "dispatch", "task_ids": ["task-1"]},
+    )
+    assert response.status_code == 202
+    assert response.json()["operation_id"] == "operation-1"
+    assert "request" not in response.json()
+    assert "request_fingerprint" not in response.json()
+    assert coordinator.calls[0][-1]["idempotency_key"] == "dispatch-once"
+
+    assert (await http.get("/api/v2/roots/root/tasks")).json()["revision"] == 3
+    assert (await http.get("/api/v2/roots/root/tasks/task-1")).json()["task_id"] == "task-1"
+    assert (await http.get("/api/v2/roots/root/attempts")).json()["items"][0]["attempt_id"] == "attempt-1"
+    assert (await http.get("/api/v2/roots/root/attempts/attempt-1")).json()["attempt_id"] == "attempt-1"
+    assert (await http.get("/api/v2/roots/root/validations")).json()["items"][0]["validation_id"] == "validation-1"
+    assert (
+        await http.get("/api/v2/roots/root/validations/validation-1")
+    ).json()["validation_id"] == "validation-1"
+    assert (await http.get("/api/v2/roots/root/events")).json()["events"][0]["sequence"] == 1
+
+    missing = await http.get("/api/v2/roots/root/tasks/missing")
+    assert missing.status_code == 404
+    assert missing.json()["code"] == "not_found"
+    invalid = await http.post(
+        "/api/v2/roots/root/operations", json={"kind": "unknown"}
+    )
+    assert invalid.status_code == 422
+    assert invalid.json()["code"] == "invalid_request"
+    invalid_shape = await http.post(
+        "/api/v2/roots/root/operations", json=["not", "an", "object"]
+    )
+    assert invalid_shape.status_code == 422
+    assert invalid_shape.json()["code"] == "invalid_request"
+    missing_body = await http.post("/api/v2/roots/root/operations")
+    assert missing_body.status_code == 422
+    assert missing_body.json()["code"] == "invalid_request"
+
+    for path in (
+        "/api/v2/roots/root/attempts/missing",
+        "/api/v2/roots/root/validations/missing",
+        "/api/v2/roots/missing/tasks",
+    ):
+        response = await http.get(path)
+        assert response.status_code == 404
+        assert response.json()["code"] == "not_found"
+
+    for query in ("limit=nope", "limit=0", "wait_seconds=31"):
+        response = await http.get(f"/api/v2/roots/root/events?{query}")
+        assert response.status_code == 422
+        assert response.json()["code"] == "invalid_request"
+    empty = await http.get(
+        "/api/v2/roots/root/events?cursor=end&wait_seconds=0.001"
+    )
+    assert empty.status_code == 200
+    assert empty.json()["events"] == []
+
+    bad_cursor = await http.get("/api/v2/roots/root/events?cursor=bad")
+    assert bad_cursor.status_code == 422
+    for operation_id, status, code in (
+        ("conflict", 409, "conflict"),
+        ("unavailable", 503, "storage_unavailable"),
+    ):
+        response = await http.get(
+            f"/api/v2/roots/root/operations/{operation_id}"
+        )
+        assert response.status_code == status
+        assert response.json()["code"] == code
+
+    for key in ("", "x" * 201, "contains space"):
+        response = await http.post(
+            "/api/v2/roots/root/operations",
+            headers={"Idempotency-Key": key},
+            json={"kind": "dispatch", "task_ids": ["task-1"]},
+        )
+        assert response.status_code == 422
+        assert response.json()["code"] == "invalid_request"
+
+    schema = (await http.get("/openapi.json")).json()
+    operation_schema = schema["paths"]["/api/v2/roots/{root_trace_id}/operations"]["post"]
+    assert "discriminator" in operation_schema["requestBody"]["content"]["application/json"]["schema"]
+    assert operation_schema["responses"]["202"]["content"]["application/json"]["schema"]
+    assert operation_schema["responses"]["409"]["content"]["application/json"]["schema"]
+    await http.aclose()
+
+
+@pytest.mark.asyncio
+async def test_async_client_round_trip_and_watch(api_fixture):
+    _, http = api_fixture
+    client = OrchestrationClient("http://test", client=http)
+
+    operation = await client.start_revalidation(
+        "root",
+        "task-1",
+        "attempt-1",
+        deadline_at="2026-07-19T00:00:00+00:00",
+        idempotency_key="revalidate-once",
+    )
+    assert operation.kind == "revalidate"
+    assert (await client.poll("root", "operation-1")).operation_id == "operation-1"
+    assert (await client.stop("root", "operation-1")).status == "stopped"
+    assert (await client.resume("root", "operation-1")).status == "pending"
+    assert (await client.list_tasks("root")).items[0].task_id == "task-1"
+    assert (await client.get_task("root", "task-1")).task_id == "task-1"
+    assert (await client.list_attempts("root")).items[0].attempt_id == "attempt-1"
+    assert (await client.get_attempt("root", "attempt-1")).attempt_id == "attempt-1"
+    assert (
+        await client.list_validations("root")
+    ).items[0].validation_id == "validation-1"
+    assert (
+        await client.get_validation("root", "validation-1")
+    ).validation_id == "validation-1"
+    assert (await client.list_events("root", cursor=None)).next_cursor == "end"
+
+    stream = client.watch_events("root", wait_seconds=0)
+    assert (await anext(stream)).event_id == "event-1"
+    await stream.aclose()
+    await http.aclose()
+
+
+@pytest.mark.asyncio
+async def test_client_transport_retry_and_non_problem_error():
+    calls = 0
+
+    async def flaky(request):
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            raise httpx.ConnectError("temporary", request=request)
+        return httpx.Response(418, text="not problem details")
+
+    http = httpx.AsyncClient(
+        transport=httpx.MockTransport(flaky), base_url="http://test"
+    )
+    client = OrchestrationClient("http://test", client=http, max_retries=1)
+    with pytest.raises(OrchestrationAPIError) as caught:
+        await client.poll("root", "operation-1")
+    assert calls == 2
+    assert caught.value.code == "http_error"
+    assert caught.value.status_code == 418
+    await http.aclose()
+
+
+@pytest.mark.asyncio
+async def test_client_retries_mutation_only_with_idempotency_key():
+    attempts = {"unsafe": 0, "safe": 0}
+
+    async def handler(request):
+        key = "safe" if request.headers.get("Idempotency-Key") else "unsafe"
+        attempts[key] += 1
+        if attempts[key] < 2:
+            return httpx.Response(
+                503,
+                json={
+                    "title": "Unavailable",
+                    "status": 503,
+                    "detail": "retry",
+                    "code": "service_unavailable",
+                },
+            )
+        return httpx.Response(
+            200,
+            json={
+                "operation_id": "operation-1",
+                "root_trace_id": "root",
+                "kind": "dispatch",
+                "status": "pending",
+                "task_ids": ["task-1"],
+                "attempt_ids": [],
+                "validation_ids": [],
+                "result_ref": None,
+                "execution_epoch": 0,
+                "deadline_at": None,
+                "error": None,
+                "started_at": None,
+                "completed_at": None,
+                "created_at": "now",
+                "updated_at": "now",
+            },
+        )
+
+    http = httpx.AsyncClient(
+        transport=httpx.MockTransport(handler), base_url="http://test"
+    )
+    client = OrchestrationClient("http://test", client=http, max_retries=2)
+    with pytest.raises(OrchestrationAPIError):
+        await client.start_dispatch("root", ["task-1"])
+    assert attempts["unsafe"] == 1
+
+    result = await client.start_dispatch(
+        "root", ["task-1"], idempotency_key="safe-on-retry"
+    )
+    assert result.operation_id == "operation-1"
+    assert attempts["safe"] == 2
+    await http.aclose()