distinct_protocol/models.py

"""Small, dependency-free contracts for Distinct's Gradio control plane.

The objects in this module cross a trust boundary.  Constructors therefore
validate sizes and types instead of relying on callers to be well behaved.
"""

from __future__ import annotations

import json
import math
import re
import time
from dataclasses import dataclass, field
from enum import Enum
from types import MappingProxyType
from typing import Any, Dict, Mapping, Optional, Tuple

PROTOCOL_VERSION = "1.0"
_ID_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_.:@-]{0,127}$")


#: The complete vocabulary of run phases a worker may report.
#:
#: A phase is a *closed set*, not free text, for one reason: the interface
#: promises to say why a run is waiting, and a promise kept with worker-supplied
#: prose is not kept at all.  The worker names a phase from this list; the server
#: maps it to copy it wrote itself.  An unrecognised phase degrades to
#: ``PHASE_WORKING`` rather than reaching a user, so a hostile or merely outdated
#: worker can move the spinner but never write the caption.
PHASE_WAITING_FOR_WORKER = "waiting-for-worker"
PHASE_OFFERED = "offered"
PHASE_QUEUED_ON_WORKER = "queued-on-worker"
PHASE_FETCHING_WEIGHTS = "fetching-weights"
PHASE_LOADING_MODEL = "loading-model"
PHASE_GUARD_INPUT = "guard-input"
PHASE_PLANNING = "planning"
PHASE_GENERATING = "generating"
PHASE_CALLING_TOOL = "calling-tool"
PHASE_GUARD_OUTPUT = "guard-output"
PHASE_WORKING = "working"
PHASE_DONE = "done"

RUN_PHASES: Tuple[str, ...] = (
    PHASE_WAITING_FOR_WORKER,
    PHASE_OFFERED,
    PHASE_QUEUED_ON_WORKER,
    PHASE_FETCHING_WEIGHTS,
    PHASE_LOADING_MODEL,
    PHASE_GUARD_INPUT,
    PHASE_PLANNING,
    PHASE_GENERATING,
    PHASE_CALLING_TOOL,
    PHASE_GUARD_OUTPUT,
    PHASE_WORKING,
    PHASE_DONE,
)

#: Kinds of live step a worker may stream while a run is in flight.
STEP_PHASE = "phase"
STEP_MODEL = "model"
STEP_TOOL_CALL = "tool-call"
STEP_TOOL_RESULT = "tool-result"
STEP_KINDS: Tuple[str, ...] = (STEP_PHASE, STEP_MODEL, STEP_TOOL_CALL, STEP_TOOL_RESULT)

#: The two agent modes, named once. "simple" is a planning pass, the tools it
#: named, and an answering pass: two inferences, no loop, which is what a
#: 0.6B model can actually sustain. "agent" is a react loop that sees each
#: tool result before deciding the next call, which is stronger and costs a
#: whole inference per step.
MODE_SIMPLE = "simple"
MODE_AGENT = "agent"
AGENT_MODES: Tuple[str, ...] = (MODE_SIMPLE, MODE_AGENT)

MAX_LIVE_STEPS_PER_JOB = 24

#: A finished run may report more steps than a live stream carries: the live
#: stream is a window on what is happening now, and this is the whole record.
#: Still bounded, because it crosses the wire and lands in a browser.
MAX_RUN_STEPS = 64
MAX_LIVE_STEP_TEXT = 600


def normalise_phase(value: Any) -> str:
    """Coerce an untrusted phase name into the vocabulary, never past it."""

    text = str(value or "").strip()
    return text if text in RUN_PHASES else PHASE_WORKING


class ProtocolError(ValueError):
    """Raised when an untrusted wire value violates the protocol."""


def utc_now() -> float:
    return time.time()


def canonical_json(value: Any) -> str:
    """Serialize a value in the one form used for HMAC signatures."""

    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False)


def _identifier(value: Any, label: str) -> str:
    if not isinstance(value, str) or not _ID_RE.fullmatch(value):
        raise ProtocolError(f"{label} is not a valid identifier")
    return value


