Editing Openai/694057b6-101c-8007-9a65-c40578c7252d (section)

=== Appendix A: Types & Protocols (public contract) ===

<syntaxhighlight lang="python">from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from typing import Any, Mapping, Sequence

from typing_extensions import (
    Literal,
    NotRequired,
    Protocol,
    TypedDict,
    runtime_checkable,
)

class WorkerState(str, Enum):
    RUNNING = "running"
    READY = "ready"
    FAILED = "failed"
    STOPPED = "stopped"

class RequestState(str, Enum):
    RUNNING = "running"
    TOOL_RUNNING = "tool_running"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELED = "canceled"

RequestFailReason = Literal[
    "worker_restarted",
    "server_died",
    "connect_failed",
    "headers_timeout",
    "stall_timeout",
    "tool_parse_error",
    "tool_execution_error",
    "repeated_line_loop",
    "canceled",
    "unknown_error",
]

RequestFinishReason = Literal[
    "stop",
    "max_tokens",
    "canceled",
    "failed",
]

class ChatMessage(TypedDict, total=False):
    role: Literal["system", "user", "assistant", "tool"]
    content: str
    name: str
    tool_call_id: str
    # Backends may include extra fields; keep permissive.
    tool_calls: Any

class ToolFunctionDef(TypedDict, total=False):
    name: str
    description: str
    parameters: dict[str, Any]

class ToolDef(TypedDict, total=False):
    type: Literal["function"]
    function: ToolFunctionDef

class ToolCall(TypedDict, total=False):
    id: str
    type: Literal["function"]
    function: dict[str, Any]  # expects {"name": str, "arguments": str|dict}

class ExitSignal(TypedDict, total=False):
    tool_name: str
    arguments: dict[str, Any]
    emitted_at: float

class SubmitOk(TypedDict):
    ok: Literal[True]
    request_id: int

class SubmitErr(TypedDict):
    ok: Literal[False]
    error: Literal["NO_SLOT_AVAILABLE", "WORKER_NOT_READY", "WORKER_FAILED"]

SubmitResult = SubmitOk | SubmitErr

class RequestStatus(TypedDict, total=False):
    request_id: int
    job_name: str
    state: RequestState

    created_at: float
    dispatched_at: NotRequired[float]
    completed_at: NotRequired[float]

    last_progress_at: NotRequired[float]
    output_chars: NotRequired[int]

    # optional nice-to-haves
    tokens_received: NotRequired[int]
    tokens_per_second: NotRequired[float]

    # normal tool iteration budget remaining
    tool_iters_remaining: NotRequired[int]

    # exit-tool signals recorded
    signals: NotRequired[list[ExitSignal]]

    # terminal details
    fail_reason: NotRequired[RequestFailReason]
    fail_detail: NotRequired[str]

class RequestResult(TypedDict):
    request_id: int
    job_name: str

    state: Literal["completed", "failed", "canceled"]
    finish_reason: RequestFinishReason

    # Always returned (even on failure/cancel). May be "".
    text: str

    signals: NotRequired[list[ExitSignal]]
    fail_reason: NotRequired[RequestFailReason]
    fail_detail: NotRequired[str]

class NotFound(TypedDict):
    ok: Literal[False]
    error: Literal["NOT_FOUND"]

GetResultResponse = RequestResult | NotFound
GetStatusResponse = RequestStatus | NotFound

class WorkerStatus(TypedDict, total=False):
    state: WorkerState
    slots_total: int
    slots_used: int
    active_request_ids: list[int]

    restart_count: int
    last_error: NotRequired[str]
    last_ready_at: NotRequired[float]

class WorkerDebugInfo(TypedDict, total=False):
    recent_logs: list[str]
    recent_restart_reasons: list[str]

@dataclass(frozen=True, slots=True)
class TimeoutProfile:
    connect_timeout_s: float
    headers_timeout_s: float

    ttft_timeout_s: float | None
    prefill_liveness_timeout_s: float | None
    idle_stream_timeout_s: float | None
    absolute_timeout_s: float | None

    liveness_probe_interval_s: float

    restart_backoff_s: float
    restart_window_s: float
    max_restarts_per_window: int

@dataclass(frozen=True, slots=True)
class BiosContext:
    now: datetime
    timezone_name: str
    worker_name: str

    tool_iters_remaining: int
    normal_tools: Sequence[ToolDef]
    exit_tools: Sequence[ToolDef]

    bios_version: str = "bios-v1"

@runtime_checkable
class BiosProvider(Protocol):
    def __call__(self, ctx: BiosContext) -> str: ...

def build_message_stack(
    *,
    bios_text: str,
    caller_system_prompt: str,
    conversation: Sequence[ChatMessage],
) -> list[ChatMessage]:
    """Pure function: returns full message list in required order."""
    raise NotImplementedError

@runtime_checkable
class ToolRunner(Protocol):
    async def run_tool(
        self,
        *,
        name: str,
        arguments: dict[str, Any],
        request_id: int,
        job_name: str,
    ) -> Any:
        """
        Return any JSON-serializable result. "No results" is not an error;
        represent it as an empty structure ([], {"results": []}, etc.).
        """
        ...

GenerationParams = Mapping[str, Any]

@dataclass(frozen=True, slots=True)
class WorkerConfig:
    name: str

    host: str
    port: int

    # full command including executable
    server_cmd: Sequence[str]
    env: Mapping[str, str]

    slots: int
    timeouts: TimeoutProfile

    # Normal tools (round-trip)
    normal_tools: Sequence[ToolDef]
    tool_runner: ToolRunner | None

    # Exit tools (one-way)
    exit_tools: Sequence[ToolDef]

    # BIOS
    bios_provider: BiosProvider
    timezone_name: str

@dataclass(frozen=True, slots=True)
class ReadinessProbe:
    method: Literal["GET"] = "GET"
    path: str = "/v1/models"

</syntaxhighlight>