Agent/cyber_agent/application/ports.py

"""Application-owned service ports used by the framework runtime."""

from __future__ import annotations

from typing import Any, Literal, Protocol, Sequence

from pydantic import Field

from cyber_agent.application.models import ApplicationModel, ApplicationRef
from cyber_agent.core.task_protocol import ContextRef, TaskBrief


class ContextRequest(ApplicationModel):
    application_ref: ApplicationRef
    root_trace_id: str = Field(min_length=1)
    trace_id: str = Field(min_length=1)
    parent_trace_id: str | None = None
    uid: str | None = None
    role_id: str = Field(min_length=1)
    task_brief: TaskBrief | None = None
    task_brief_revision: int = Field(default=0, ge=0)
    authorized_context_refs: tuple[ContextRef, ...] = ()


class ContextResourceDescriptor(ApplicationModel):
    application_ref: ApplicationRef
    root_trace_id: str = Field(min_length=1)
    trace_id: str = Field(min_length=1)
    uid: str | None = None
    role_id: str = Field(min_length=1)
    source_id: str = Field(min_length=1, max_length=500)
    source_version: str = Field(min_length=1, max_length=200)
    content_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
    kind: str = Field(min_length=1, max_length=200)
    summary: str = Field(min_length=1, max_length=500)
    inheritance: Literal["task_only", "inheritable"] = "task_only"
    priority: int = 0
    estimated_chars: int = Field(default=0, ge=0)
    authorization: dict[str, Any] = Field(default_factory=dict)


class ContextMaterial(ApplicationModel):
    descriptor: ContextResourceDescriptor
    content: dict[str, Any]


class TaskContextProvider(Protocol):
    async def list_context(
        self,
        request: ContextRequest,
    ) -> Sequence[ContextResourceDescriptor]: ...

    async def resolve_context(
        self,
        request: ContextRequest,
        descriptor: ContextResourceDescriptor,
    ) -> ContextMaterial: ...


class RunEventProjector(Protocol):
    """Application projection port; cursor movement is part of its transaction."""

    async def load_cursor(
        self,
        application_ref: ApplicationRef,
        root_trace_id: str,
    ) -> int: ...

    async def project(self, event: Any, expected_cursor: int) -> None: ...

    async def advance_cursor(
        self,
        application_ref: ApplicationRef,
        root_trace_id: str,
        event_id: int,
        expected_cursor: int,
    ) -> None:
        """Atomically skip an unsupported event without updating business views."""
        ...