def _text(value: Any, label: str, maximum: int, *, allow_empty: bool = False) -> str:
    if not isinstance(value, str):
        raise ProtocolError(f"{label} must be text")
    value = value.strip()
    if not allow_empty and not value:
        raise ProtocolError(f"{label} cannot be empty")
    if len(value) > maximum:
        raise ProtocolError(f"{label} exceeds {maximum} characters")
    return value


def _finite(value: Any, label: str, *, minimum: float = 0.0) -> float:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ProtocolError(f"{label} must be numeric")
    result = float(value)
    if not math.isfinite(result) or result < minimum:
        raise ProtocolError(f"{label} must be finite and >= {minimum}")
    return result


def _plain(value: Any, *, depth: int = 0) -> Any:
    if depth > 8:
        raise ProtocolError("nested value is too deep")
    if value is None or isinstance(value, (str, bool, int)):
        return value
    if isinstance(value, float):
        if not math.isfinite(value):
            raise ProtocolError("non-finite numbers are not valid JSON")
        return value
    if isinstance(value, Mapping):
        if len(value) > 128:
            raise ProtocolError("mapping has too many entries")
        return {str(k): _plain(v, depth=depth + 1) for k, v in value.items()}
    if isinstance(value, (list, tuple)):
        if len(value) > 256:
            raise ProtocolError("sequence has too many entries")
        return [_plain(v, depth=depth + 1) for v in value]
    raise ProtocolError(f"unsupported JSON value: {type(value).__name__}")


def _mapping(value: Optional[Mapping[str, Any]]) -> Mapping[str, Any]:
    return MappingProxyType(_plain(value or {}))


class JobStatus(str, Enum):
    BLOCKED = "blocked"
    OFFERED = "offered"
    ACCEPTED = "accepted"
    QUEUED = "queued"
    RUNNING = "running"
    #: Non-terminal: the agent holding this job has stopped polling. The job
    #: is not failed and not expired; it is waiting for a decision (wait for
    #: the agent, reassign, or cancel), and it returns to its previous state
    #: if the agent comes back.
    STRANDED = "stranded"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"
    EXPIRED = "expired"


class AgentStatus(str, Enum):
    ONLINE = "online"
    BUSY = "busy"
    OVERLOADED = "overloaded"
    DRAINING = "draining"
    OFFLINE = "offline"


@dataclass(frozen=True)
class ChatMessage:
    role: str
    content: str

    def __post_init__(self) -> None:
        if self.role not in {"system", "user", "assistant", "tool"}:
            raise ProtocolError("message role is not supported")
        object.__setattr__(self, "content", _text(self.content, "content", 65_536))

    def to_dict(self) -> Dict[str, Any]:
        return {"role": self.role, "content": self.content}

    @classmethod
    def from_dict(cls, value: Mapping[str, Any]) -> "ChatMessage":
        return cls(role=value.get("role", ""), content=value.get("content", ""))


@dataclass(frozen=True)
class ToolSelection:
    id: str
    version: str = "1"
    config: Mapping[str, Any] = field(default_factory=dict)

    def __post_init__(self) -> None:
        object.__setattr__(self, "id", _identifier(self.id, "tool id"))
        object.__setattr__(self, "version", _text(self.version, "tool version", 32))
        object.__setattr__(self, "config", _mapping(self.config))

    def to_dict(self) -> Dict[str, Any]:
        return {"id": self.id, "version": self.version, "config": _plain(self.config)}

    @classmethod
    def from_dict(cls, value: Mapping[str, Any]) -> "ToolSelection":
        return cls(
            id=value.get("id", ""),
            version=value.get("version", "1"),
            config=value.get("config") or {},
        )


