script-build-and-framework-tmp/agent/agent/orchestration/wire.py

"""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 agent.failures import FailureDetail

from .models import (
    ArtifactSnapshot,
    BackgroundOperation,
    AttemptStatus,
    DecisionAction,
    FailureCode,
    OperationKind,
    OperationStatus,
    OrchestrationEvent,
    PlannerDecision,
    TaskAttempt,
    TaskRecord,
    TaskStatus,
    ValidationReport,
    ValidationMode,
    ValidationRunStatus,
    ValidationVerdict,
    json_values,
)


class WireModel(BaseModel):
    model_config = ConfigDict(extra="forbid")


class ResourceClaimView(WireModel):
    uri: str
    exclusive: bool = True


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
    resource_claims: List["ResourceClaimView"] = Field(default_factory=list)
    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]
    resource_claims: List[ResourceClaimView] = Field(default_factory=list)
    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]
    preflight_rule_ids: List[str]
    max_evidence_queries: int
    max_evidence_items_per_query: int
    evidence_timeout_seconds: float


class ValidationEvidenceGrantView(WireModel):
    artifact_ref: ArtifactRefView
    handle: str
    source: str
    granted_at: str


class ValidationEvidenceSnapshotView(WireModel):
    revision: int
    grants: List[ValidationEvidenceGrantView]
    updated_at: str


class ValidationReadSessionView(WireModel):
    evidence_revision: int
    required_handles: List[str]
    authorized_handles: List[str]
    completed_handles: List[str]
    next_cursors: Dict[str, Optional[str]]
    active_reads: Dict[str, Optional[str]]
    read_results: Dict[str, Dict[str, Any]]


class ExecutionStatsView(WireModel):
    primary_model: Optional[str] = None
    total_tokens: Optional[int] = None
    total_cost: Optional[float] = None
    failure_code: Optional[FailureCode] = None


class FailureDetailView(WireModel):
    code: str
    message: str
    disposition: str
    source_tool: Optional[str] = None
    details: Dict[str, Any]

    @classmethod
    def from_domain(cls, value: FailureDetail) -> "FailureDetailView":
        return cls.model_validate(value.to_dict())


class TaskView(WireModel):
    task_id: str
    # Deprecated compatibility field. Explicit ledgers no longer persist or
    # project GoalTree identities, so new responses always expose ``None``.
    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
    accepted_child_decision_ids: Optional[List[str]] = None
    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
    failure: Optional[FailureDetailView] = None
    worker_summary: str = ""
    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
    evidence_snapshot: Optional[ValidationEvidenceSnapshotView] = None
    read_session: Optional[ValidationReadSessionView] = None
    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
    failure: Optional[FailureDetailView] = 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 PlannerDecisionView(WireModel):
    decision_id: str
    task_id: str
    action: DecisionAction
    reason: str
    from_status: TaskStatus
    to_status: TaskStatus
    attempt_id: Optional[str] = None
    validation_id: Optional[str] = None
    payload: Dict[str, Any]
    created_at: str

    @classmethod
    def from_domain(cls, value: PlannerDecision) -> "PlannerDecisionView":
        return cls.model_validate(json_values(asdict(value)))


class MissionSnapshotView(WireModel):
    schema_version: Literal[1] = 1
    revision: int
    root_trace_id: str
    root_task_id: Optional[str] = None
    root_objective: str
    focused_task_id: Optional[str] = None
    tasks: List[TaskView]
    attempts: List[AttemptView]
    validations: List[ValidationView]
    decisions: List[PlannerDecisionView]
    operations: List[OperationView]


class RootCompletionView(WireModel):
    root_task_id: str
    root_objective: str
    status: TaskStatus
    blocked_reason: Optional[str] = None
    result_summary: Optional[str] = None


class ArtifactSnapshotView(WireModel):
    snapshot_id: str
    attempt_id: str
    sha256: str
    artifact_refs: List[ArtifactRefView]
    evidence_refs: List[ArtifactRefView]
    created_at: str

    @classmethod
    def from_domain(cls, value: ArtifactSnapshot) -> "ArtifactSnapshotView":
        return cls(
            snapshot_id=value.snapshot_id,
            attempt_id=value.attempt_id,
            sha256=value.sha256,
            artifact_refs=json_values([asdict(item) for item in value.artifact_refs]),
            evidence_refs=json_values([asdict(item) for item in value.evidence_refs]),
            created_at=value.created_at,
        )


class TaskList(WireModel):
    revision: int
    root_task_id: Optional[str] = None
    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 CapabilitiesView(WireModel):
    api_version: Literal["v2"] = "v2"
    mission_snapshot_schema_version: Literal[1] = 1
    event_schema_version: Literal[2] = 2
    capabilities: List[str]


class ProblemDetails(WireModel):
    type: str = "about:blank"
    title: str
    status: int
    detail: str
    code: str
    instance: Optional[str] = None


__all__ = [
    "ArtifactSnapshotView",
    "AttemptList",
    "AttemptView",
    "CapabilitiesView",
    "DispatchOperationRequest",
    "EventPageView",
    "EventView",
    "ExecutionStatsView",
    "MissionSnapshotView",
    "OperationRequest",
    "OperationView",
    "PlannerDecisionView",
    "ProblemDetails",
    "RevalidateOperationRequest",
    "RootCompletionView",
    "TaskList",
    "TaskView",
    "ValidationList",
    "ValidationView",
]