"""Host tunnel frame schema for ``omnigent host``. Host-specific frame kinds, all JSON (see :class:`HostFrameKind`), plus reuse of ``PingFrame``/``PongFrame`` from the runner tunnel for keepalive. Host frames carry only control messages (launch/stop runner requests and their results). They do NOT carry HTTP request/response traffic — runners connect directly to the server with their own tunnels. This module is intentionally separate from the runner tunnel's ``frames.py`` to keep the two protocols partitioned. The runner module has a closed ``FrameKind`` enum and ``decode_frame`` match statement that handles all runner frame kinds. Adding host kinds there would force runner-side decoders to handle frames they never see. """ from __future__ import annotations import json from dataclasses import dataclass, field from enum import Enum from typing import Any HarnessAvailability = bool | str # Structured error code carried in ``HostLaunchRunnerResultFrame.error_code`` # when the host refuses a launch because the session's harness is not # configured on that machine (CLI missing or no default credential). Shared # by the daemon (producer), server (maps it to # ``ErrorCode.HARNESS_NOT_CONFIGURED``), and tests. HARNESS_NOT_CONFIGURED_ERROR_CODE = "harness_not_configured" class HostFrameKind(str, Enum): """All host frame kinds; the value is the JSON wire string.""" HELLO = "host.hello" LAUNCH_RUNNER = "host.launch_runner" LAUNCH_RUNNER_RESULT = "host.launch_runner_result" STOP_RUNNER = "host.stop_runner" STOP_RUNNER_RESULT = "host.stop_runner_result" RUNNER_EXITED = "host.runner_exited" STAT = "host.stat" STAT_RESULT = "host.stat_result" LIST_DIR = "host.list_dir" LIST_DIR_RESULT = "host.list_dir_result" CREATE_WORKTREE = "host.create_worktree" CREATE_WORKTREE_RESULT = "host.create_worktree_result" REMOVE_WORKTREE = "host.remove_worktree" REMOVE_WORKTREE_RESULT = "host.remove_worktree_result" LIST_WORKTREES = "host.list_worktrees" LIST_WORKTREES_RESULT = "host.list_worktrees_result" CREATE_DIR = "host.create_dir" CREATE_DIR_RESULT = "host.create_dir_result" # ── Frame dataclasses ──────────────────────────────────── @dataclass class HostHelloFrame: """Host's first frame on a fresh tunnel. :param version: Host software version, e.g. ``"0.1.0"``. :param frame_protocol_version: Wire-protocol major. Server refuses on major mismatch. :param name: Human-readable host name from ``config.yaml``, e.g. ``"corey-laptop"``. :param runners: Runner IDs currently alive on this host. Enables state reconciliation on reconnect — the server diffs this against sessions in the DB. :param configured_harnesses: Per-harness readiness on this machine, e.g. ``{"claude-sdk": True, "codex": False}`` (see ``omnigent.onboarding.harness_readiness``). Keys cover every accepted harness spelling. ``None`` means unknown (an older host that doesn't report it) — never treat ``None`` as "nothing is configured". Recomputed on each (re)connect; the launch-time check is authoritative. """ version: str frame_protocol_version: int name: str runners: list[str] = field(default_factory=list) configured_harnesses: dict[str, HarnessAvailability] | None = None @dataclass class HostLaunchRunnerFrame: """Server → host: spawn a new runner process. :param request_id: Unique ID for correlating the result, e.g. ``"req_abc123"``. :param binding_token: Secret token the runner must present when connecting. The server derives ``runner_id`` from this via ``token_bound_runner_id()``. :param workspace: Absolute path on the host machine to use as the runner's working directory, e.g. ``"/Users/corey/projects/frontend"``. :param session_id: Conversation/session ID the runner is being launched for, e.g. ``"conv_abc123"``. ``None`` means an older server did not include it. :param harness: Canonical harness the session will run, e.g. ``"claude-sdk"``. The host checks it is configured before spawning and refuses with :data:`HARNESS_NOT_CONFIGURED_ERROR_CODE` when not. ``None`` (older server, or no resolvable harness) skips the check — fail open. """ request_id: str binding_token: str workspace: str session_id: str | None = None harness: str | None = None @dataclass class HostLaunchRunnerResultFrame: """Host → server: outcome of a launch request. :param request_id: Correlates to the :class:`HostLaunchRunnerFrame`, e.g. ``"req_abc123"``. :param status: ``"launched"`` or ``"failed"``. :param runner_id: Runner ID derived from the binding token. Confirms the host spawned the expected runner. ``None`` when ``status`` is ``"failed"``. :param error: Error message when ``status`` is ``"failed"``, e.g. ``"workspace path does not exist"``. ``None`` on success. :param error_code: Machine-readable failure category when ``status`` is ``"failed"``, e.g. :data:`HARNESS_NOT_CONFIGURED_ERROR_CODE`. ``None`` for uncategorized failures and on success (and always from older hosts that don't send it). """ request_id: str status: str runner_id: str | None = None error: str | None = None error_code: str | None = None @dataclass class HostStopRunnerFrame: """Server → host: terminate a runner process. :param request_id: Unique ID for correlating the result, e.g. ``"req_def456"``. :param runner_id: Runner to stop, e.g. ``"runner_token_abc123..."``. """ request_id: str runner_id: str @dataclass class HostStopRunnerResultFrame: """Host → server: outcome of a stop request. :param request_id: Correlates to the :class:`HostStopRunnerFrame`, e.g. ``"req_def456"``. :param status: ``"stopped"`` or ``"failed"``. :param error: Error message when ``status`` is ``"failed"``. ``None`` on success. """ request_id: str status: str error: str | None = None @dataclass class HostRunnerExitedFrame: """Host → server: a spawned runner process died unexpectedly. One-way report (no result frame). The host daemon watches every runner it spawns; when one exits without a ``host.stop_runner`` request, the daemon composes a human-readable error — exit code plus the tail of the runner's captured log — and reports it here. The server stashes it so the runner status endpoint can answer "offline, and here is why" — a client waiting for the runner to connect fails fast with the actual cause instead of polling to a timeout and pointing the user at a log directory on the host. :param runner_id: The runner that died, e.g. ``"runner_abc123..."``. :param error: Human-readable cause, e.g. ``"runner process exited with code 1 (log: ~/...) ..."``, including the trailing lines of the runner's log. """ runner_id: str error: str @dataclass class HostStatFrame: """Server → host: stat a path on the host's filesystem. Used by session-create validation to verify that a workspace path (or an agent's ``os_env.cwd`` boundary) exists and is a directory before storing the session row. Single round-trip; no directory walking. :param request_id: Unique ID for correlating the result, e.g. ``"req_stat_1"``. :param path: Absolute path on the host (e.g. ``"/Users/corey/universe"``) OR a tilde-prefixed path (``"~/foo"``). The host expands ``~`` against its own process owner's home directory before stating. Only the host knows its own ``HOME`` — the server never expands tildes itself. """ request_id: str path: str @dataclass class HostStatResultFrame: """Host → server: outcome of a stat request. :param request_id: Correlates to the :class:`HostStatFrame`. :param status: ``"ok"`` or ``"failed"``. ``"failed"`` is reserved for unexpected I/O errors (e.g. EIO); EACCES and ENOENT both produce ``status: "ok", exists: false`` so the caller can treat them uniformly. Validation messages distinguishing missing-vs-unreadable can be added later if users find the collapse confusing — see designs/SESSION_WORKSPACE_SELECTION.md. :param exists: ``True`` when the path exists, is accessible to the host process, and (for symlinks) the target also exists. ``False`` for non-existent paths, dangling symlinks, and permission-denied paths. :param type: ``"directory"``, ``"file"``, or ``"other"``. Reflects the **target's** type after symlink resolution — a symlink to a directory returns ``"directory"``, never ``"symlink"``. ``None`` when ``exists`` is ``False``. :param canonical_path: Absolute, normalized realpath, e.g. ``"/Users/corey/universe"``. ``None`` when ``exists`` is ``False``. The server stores this on the session row instead of the user's input so symlinks cannot smuggle a workspace out of an agent's ``os_env.cwd`` boundary. :param error: Filesystem error message when ``status`` is ``"failed"``. ``None`` on success (including the ``exists: false`` case). """ request_id: str status: str exists: bool = False type: str | None = None canonical_path: str | None = None error: str | None = None @dataclass class HostListDirEntry: """A single entry in a host.list_dir result. Mirrors the runner's ``FilesystemEntry`` shape so the Web UI's existing tree component can consume host browse results without a different mapping. :param name: Basename of the entry, e.g. ``"src"``. :param path: Absolute path on the host, e.g. ``"/Users/corey/universe/src"``. The host returns absolute paths so the Web UI can address each entry directly via the same REST endpoint without re-resolving relatives. :param type: ``"directory"``, ``"file"``, or ``"other"``. Reflects the target type after symlink resolution; symlinks themselves are not surfaced (consistent with ``host.stat_result``). :param bytes: File size for regular files; ``None`` for directories and other types. Lets the UI render sizes without an extra stat per entry. :param modified_at: Unix epoch seconds of last modification, e.g. ``1779980000``. Drives "modified" timestamps in the directory tree. """ name: str path: str type: str bytes: int | None modified_at: int @dataclass class HostListDirFrame: """Server → host: list contents of a directory on the host. Used by ``GET /v1/hosts/{id}/filesystem/{path}`` to render the directory picker before any runner exists. The host owns ``~`` resolution; the server passes whatever the user supplied (or ``~`` when the REST path is empty). :param request_id: Unique ID for correlating the result, e.g. ``"req_list_1"``. :param path: Absolute or tilde-prefixed directory path, e.g. ``"/Users/corey/projects"`` or ``"~/projects"``. Same rules as ``host.stat`` — the host expands ``~`` against its own process owner's home. :param limit: Maximum entries to return per page, e.g. ``20``. Pagination is in-memory at the host since most directories fit easily in one page. :param after: Optional cursor (entry ``path``) for forward pagination. ``None`` returns the first page. :param before: Optional cursor for backward pagination. ``None`` paginates forward only. """ request_id: str path: str limit: int = 20 after: str | None = None before: str | None = None @dataclass class HostListDirResultFrame: """Host → server: outcome of a list_dir request. :param request_id: Correlates to the :class:`HostListDirFrame`, e.g. ``"req_list_1"``. :param status: ``"ok"`` or ``"failed"``. ``"failed"`` is reserved for unexpected I/O errors; missing path collapses to a normal ``"ok"`` with an empty entries list and a descriptive error (the route layer maps these into 404). :param entries: Directory contents, possibly paginated. Empty list when the directory is empty or when the path doesn't exist (callers should check ``error`` to distinguish). :param has_more: ``True`` when more pages exist; ``False`` for the last page (or when entries are empty). :param error: Filesystem error, e.g. ``"path does not exist"`` or ``"permission denied"``. ``None`` on success. Populated even when ``status`` is ``"ok"`` so a missing path still carries a useful message into the REST response. """ request_id: str status: str entries: list[HostListDirEntry] = field(default_factory=list) has_more: bool = False error: str | None = None @dataclass class HostCreateWorktreeFrame: """Server → host: create a git worktree for a new branch. See designs/SESSION_GIT_WORKTREE.md. :param request_id: Correlates the result, e.g. ``"req_wt_1"``. :param repo_path: Absolute path inside the source repo (the picked dir or a subdir), e.g. ``"/Users/alice/myrepo"``. :param branch_name: New branch to create, e.g. ``"feature/login"``. :param base_branch: Optional base ref, e.g. ``"main"``. ``None`` branches from ``HEAD``. """ request_id: str repo_path: str branch_name: str base_branch: str | None = None @dataclass class HostCreateWorktreeResultFrame: """Host → server: outcome of a create-worktree request. :param request_id: Correlates to the :class:`HostCreateWorktreeFrame`, e.g. ``"req_wt_1"``. :param status: ``"ok"`` or ``"failed"``. :param worktree_path: Created worktree directory (stored as the session ``workspace``), e.g. ``"/Users/alice/myrepo-worktrees/feature-login"``. ``None`` on failure. :param branch: Branch checked out, e.g. ``"feature/login"``. ``None`` on failure. :param error: Error message when ``status`` is ``"failed"``, e.g. ``"not a git repository"``. ``None`` on success. """ request_id: str status: str worktree_path: str | None = None branch: str | None = None error: str | None = None @dataclass class HostRemoveWorktreeFrame: """Server → host: remove a git worktree (opt-in session cleanup). The host derives the main repo from ``worktree_path`` itself, so no repo path is carried. See designs/SESSION_GIT_WORKTREE.md. :param request_id: Correlates the result, e.g. ``"req_wt_rm_1"``. :param worktree_path: Worktree directory to remove (the stored session ``workspace``), e.g. ``"/Users/alice/myrepo-worktrees/feature-login"``. :param branch: Branch to delete when ``delete_branch`` is ``True``, e.g. ``"feature/login"``. ``None`` skips deletion. :param delete_branch: When ``True``, ``git branch -D`` after removing the directory; when ``False``, remove only the directory. """ request_id: str worktree_path: str branch: str | None = None delete_branch: bool = False @dataclass class HostRemoveWorktreeResultFrame: """Host → server: outcome of a remove-worktree request. :param request_id: Correlates to the :class:`HostRemoveWorktreeFrame`, e.g. ``"req_wt_rm_1"``. :param status: ``"ok"`` or ``"failed"``. :param error: Error message when ``status`` is ``"failed"``. ``None`` on success. """ request_id: str status: str error: str | None = None @dataclass class HostListWorktreesFrame: """Server → host: list the git worktrees of a repository. Backs ``GET /v1/hosts/{id}/worktrees``, used by the Web UI's new-session worktree picker to show worktrees a session can start in directly. Read-only; the host derives the main work tree from ``repo_path`` (so a linked worktree resolves the same list). :param request_id: Correlates the result, e.g. ``"req_wt_ls_1"``. :param repo_path: Absolute path inside the repo (the picked dir or a subdir), e.g. ``"/Users/alice/myrepo"``. """ request_id: str repo_path: str @dataclass class HostListWorktreesResultFrame: """Host → server: outcome of a list-worktrees request. :param request_id: Correlates to the :class:`HostListWorktreesFrame`, e.g. ``"req_wt_ls_1"``. :param status: ``"ok"`` or ``"failed"``. :param worktrees: One dict per worktree with keys ``path`` (str), ``branch`` (str | None), ``is_main`` (bool), ``detached`` (bool), main first. ``None`` on failure. :param error: Error message when ``status`` is ``"failed"``, e.g. ``"not a git repository"``. ``None`` on success. """ request_id: str status: str worktrees: list[dict[str, Any]] | None = None error: str | None = None @dataclass class HostCreateDirFrame: """Server → host: create a new directory on the host. Backs ``POST /v1/hosts/{id}/directories``, used by the Web UI's workspace picker so a user can make a fresh folder to start a session in without dropping to a terminal. The host owns ``~`` resolution, same rules as ``host.list_dir`` / ``host.stat``. :param request_id: Correlates the result, e.g. ``"req_mkdir_1"``. :param path: Absolute or tilde-prefixed directory path to create, e.g. ``"/Users/corey/projects/new-app"`` or ``"~/scratch"``. Missing parent directories are created (``os.makedirs``). """ request_id: str path: str @dataclass class HostCreateDirResultFrame: """Host → server: outcome of a create-dir request. :param request_id: Correlates to the :class:`HostCreateDirFrame`, e.g. ``"req_mkdir_1"``. :param status: ``"ok"`` or ``"failed"``. ``"failed"`` is reserved for unexpected I/O errors; an expected filesystem error (the directory already exists, permission denied, a parent path component is a file) collapses to ``"ok"`` with a descriptive ``error`` so the route layer can map it to a 409 rather than a 500 — same posture as ``host.list_dir`` for a missing path. :param path: Absolute path of the created directory, e.g. ``"/Users/corey/projects/new-app"``. ``None`` when the directory was not created. :param error: Filesystem error, e.g. ``"directory already exists"`` or ``"permission denied"``. ``None`` on success. """ request_id: str status: str path: str | None = None error: str | None = None HostFrame = ( HostHelloFrame | HostLaunchRunnerFrame | HostLaunchRunnerResultFrame | HostStopRunnerFrame | HostStopRunnerResultFrame | HostRunnerExitedFrame | HostStatFrame | HostStatResultFrame | HostListDirFrame | HostListDirResultFrame | HostCreateWorktreeFrame | HostCreateWorktreeResultFrame | HostRemoveWorktreeFrame | HostRemoveWorktreeResultFrame | HostListWorktreesFrame | HostListWorktreesResultFrame | HostCreateDirFrame | HostCreateDirResultFrame ) # ── Encode / decode ────────────────────────────────────── def _encode_payload(payload: dict[str, Any]) -> str: """Serialize a frame payload, injecting the active trace context. Centralized so every host frame carries a W3C ``traceparent`` (and ``tracestate`` when set) whenever it is encoded inside an active span — the host tunnel is a JSON-frame transport no OTel auto-instrumentor can see, so this is how the Host Daemon ↔ Server boundary joins the distributed trace. When no span is active the payload is unchanged. Decoders ignore the extra envelope keys, so this stays wire-compatible with peers that do not read them. :param payload: The frame fields about to be serialized. :returns: The JSON wire string. """ from omnigent.runtime import telemetry # Record the outbound body on the active span (redacted, gated by # content capture) before injecting propagation keys, so the span # shows exactly what this side sent. telemetry.record_message_payload(payload) telemetry.inject_trace_context(payload) return json.dumps(payload) def encode_host_frame(frame: HostFrame) -> str: """Serialize a host frame to its JSON wire form. :param frame: The host frame dataclass to encode. :returns: JSON string for the WebSocket text message. :raises TypeError: If ``frame`` is not a known host frame type. """ if isinstance(frame, HostHelloFrame): return _encode_payload( { "kind": HostFrameKind.HELLO.value, "version": frame.version, "frame_protocol_version": frame.frame_protocol_version, "name": frame.name, "runners": list(frame.runners), "configured_harnesses": frame.configured_harnesses, } ) if isinstance(frame, HostLaunchRunnerFrame): return _encode_payload( { "kind": HostFrameKind.LAUNCH_RUNNER.value, "request_id": frame.request_id, "binding_token": frame.binding_token, "workspace": frame.workspace, "session_id": frame.session_id, "harness": frame.harness, } ) if isinstance(frame, HostLaunchRunnerResultFrame): return _encode_payload( { "kind": HostFrameKind.LAUNCH_RUNNER_RESULT.value, "request_id": frame.request_id, "status": frame.status, "runner_id": frame.runner_id, "error": frame.error, "error_code": frame.error_code, } ) if isinstance(frame, HostStopRunnerFrame): return _encode_payload( { "kind": HostFrameKind.STOP_RUNNER.value, "request_id": frame.request_id, "runner_id": frame.runner_id, } ) if isinstance(frame, HostStopRunnerResultFrame): return _encode_payload( { "kind": HostFrameKind.STOP_RUNNER_RESULT.value, "request_id": frame.request_id, "status": frame.status, "error": frame.error, } ) if isinstance(frame, HostRunnerExitedFrame): return _encode_payload( { "kind": HostFrameKind.RUNNER_EXITED.value, "runner_id": frame.runner_id, "error": frame.error, } ) if isinstance(frame, HostStatFrame): return _encode_payload( { "kind": HostFrameKind.STAT.value, "request_id": frame.request_id, "path": frame.path, } ) if isinstance(frame, HostStatResultFrame): return _encode_payload( { "kind": HostFrameKind.STAT_RESULT.value, "request_id": frame.request_id, "status": frame.status, "exists": frame.exists, "type": frame.type, "canonical_path": frame.canonical_path, "error": frame.error, } ) if isinstance(frame, HostListDirFrame): return _encode_payload( { "kind": HostFrameKind.LIST_DIR.value, "request_id": frame.request_id, "path": frame.path, "limit": frame.limit, "after": frame.after, "before": frame.before, } ) if isinstance(frame, HostListDirResultFrame): return _encode_payload( { "kind": HostFrameKind.LIST_DIR_RESULT.value, "request_id": frame.request_id, "status": frame.status, "entries": [ { "name": entry.name, "path": entry.path, "type": entry.type, "bytes": entry.bytes, "modified_at": entry.modified_at, } for entry in frame.entries ], "has_more": frame.has_more, "error": frame.error, } ) if isinstance(frame, HostCreateWorktreeFrame): return _encode_payload( { "kind": HostFrameKind.CREATE_WORKTREE.value, "request_id": frame.request_id, "repo_path": frame.repo_path, "branch_name": frame.branch_name, "base_branch": frame.base_branch, } ) if isinstance(frame, HostCreateWorktreeResultFrame): return _encode_payload( { "kind": HostFrameKind.CREATE_WORKTREE_RESULT.value, "request_id": frame.request_id, "status": frame.status, "worktree_path": frame.worktree_path, "branch": frame.branch, "error": frame.error, } ) if isinstance(frame, HostRemoveWorktreeFrame): return _encode_payload( { "kind": HostFrameKind.REMOVE_WORKTREE.value, "request_id": frame.request_id, "worktree_path": frame.worktree_path, "branch": frame.branch, "delete_branch": frame.delete_branch, } ) if isinstance(frame, HostRemoveWorktreeResultFrame): return _encode_payload( { "kind": HostFrameKind.REMOVE_WORKTREE_RESULT.value, "request_id": frame.request_id, "status": frame.status, "error": frame.error, } ) if isinstance(frame, HostListWorktreesFrame): return _encode_payload( { "kind": HostFrameKind.LIST_WORKTREES.value, "request_id": frame.request_id, "repo_path": frame.repo_path, } ) if isinstance(frame, HostListWorktreesResultFrame): return _encode_payload( { "kind": HostFrameKind.LIST_WORKTREES_RESULT.value, "request_id": frame.request_id, "status": frame.status, "worktrees": frame.worktrees, "error": frame.error, } ) if isinstance(frame, HostCreateDirFrame): return _encode_payload( { "kind": HostFrameKind.CREATE_DIR.value, "request_id": frame.request_id, "path": frame.path, } ) if isinstance(frame, HostCreateDirResultFrame): return _encode_payload( { "kind": HostFrameKind.CREATE_DIR_RESULT.value, "request_id": frame.request_id, "status": frame.status, "path": frame.path, "error": frame.error, } ) raise TypeError(f"unknown host frame type: {type(frame).__name__}") def decode_host_frame(text: str) -> HostFrame: """Parse a JSON wire frame back into its host frame dataclass. :param text: Raw JSON frame text from the WebSocket. :returns: The typed host frame dataclass. :raises ValueError: On malformed JSON, missing ``kind``, unknown kind, or missing required fields. """ msg = _parse_frame_object(text) kind = _parse_host_frame_kind(msg) return _decode_known_host_frame(kind, msg) def _parse_frame_object(text: str) -> dict[str, Any]: """Parse a JSON frame object. :param text: Raw JSON frame text. :returns: Decoded frame object. :raises ValueError: If the payload is not a JSON object. """ try: msg = json.loads(text) except json.JSONDecodeError as exc: raise ValueError(f"frame is not valid JSON: {exc}") from exc if not isinstance(msg, dict): raise ValueError(f"frame must be a JSON object, got {type(msg).__name__}") return msg def _parse_host_frame_kind(msg: dict[str, Any]) -> HostFrameKind: """Parse the host frame kind discriminator. :param msg: Decoded frame object. :returns: Host frame kind enum. :raises ValueError: If ``kind`` is missing or unknown. """ kind = msg.get("kind") if not isinstance(kind, str): raise ValueError("frame missing 'kind' field") try: return HostFrameKind(kind) except ValueError as exc: raise ValueError(f"unknown host frame kind: {kind!r}") from exc def _decode_known_host_frame( kind: HostFrameKind, msg: dict[str, Any], ) -> HostFrame: """Decode a host frame with a validated kind. :param kind: Parsed host frame kind. :param msg: Decoded frame object. :returns: The typed host frame dataclass. :raises ValueError: If the kind is unexpectedly unhandled. """ match kind: case HostFrameKind.HELLO: return _decode_host_hello(msg) case HostFrameKind.LAUNCH_RUNNER: return _decode_launch_runner(msg) case HostFrameKind.LAUNCH_RUNNER_RESULT: return _decode_launch_runner_result(msg) case HostFrameKind.STOP_RUNNER: return _decode_stop_runner(msg) case HostFrameKind.STOP_RUNNER_RESULT: return _decode_stop_runner_result(msg) case HostFrameKind.RUNNER_EXITED: return _decode_runner_exited(msg) case HostFrameKind.STAT: return _decode_stat(msg) case HostFrameKind.STAT_RESULT: return _decode_stat_result(msg) case HostFrameKind.LIST_DIR: return _decode_list_dir(msg) case HostFrameKind.LIST_DIR_RESULT: return _decode_list_dir_result(msg) case HostFrameKind.CREATE_WORKTREE: return _decode_create_worktree(msg) case HostFrameKind.CREATE_WORKTREE_RESULT: return _decode_create_worktree_result(msg) case HostFrameKind.REMOVE_WORKTREE: return _decode_remove_worktree(msg) case HostFrameKind.REMOVE_WORKTREE_RESULT: return _decode_remove_worktree_result(msg) case HostFrameKind.LIST_WORKTREES: return _decode_list_worktrees(msg) case HostFrameKind.LIST_WORKTREES_RESULT: return _decode_list_worktrees_result(msg) case HostFrameKind.CREATE_DIR: return _decode_create_dir(msg) case HostFrameKind.CREATE_DIR_RESULT: return _decode_create_dir_result(msg) raise ValueError(f"unhandled host frame kind: {kind.value!r}") # pragma: no cover def _decode_host_hello(msg: dict[str, Any]) -> HostHelloFrame: """Decode a host hello frame. :param msg: Decoded frame object. :returns: Typed host hello frame. """ return HostHelloFrame( version=_required_str(msg, "version"), frame_protocol_version=_required_int(msg, "frame_protocol_version"), name=_required_str(msg, "name"), runners=_optional_str_list(msg, "runners"), configured_harnesses=_optional_str_availability_map(msg, "configured_harnesses"), ) def _decode_launch_runner(msg: dict[str, Any]) -> HostLaunchRunnerFrame: """Decode a launch-runner frame. :param msg: Decoded frame object. :returns: Typed launch-runner frame. """ return HostLaunchRunnerFrame( request_id=_required_str(msg, "request_id"), binding_token=_required_str(msg, "binding_token"), workspace=_required_str(msg, "workspace"), session_id=_optional_nullable_str(msg, "session_id"), harness=_optional_nullable_str(msg, "harness"), ) def _decode_launch_runner_result( msg: dict[str, Any], ) -> HostLaunchRunnerResultFrame: """Decode a launch-runner-result frame. :param msg: Decoded frame object. :returns: Typed launch-runner-result frame. """ return HostLaunchRunnerResultFrame( request_id=_required_str(msg, "request_id"), status=_required_str(msg, "status"), runner_id=_optional_nullable_str(msg, "runner_id"), error=_optional_nullable_str(msg, "error"), error_code=_optional_nullable_str(msg, "error_code"), ) def _decode_stop_runner(msg: dict[str, Any]) -> HostStopRunnerFrame: """Decode a stop-runner frame. :param msg: Decoded frame object. :returns: Typed stop-runner frame. """ return HostStopRunnerFrame( request_id=_required_str(msg, "request_id"), runner_id=_required_str(msg, "runner_id"), ) def _decode_stop_runner_result( msg: dict[str, Any], ) -> HostStopRunnerResultFrame: """Decode a stop-runner-result frame. :param msg: Decoded frame object. :returns: Typed stop-runner-result frame. """ return HostStopRunnerResultFrame( request_id=_required_str(msg, "request_id"), status=_required_str(msg, "status"), error=_optional_nullable_str(msg, "error"), ) def _decode_runner_exited(msg: dict[str, Any]) -> HostRunnerExitedFrame: """Decode a host.runner_exited report frame. :param msg: Decoded frame object. :returns: Typed host.runner_exited frame. """ return HostRunnerExitedFrame( runner_id=_required_str(msg, "runner_id"), error=_required_str(msg, "error"), ) def _decode_stat(msg: dict[str, Any]) -> HostStatFrame: """Decode a host.stat request frame. :param msg: Decoded frame object. :returns: Typed host.stat frame. """ return HostStatFrame( request_id=_required_str(msg, "request_id"), path=_required_str(msg, "path"), ) def _decode_stat_result(msg: dict[str, Any]) -> HostStatResultFrame: """Decode a host.stat_result frame. :param msg: Decoded frame object. :returns: Typed host.stat_result frame. """ return HostStatResultFrame( request_id=_required_str(msg, "request_id"), status=_required_str(msg, "status"), exists=_required_bool(msg, "exists"), type=_optional_nullable_str(msg, "type"), canonical_path=_optional_nullable_str(msg, "canonical_path"), error=_optional_nullable_str(msg, "error"), ) def _decode_list_dir(msg: dict[str, Any]) -> HostListDirFrame: """Decode a host.list_dir request frame. :param msg: Decoded frame object. :returns: Typed host.list_dir frame. """ limit_value = msg.get("limit", 20) if not isinstance(limit_value, int) or isinstance(limit_value, bool): raise ValueError("frame field must be an int: 'limit'") return HostListDirFrame( request_id=_required_str(msg, "request_id"), path=_required_str(msg, "path"), limit=limit_value, after=_optional_nullable_str(msg, "after"), before=_optional_nullable_str(msg, "before"), ) def _decode_list_dir_result(msg: dict[str, Any]) -> HostListDirResultFrame: """Decode a host.list_dir_result frame. :param msg: Decoded frame object. :returns: Typed host.list_dir_result frame. """ raw_entries = msg.get("entries", []) if not isinstance(raw_entries, list): raise ValueError("frame field must be a list: 'entries'") entries: list[HostListDirEntry] = [] for raw in raw_entries: if not isinstance(raw, dict): raise ValueError("each entry in 'entries' must be a JSON object") entries.append(_decode_list_dir_entry(raw)) has_more = msg.get("has_more", False) if not isinstance(has_more, bool): raise ValueError("frame field must be a bool: 'has_more'") return HostListDirResultFrame( request_id=_required_str(msg, "request_id"), status=_required_str(msg, "status"), entries=entries, has_more=has_more, error=_optional_nullable_str(msg, "error"), ) def _decode_list_dir_entry(msg: dict[str, Any]) -> HostListDirEntry: """Decode a single entry in a host.list_dir_result. :param msg: Decoded entry object. :returns: Typed entry. :raises ValueError: When required fields are missing or wrong type. """ bytes_val = msg.get("bytes") if bytes_val is not None and (not isinstance(bytes_val, int) or isinstance(bytes_val, bool)): raise ValueError("entry field must be int or null: 'bytes'") modified_at = msg.get("modified_at") if not isinstance(modified_at, int) or isinstance(modified_at, bool): raise ValueError("entry field must be an int: 'modified_at'") return HostListDirEntry( name=_required_str(msg, "name"), path=_required_str(msg, "path"), type=_required_str(msg, "type"), bytes=bytes_val, modified_at=modified_at, ) def _decode_create_worktree(msg: dict[str, Any]) -> HostCreateWorktreeFrame: """Decode a host.create_worktree request frame. :param msg: Decoded frame object. :returns: Typed host.create_worktree frame. """ return HostCreateWorktreeFrame( request_id=_required_str(msg, "request_id"), repo_path=_required_str(msg, "repo_path"), branch_name=_required_str(msg, "branch_name"), base_branch=_optional_nullable_str(msg, "base_branch"), ) def _decode_create_worktree_result( msg: dict[str, Any], ) -> HostCreateWorktreeResultFrame: """Decode a host.create_worktree_result frame. :param msg: Decoded frame object. :returns: Typed host.create_worktree_result frame. """ return HostCreateWorktreeResultFrame( request_id=_required_str(msg, "request_id"), status=_required_str(msg, "status"), worktree_path=_optional_nullable_str(msg, "worktree_path"), branch=_optional_nullable_str(msg, "branch"), error=_optional_nullable_str(msg, "error"), ) def _decode_remove_worktree(msg: dict[str, Any]) -> HostRemoveWorktreeFrame: """Decode a host.remove_worktree request frame. :param msg: Decoded frame object. :returns: Typed host.remove_worktree frame. """ delete_branch = msg.get("delete_branch", False) if not isinstance(delete_branch, bool): raise ValueError("frame field must be a bool: 'delete_branch'") return HostRemoveWorktreeFrame( request_id=_required_str(msg, "request_id"), worktree_path=_required_str(msg, "worktree_path"), branch=_optional_nullable_str(msg, "branch"), delete_branch=delete_branch, ) def _decode_remove_worktree_result( msg: dict[str, Any], ) -> HostRemoveWorktreeResultFrame: """Decode a host.remove_worktree_result frame. :param msg: Decoded frame object. :returns: Typed host.remove_worktree_result frame. """ return HostRemoveWorktreeResultFrame( request_id=_required_str(msg, "request_id"), status=_required_str(msg, "status"), error=_optional_nullable_str(msg, "error"), ) def _decode_list_worktrees(msg: dict[str, Any]) -> HostListWorktreesFrame: """Decode a host.list_worktrees request frame. :param msg: Decoded frame object. :returns: Typed host.list_worktrees frame. """ return HostListWorktreesFrame( request_id=_required_str(msg, "request_id"), repo_path=_required_str(msg, "repo_path"), ) def _decode_list_worktrees_result( msg: dict[str, Any], ) -> HostListWorktreesResultFrame: """Decode a host.list_worktrees_result frame. :param msg: Decoded frame object. :returns: Typed host.list_worktrees_result frame. """ raw = msg.get("worktrees") if raw is not None: if not isinstance(raw, list): raise ValueError("frame field must be a list or null: 'worktrees'") for entry in raw: if not isinstance(entry, dict): raise ValueError("each entry in 'worktrees' must be a JSON object") return HostListWorktreesResultFrame( request_id=_required_str(msg, "request_id"), status=_required_str(msg, "status"), worktrees=raw, error=_optional_nullable_str(msg, "error"), ) def _decode_create_dir(msg: dict[str, Any]) -> HostCreateDirFrame: """Decode a host.create_dir request frame. :param msg: Decoded frame object. :returns: Typed host.create_dir frame. """ return HostCreateDirFrame( request_id=_required_str(msg, "request_id"), path=_required_str(msg, "path"), ) def _decode_create_dir_result(msg: dict[str, Any]) -> HostCreateDirResultFrame: """Decode a host.create_dir_result frame. :param msg: Decoded frame object. :returns: Typed host.create_dir_result frame. """ return HostCreateDirResultFrame( request_id=_required_str(msg, "request_id"), status=_required_str(msg, "status"), path=_optional_nullable_str(msg, "path"), error=_optional_nullable_str(msg, "error"), ) # ── Field validators ───────────────────────────────────── def _required_str(msg: dict[str, Any], key: str) -> str: """Return a required string field. :param msg: Decoded frame object. :param key: Field name, e.g. ``"version"``. :returns: The string value. :raises ValueError: If the field is missing or not a string. """ val = msg.get(key) if not isinstance(val, str): raise ValueError(f"frame missing required string field: {key!r}") return val def _required_int(msg: dict[str, Any], key: str) -> int: """Return a required integer field. :param msg: Decoded frame object. :param key: Field name, e.g. ``"frame_protocol_version"``. :returns: The integer value. :raises ValueError: If the field is missing or not an integer. """ val = msg.get(key) if not isinstance(val, int) or isinstance(val, bool): raise ValueError(f"frame missing required int field: {key!r}") return val def _required_bool(msg: dict[str, Any], key: str) -> bool: """Return a required boolean field. :param msg: Decoded frame object. :param key: Field name, e.g. ``"exists"``. :returns: The boolean value. :raises ValueError: If the field is missing or not a bool. """ val = msg.get(key) if not isinstance(val, bool): raise ValueError(f"frame missing required bool field: {key!r}") return val def _optional_str_list(msg: dict[str, Any], key: str) -> list[str]: """Return an optional list of strings. :param msg: Decoded frame object. :param key: Field name, e.g. ``"runners"``. :returns: A list of strings, empty when absent. :raises ValueError: If the field is not a string list. """ val = msg.get(key, []) if not isinstance(val, list) or not all(isinstance(item, str) for item in val): raise ValueError(f"frame field must be a list of strings: {key!r}") return list(val) def _optional_str_availability_map( msg: dict[str, Any], key: str ) -> dict[str, HarnessAvailability] | None: """Return an optional string→availability mapping field. Tolerant by design: absent, null, or non-mapping values all decode to ``None`` ("unknown") rather than raising, so an older or newer peer's hello never breaks the tunnel handshake. Entries with a non-string key or non-bool/string value are dropped for the same reason. :param msg: Decoded frame object. :param key: Field name, e.g. ``"configured_harnesses"``. :returns: The mapping, e.g. ``{"claude-sdk": True, "codex": "needs-auth"}``, or ``None`` when absent / null / not a JSON object. """ val = msg.get(key) if not isinstance(val, dict): return None return {k: v for k, v in val.items() if isinstance(k, str) and isinstance(v, (bool, str))} def _optional_nullable_str(msg: dict[str, Any], key: str) -> str | None: """Return an optional nullable string field. :param msg: Decoded frame object. :param key: Field name, e.g. ``"error"``. :returns: The string value, or ``None`` when absent or null. :raises ValueError: If the field is present and not a string or null. """ val = msg.get(key) if val is None: return None if not isinstance(val, str): raise ValueError(f"frame field must be a string or null: {key!r}") return val