@dataclass(frozen=True)
class ModelManifest:
    """A pinned model file.

    ``sha256`` is the weights identity.  An empty digest is *not* a passing
    verification: it means no digest has been recorded yet, and callers must
    surface that state rather than silently treating the file as verified.
    ``source_repo`` and ``source_revision`` say which upstream publication the
    digest was taken from, because the same nominal quantisation published by
    two different repositories is two different files with two different
    digests.
    """

    id: str
    label: str
    filename: str
    min_ram_gb: float
    sha256: str = ""
    context_length: int = 4096
    source_repo: str = ""
    source_revision: str = ""

    def __post_init__(self) -> None:
        object.__setattr__(self, "id", _identifier(self.id, "model id"))
        object.__setattr__(self, "label", _text(self.label, "model label", 120))
        object.__setattr__(self, "filename", _text(self.filename, "model filename", 255))
        object.__setattr__(self, "min_ram_gb", _finite(self.min_ram_gb, "minimum RAM", minimum=0.1))
        if self.sha256 and not re.fullmatch(r"[0-9a-fA-F]{64}", self.sha256):
            raise ProtocolError("model sha256 must be 64 hexadecimal characters")
        object.__setattr__(self, "sha256", self.sha256.lower())
        object.__setattr__(
            self, "source_repo", _text(self.source_repo, "source repo", 200, allow_empty=True)
        )
        object.__setattr__(
            self,
            "source_revision",
            _text(self.source_revision, "source revision", 120, allow_empty=True),
        )
        if self.sha256 and not self.source_repo:
            # A digest with no stated origin cannot be re-derived or audited.
            raise ProtocolError("a recorded sha256 requires a source_repo")
        if isinstance(self.context_length, bool) or not isinstance(self.context_length, int):
            raise ProtocolError("context length must be an integer")
        if not 256 <= self.context_length <= 1_048_576:
            raise ProtocolError("context length is outside the supported range")

    @property
    def digest_recorded(self) -> bool:
        """True only when a real weights digest is pinned for this model."""

        return bool(self.sha256)

    @property
    def identity_status(self) -> str:
        """The word the UI must show for this model's weights identity."""

        return "pinned" if self.digest_recorded else "not recorded"

    def to_dict(self) -> Dict[str, Any]:
        return {
            "id": self.id,
            "label": self.label,
            "filename": self.filename,
            "min_ram_gb": self.min_ram_gb,
            "sha256": self.sha256,
            "context_length": self.context_length,
            "source_repo": self.source_repo,
            "source_revision": self.source_revision,
        }

    @classmethod
    def from_dict(cls, value: Mapping[str, Any]) -> "ModelManifest":
        return cls(
            id=value.get("id", ""),
            label=value.get("label", ""),
            filename=value.get("filename", ""),
            min_ram_gb=value.get("min_ram_gb", 0),
            sha256=value.get("sha256", ""),
            context_length=value.get("context_length", 4096),
            source_repo=value.get("source_repo", ""),
            source_revision=value.get("source_revision", ""),
        )


@dataclass(frozen=True)
class JobSpec:
    id: str
    session_id: str
    conversation_id: str
    parent_job_id: Optional[str]
    prompt: str
    model_id: str
    allowed_tools: Tuple[ToolSelection, ...]
    target_agent_id: str
    created_at: float
    messages: Tuple[ChatMessage, ...] = ()
    limits: Mapping[str, Any] = field(default_factory=dict)

    def __post_init__(self) -> None:
        for label in ("id", "session_id", "conversation_id", "model_id", "target_agent_id"):
            object.__setattr__(self, label, _identifier(getattr(self, label), label))
        if self.parent_job_id is not None:
            object.__setattr__(self, "parent_job_id", _identifier(self.parent_job_id, "parent job id"))
        object.__setattr__(self, "prompt", _text(self.prompt, "prompt", 65_536))
        object.__setattr__(self, "created_at", _finite(self.created_at, "created_at"))
        tools = tuple(
            item if isinstance(item, ToolSelection) else ToolSelection.from_dict(item)
            for item in self.allowed_tools
        )
        if len(tools) > 16 or len({tool.id for tool in tools}) != len(tools):
            raise ProtocolError("tools must be unique and limited to 16")
        object.__setattr__(self, "allowed_tools", tools)
        messages = tuple(
            item if isinstance(item, ChatMessage) else ChatMessage.from_dict(item)
            for item in self.messages
        )
        if len(messages) > 256:
            raise ProtocolError("conversation snapshot contains too many messages")
        object.__setattr__(self, "messages", messages)
        object.__setattr__(self, "limits", _mapping(self.limits))

    def to_dict(self) -> Dict[str, Any]:
        return {
            "protocol_version": PROTOCOL_VERSION,
            "id": self.id,
            "session_id": self.session_id,
            "conversation_id": self.conversation_id,
            "parent_job_id": self.parent_job_id,
            "prompt": self.prompt,
            "model_id": self.model_id,
            "allowed_tools": [tool.to_dict() for tool in self.allowed_tools],
            "target_agent_id": self.target_agent_id,
            "created_at": self.created_at,
            "messages": [message.to_dict() for message in self.messages],
            "limits": _plain(self.limits),
        }

    @classmethod
    def from_dict(cls, value: Mapping[str, Any]) -> "JobSpec":
        version = value.get("protocol_version", PROTOCOL_VERSION)
        if version != PROTOCOL_VERSION:
            raise ProtocolError(f"unsupported protocol version {version!r}")
        return cls(
            id=value.get("id", ""),
            session_id=value.get("session_id", ""),
            conversation_id=value.get("conversation_id", ""),
            parent_job_id=value.get("parent_job_id"),
            prompt=value.get("prompt", ""),
            model_id=value.get("model_id", ""),
            allowed_tools=tuple(ToolSelection.from_dict(v) for v in value.get("allowed_tools", ())),
            target_agent_id=value.get("target_agent_id", ""),
            created_at=value.get("created_at", 0),
            messages=tuple(ChatMessage.from_dict(v) for v in value.get("messages", ())),
            limits=value.get("limits") or {},
        )


