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: ...