@dataclass(frozen=True)
class AgentCapabilities:
    agent_id: str
    name: str
    os: str
    arch: str
    cpu: str
    ram_gb: float
    models: Tuple[str, ...]
    tools: Tuple[str, ...]
    energy_provider: str
    #: Which agent mode this worker will run. Advertised rather than assumed,
    #: because the user choosing a worker is choosing how their request will
    #: be answered: "simple" is one planning pass, the tools it named, and one
    #: answering pass, and "agent" is a loop that can react to what a tool
    #: returned. A worker offering only one of them cannot be sent the other.
    modes: Tuple[str, ...] = ()
    queue_capacity: int = 4
    max_concurrency: int = 1

    def __post_init__(self) -> None:
        if self.agent_id:
            object.__setattr__(self, "agent_id", _identifier(self.agent_id, "agent id"))
        object.__setattr__(self, "name", _text(self.name, "agent name", 80))
        object.__setattr__(self, "os", _text(self.os, "operating system", 80))
        object.__setattr__(self, "arch", _text(self.arch, "architecture", 40))
        object.__setattr__(self, "cpu", _text(self.cpu, "cpu", 160))
        object.__setattr__(self, "ram_gb", _finite(self.ram_gb, "RAM"))
        object.__setattr__(self, "models", tuple(_identifier(v, "model id") for v in self.models))
        object.__setattr__(self, "tools", tuple(_identifier(v, "tool id") for v in self.tools))
        modes = tuple(str(v) for v in self.modes if str(v) in AGENT_MODES)
        # An older worker advertises nothing here. It predates the choice, and
        # what it actually runs is the simple mode, so that is what it is
        # recorded as offering rather than being dropped from every list.
        object.__setattr__(self, "modes", modes or (MODE_SIMPLE,))
        for label in ("queue_capacity", "max_concurrency"):
            value = getattr(self, label)
            if isinstance(value, bool) or not isinstance(value, int) or not 1 <= value <= 64:
                raise ProtocolError(f"{label} must be an integer from 1 to 64")
        if self.max_concurrency > self.queue_capacity:
            raise ProtocolError("max_concurrency cannot exceed queue_capacity")

    def to_dict(self) -> Dict[str, Any]:
        return {
            "agent_id": self.agent_id,
            "name": self.name,
            "os": self.os,
            "arch": self.arch,
            "cpu": self.cpu,
            "ram_gb": self.ram_gb,
            "models": list(self.models),
            "tools": list(self.tools),
            "energy_provider": self.energy_provider,
            "modes": list(self.modes),
            "queue_capacity": self.queue_capacity,
            "max_concurrency": self.max_concurrency,
        }

    @classmethod
    def from_dict(cls, value: Mapping[str, Any]) -> "AgentCapabilities":
        return cls(
            agent_id=value.get("agent_id", ""),
            name=value.get("name", "Unnamed agent"),
            os=value.get("os", "unknown"),
            arch=value.get("arch", "unknown"),
            cpu=value.get("cpu", "unknown"),
            ram_gb=value.get("ram_gb", 0),
            models=tuple(value.get("models", ())),
            tools=tuple(value.get("tools", ())),
            energy_provider=value.get("energy_provider", "unavailable"),
            modes=tuple(value.get("modes", ())),
            queue_capacity=value.get("queue_capacity", 4),
            max_concurrency=value.get("max_concurrency", 1),
        )


@dataclass(frozen=True)
class AgentSnapshot:
    agent_id: str
    status: AgentStatus
    active_job_ids: Tuple[str, ...]
    queued_job_ids: Tuple[str, ...]
    queue_capacity: int
    estimated_wait_s: float
    progress: Mapping[str, float]
    energy_available: bool
    last_seen: float
    #: Short, human-readable per-job progress notes ("Local model step 2",
    #: "Calling allowed tool calculate"). Worker-supplied and therefore
    #: untrusted display text; the server escapes it like every other
    #: worker-supplied string. Optional on the wire for older workers.
    progress_notes: Mapping[str, str] = field(default_factory=dict)
    #: Machine-readable phase per job, drawn from :data:`RUN_PHASES`. This is
    #: what lets the interface say *why* a run is waiting instead of only that
    #: it is. Anything outside the vocabulary becomes ``PHASE_WORKING`` here, at
    #: the trust boundary, rather than being escaped further downstream.
    progress_phase: Mapping[str, str] = field(default_factory=dict)
    #: Bounded live step stream per job: what the model just did, which tool it
    #: called, what came back. Replaced wholesale on every poll rather than
    #: appended to, so a worker cannot grow the snapshot without bound.
    live_steps: Mapping[str, Tuple[Mapping[str, Any], ...]] = field(default_factory=dict)
    #: Monotonic-free wall clock at which the worker began the run, used by the
    #: server to price an estimate against comparable completed runs.
    started_at: Mapping[str, float] = field(default_factory=dict)
    #: Seconds this worker has been running, self-reported.
    uptime_seconds: float = 0.0
    #: Energy this worker has spent on runs since it started, in joules, or
    #: ``None`` when nothing was ever measured. ``None`` is not zero, and the
    #: two must not be conflated: a worker with no usable meter has spent
    #: energy it cannot report.
    lifetime_joules: Optional[float] = None
    #: How many runs this worker has completed, and how many of those
    #: contributed a measurement. The pair is what lets a total declare itself
    #: a floor rather than a sum.
    lifetime_runs: int = 0
    lifetime_measured_runs: int = 0
    #: The measurement boundary of ``lifetime_joules`` (``cpu-package``, ``gpu``
    #: and so on). Different scopes are different questions and never add.
    energy_scope: str = ""

    def __post_init__(self) -> None:
        object.__setattr__(self, "agent_id", _identifier(self.agent_id, "agent id"))
        if not isinstance(self.status, AgentStatus):
            object.__setattr__(self, "status", AgentStatus(self.status))
        object.__setattr__(self, "active_job_ids", tuple(_identifier(v, "job id") for v in self.active_job_ids))
        object.__setattr__(self, "queued_job_ids", tuple(_identifier(v, "job id") for v in self.queued_job_ids))
        if isinstance(self.queue_capacity, bool) or not isinstance(self.queue_capacity, int):
            raise ProtocolError("queue capacity must be an integer")
        if not 1 <= self.queue_capacity <= 64:
            raise ProtocolError("queue capacity is outside the supported range")
        object.__setattr__(self, "estimated_wait_s", _finite(self.estimated_wait_s, "estimated wait"))
        progress = {str(k): _finite(v, "progress") for k, v in self.progress.items()}
        if any(value > 1.0 for value in progress.values()):
            raise ProtocolError("progress must be between zero and one")
        object.__setattr__(self, "progress", MappingProxyType(progress))
        notes: Dict[str, str] = {}
        if not isinstance(self.progress_notes, Mapping):
            raise ProtocolError("progress notes must be a mapping")
        if len(self.progress_notes) > 64:
            raise ProtocolError("too many progress notes")
        for key, note in self.progress_notes.items():
            if not isinstance(note, str):
                raise ProtocolError("progress notes must be text")
            notes[_identifier(key, "job id")] = note[:200]
        object.__setattr__(self, "progress_notes", MappingProxyType(notes))
        object.__setattr__(self, "last_seen", _finite(self.last_seen, "last seen"))
        object.__setattr__(self, "progress_phase", _phase_map(self.progress_phase))
        object.__setattr__(self, "live_steps", _live_step_map(self.live_steps))
        object.__setattr__(self, "started_at", _started_map(self.started_at))
        object.__setattr__(
            self, "uptime_seconds", _finite(self.uptime_seconds, "uptime", minimum=0.0)
        )
        if self.lifetime_joules is not None:
            object.__setattr__(
                self,
                "lifetime_joules",
                _finite(self.lifetime_joules, "lifetime joules", minimum=0.0),
            )
        for name in ("lifetime_runs", "lifetime_measured_runs"):
            value = getattr(self, name)
            if isinstance(value, bool) or not isinstance(value, int) or value < 0:
                raise ProtocolError(f"{name.replace('_', ' ')} must be a non-negative integer")
        if self.lifetime_measured_runs > self.lifetime_runs:
            raise ProtocolError("measured runs cannot exceed completed runs")
        object.__setattr__(self, "energy_scope", str(self.energy_scope or "")[:64])

    def to_dict(self) -> Dict[str, Any]:
        return {
            "agent_id": self.agent_id,
            "status": self.status.value,
            "active_job_ids": list(self.active_job_ids),
            "queued_job_ids": list(self.queued_job_ids),
            "queue_capacity": self.queue_capacity,
            "estimated_wait_s": self.estimated_wait_s,
            "progress": dict(self.progress),
            "progress_notes": dict(self.progress_notes),
            "progress_phase": dict(self.progress_phase),
            "live_steps": {key: [dict(step) for step in steps] for key, steps in self.live_steps.items()},
            "started_at": dict(self.started_at),
            "energy_available": self.energy_available,
            "last_seen": self.last_seen,
            "uptime_seconds": self.uptime_seconds,
            "lifetime_joules": self.lifetime_joules,
            "lifetime_runs": self.lifetime_runs,
            "lifetime_measured_runs": self.lifetime_measured_runs,
            "energy_scope": self.energy_scope,
        }

    @classmethod
    def from_dict(cls, value: Mapping[str, Any]) -> "AgentSnapshot":
        return cls(
            agent_id=value.get("agent_id", ""),
            status=AgentStatus(value.get("status", AgentStatus.OFFLINE.value)),
            active_job_ids=tuple(value.get("active_job_ids", ())),
            queued_job_ids=tuple(value.get("queued_job_ids", ())),
            queue_capacity=value.get("queue_capacity", 4),
            estimated_wait_s=value.get("estimated_wait_s", 0),
            progress=value.get("progress") or {},
            energy_available=bool(value.get("energy_available", False)),
            last_seen=value.get("last_seen", 0),
            progress_notes=value.get("progress_notes") or {},
            progress_phase=value.get("progress_phase") or {},
            live_steps=value.get("live_steps") or {},
            started_at=value.get("started_at") or {},
            uptime_seconds=value.get("uptime_seconds", 0.0),
            lifetime_joules=value.get("lifetime_joules"),
            lifetime_runs=value.get("lifetime_runs", 0),
            lifetime_measured_runs=value.get("lifetime_measured_runs", 0),
            energy_scope=value.get("energy_scope", ""),
        )


def _phase_map(value: Any) -> Mapping[str, str]:
    if not isinstance(value, Mapping):
        raise ProtocolError("progress phases must be a mapping")
    if len(value) > 64:
        raise ProtocolError("too many progress phases")
    return MappingProxyType(
        {_identifier(key, "job id"): normalise_phase(phase) for key, phase in value.items()}
    )


def _started_map(value: Any) -> Mapping[str, float]:
    if not isinstance(value, Mapping):
        raise ProtocolError("start times must be a mapping")
    if len(value) > 64:
        raise ProtocolError("too many start times")
    return MappingProxyType(
        {_identifier(key, "job id"): _finite(started, "start time") for key, started in value.items()}
    )


def _run_step(value: Any) -> Mapping[str, Any]:
    """One step of a run, normalised. Worker-supplied, so nothing is trusted.

    Shares its vocabulary with the live stream on purpose: the same step is
    shown while a run is in flight and again once it has finished, and two
    shapes for one fact would eventually disagree.
    """

    if not isinstance(value, Mapping):
        raise ProtocolError("each step must be a mapping")
    kind = str(value.get("kind") or STEP_PHASE)
    entry: Dict[str, Any] = {
        "kind": kind if kind in STEP_KINDS else STEP_PHASE,
        "phase": normalise_phase(value.get("phase")),
        "text": str(value.get("text") or "")[:MAX_LIVE_STEP_TEXT],
    }
    tool = value.get("tool")
    if isinstance(tool, str) and tool:
        entry["tool"] = tool[:128]
    if "ok" in value:
        entry["ok"] = bool(value.get("ok"))
    elapsed = value.get("elapsed_ms")
    if isinstance(elapsed, int) and not isinstance(elapsed, bool) and elapsed >= 0:
        entry["elapsed_ms"] = min(elapsed, 86_400_000)
    return MappingProxyType(entry)


def _run_steps(value: Any) -> Tuple[Mapping[str, Any], ...]:
    """A run's whole step list, refused rather than truncated when too long.

    Truncation would silently drop the end of a trace, which is the part that
    says how the run finished. A worker that streams more than the agreed
    maximum has broken the agreement and is told so.
    """

    if not isinstance(value, (list, tuple)):
        raise ProtocolError("steps must be a sequence")
    if len(value) > MAX_RUN_STEPS:
        raise ProtocolError(f"a run may report at most {MAX_RUN_STEPS} steps")
    return tuple(_run_step(item) for item in value)


def _live_step_map(value: Any) -> Mapping[str, Tuple[Mapping[str, Any], ...]]:
    """Normalise the live step stream, discarding anything unrecognised.

    Every field here is worker-supplied. The shape is fixed, the kind comes
    from a closed vocabulary, text is truncated, and numbers are coerced or
    dropped. What survives is a record the server can render without asking
    whether the worker was honest about its own structure.
    """

    if not isinstance(value, Mapping):
        raise ProtocolError("live steps must be a mapping")
    if len(value) > 64:
        raise ProtocolError("too many live step streams")
    result: Dict[str, Tuple[Mapping[str, Any], ...]] = {}
    for key, steps in value.items():
        job_id = _identifier(key, "job id")
        if not isinstance(steps, (list, tuple)):
            raise ProtocolError("live steps for a job must be a sequence")
        if len(steps) > MAX_LIVE_STEPS_PER_JOB:
            raise ProtocolError(
                f"a job may stream at most {MAX_LIVE_STEPS_PER_JOB} live steps"
            )
        normalised: list[Mapping[str, Any]] = []
        for step in steps:
            if not isinstance(step, Mapping):
                raise ProtocolError("each live step must be a mapping")
            kind = str(step.get("kind") or STEP_PHASE)
            entry: Dict[str, Any] = {
                "kind": kind if kind in STEP_KINDS else STEP_PHASE,
                "phase": normalise_phase(step.get("phase")),
                "text": str(step.get("text") or "")[:MAX_LIVE_STEP_TEXT],
            }
            tool = step.get("tool")
            if isinstance(tool, str) and tool:
                entry["tool"] = tool[:128]
            if "ok" in step:
                entry["ok"] = bool(step.get("ok"))
            elapsed = step.get("elapsed_ms")
            if isinstance(elapsed, int) and not isinstance(elapsed, bool) and 0 <= elapsed:
                entry["elapsed_ms"] = min(elapsed, 86_400_000)
            at = step.get("at")
            if isinstance(at, (int, float)) and not isinstance(at, bool) and math.isfinite(at):
                entry["at"] = float(at)
            normalised.append(MappingProxyType(entry))
        result[job_id] = tuple(normalised)
    return MappingProxyType(result)


#: Artifact bounds. Artifacts ride the completion payload, whose whole body is
#: capped at 1 MB by the server; these caps keep a well-formed result inside
#: that with room for output and usage.
MAX_ARTIFACTS = 8
MAX_ARTIFACT_NAME = 120
MAX_ARTIFACT_BASE64 = 600_000
_ARTIFACT_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9 ._-]{0,118}$")
_BASE64_RE = re.compile(r"^[A-Za-z0-9+/=\r\n]*$")


def _artifact(value: Any) -> Mapping[str, Any]:
    """Validate one artifact mapping from an untrusted worker.

    The name is confined to a conservative filename alphabet because it will
    become a path component on the server; the payload is confined to base64
    because it will be decoded and written to disk.
    """

    if not isinstance(value, Mapping):
        raise ProtocolError("artifact must be an object")
    name = value.get("name")
    if not isinstance(name, str) or not _ARTIFACT_NAME_RE.fullmatch(name):
        raise ProtocolError("artifact name must be a plain filename")
    if ".." in name:
        raise ProtocolError("artifact name must not contain path traversal")
    media_type = value.get("media_type")
    if not isinstance(media_type, str) or not media_type or len(media_type) > 120:
        raise ProtocolError("artifact media_type must be short text")
    encoded = value.get("base64")
    if not isinstance(encoded, str) or len(encoded) > MAX_ARTIFACT_BASE64:
        raise ProtocolError("artifact payload is missing or too large")
    if not _BASE64_RE.fullmatch(encoded):
        raise ProtocolError("artifact payload is not base64")
    size = value.get("size_bytes", 0)
    if isinstance(size, bool) or not isinstance(size, int) or size < 0:
        raise ProtocolError("artifact size_bytes must be a non-negative integer")
    return MappingProxyType(
        {
            "name": name,
            "media_type": media_type,
            "base64": encoded,
            "size_bytes": size,
        }
    )


@dataclass(frozen=True)
class JobResult:
    job_id: str
    output: str
    usage: Mapping[str, Any] = field(default_factory=dict)
    energy: Mapping[str, Any] = field(default_factory=dict)
    tool_events: Tuple[Mapping[str, Any], ...] = ()
    #: Files produced inside the run's sandbox by library skills, carried back
    #: so the user's server session can store and offer them. Bounded and
    #: validated: names are plain filenames, payloads are base64.
    artifacts: Tuple[Mapping[str, Any], ...] = ()
    #: The steps the agent took, in order, kept after the run rather than only
    #: streamed during it. The live stream on :class:`AgentSnapshot` shows a
    #: run in flight and is gone the moment it finishes, which left a completed
    #: turn showing the model's final text and nothing about how it got there.
    #: Same closed vocabulary, same bounds, same escaping.
    steps: Tuple[Mapping[str, Any], ...] = ()
    error: Optional[str] = None

    def __post_init__(self) -> None:
        object.__setattr__(self, "job_id", _identifier(self.job_id, "job id"))
        object.__setattr__(self, "output", _text(self.output, "output", 1_000_000, allow_empty=True))
        object.__setattr__(self, "usage", _mapping(self.usage))
        object.__setattr__(self, "energy", _mapping(self.energy))
        events = tuple(MappingProxyType(_plain(event)) for event in self.tool_events)
        if len(events) > 64:
            raise ProtocolError("too many tool events")
        object.__setattr__(self, "tool_events", events)
        artifacts = tuple(_artifact(item) for item in self.artifacts)
        if len(artifacts) > MAX_ARTIFACTS:
            raise ProtocolError("too many artifacts")
        if sum(len(item["base64"]) for item in artifacts) > MAX_ARTIFACT_BASE64:
            raise ProtocolError("artifacts exceed the total payload budget")
        object.__setattr__(self, "artifacts", artifacts)
        object.__setattr__(self, "steps", _run_steps(self.steps))
        if self.error is not None:
            object.__setattr__(self, "error", _text(self.error, "error", 4096, allow_empty=True))

    def to_dict(self) -> Dict[str, Any]:
        return {
            "job_id": self.job_id,
            "output": self.output,
            "usage": _plain(self.usage),
            "energy": _plain(self.energy),
            "tool_events": [_plain(v) for v in self.tool_events],
            "artifacts": [dict(v) for v in self.artifacts],
            "steps": [_plain(v) for v in self.steps],
            "error": self.error,
        }

    @classmethod
    def from_dict(cls, value: Mapping[str, Any]) -> "JobResult":
        return cls(
            job_id=value.get("job_id", ""),
            output=value.get("output", ""),
            usage=value.get("usage") or {},
            energy=value.get("energy") or {},
            tool_events=tuple(value.get("tool_events", ())),
            artifacts=tuple(value.get("artifacts", ())),
            steps=tuple(value.get("steps", ())),
            error=value.get("error"),
        )