File size: 4,534 Bytes
488d96b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | """Wire protocol vocabulary for the Python side of dsh-ptc-runtime-python.
Mirrors ``src/protocol.ts``. Frames travel on fd 3 as JSON-lines (one JSON
object per line). The host validates every inbound frame; this side trusts
host replies.
The wire uses the JSON key ``global`` (a Python keyword), so the frame
``TypedDict``s that carry it are declared with the functional syntax rather than
class bodies: a class attribute cannot be named ``global``, and a ``global_``
attribute would describe a key the wire never sends. Optional-field messages
pair a required base with a ``total=False`` subclass so a required field such as
``type`` cannot be dropped while ``value``/``error``/``truncated`` stay optional.
"""
from __future__ import annotations
from typing import Any, Literal, TypedDict, Union
# The protocol fd from the child's perspective. Node passes
# ``stdio: [pipe, pipe, pipe, pipe]`` so the fourth entry (fd 3) is the
# framed-JSON channel; stdout/stderr stay clear for the program's own output.
PROTOCOL_FD = 3
class ErrorClass(TypedDict):
"""A namespace's program-visible exception class: rejected calls raise its
instances carrying the failed member name on ``memberNameProperty``."""
name: str
memberNameProperty: str
# ``global`` is a Python keyword, so the required part is declared functionally
# to hold the real wire key; ``errorClass`` is optional per the TS `errorClass?`.
_NamespaceRequired = TypedDict("_NamespaceRequired", {"global": str, "names": "list[str]"})
class Namespace(_NamespaceRequired, total=False):
"""One binding namespace declaration: the ``global`` name, its function
``names``, and an optional program-visible ``errorClass`` for rejected calls."""
errorClass: ErrorClass
class BootMessage(TypedDict):
"""Host → child, first frame on fd 3. Carries every cap and the namespaces."""
type: Literal["boot"]
cpuSeconds: int
addressSpaceBytes: int
maxLogBytes: int
maxValueBytes: int
namespaces: "list[Namespace]"
class RunMessage(TypedDict):
"""Host → child, sent after ``boot-ack``. Carries only the program body."""
type: Literal["run"]
program: str
class BootAckMessage(TypedDict):
"""Child → host: resource limits applied, ready for the run message."""
type: Literal["boot-ack"]
# ``global`` wire key: whole message declared functionally, all fields required.
CallMessage = TypedDict(
"CallMessage",
{"type": Literal["call"], "id": int, "global": str, "name": str, "args": Any},
)
_LogMessageRequired = TypedDict("_LogMessageRequired", {"type": Literal["log"], "text": str})
class LogMessage(_LogMessageRequired, total=False):
"""Child → host: one captured text chunk, streamed eagerly.
``truncated`` is set only on the frame that IS the child ledger's truncation
marker (not program output), so the host stops capturing at the same point
the child did — mirrors the TS `truncated?`. ``open`` is set on a flushed unterminated line the host appends the next frame to (mirrors `open?`).
"""
truncated: bool
open: bool
class DoneErrorField(TypedDict):
"""Child → host: the failure carried on a ``done`` frame. ``kind`` is one of
the three the host validates; ``message`` is the traceback or diagnostic."""
kind: Literal["exception", "invalid-output", "output-limit"]
message: str
_DoneMessageRequired = TypedDict("_DoneMessageRequired", {"type": Literal["done"]})
class DoneMessage(_DoneMessageRequired, total=False):
"""Child → host: the program settled. ``value`` and ``error`` are optional per the TS mirror."""
value: Any
error: DoneErrorField
ChildToHost = Union[BootAckMessage, CallMessage, LogMessage, DoneMessage]
class ReplyOk(TypedDict):
type: Literal["reply"]
id: int
ok: Literal[True]
value: Any
class ReplyErr(TypedDict):
type: Literal["reply"]
id: int
ok: Literal[False]
message: str
ReplyMessage = Union[ReplyOk, ReplyErr]
# The host sends ``boot`` and ``run`` before any ``reply``, so the child-facing
# inbound union covers all three, not replies alone.
HostToChild = Union[BootMessage, RunMessage, ReplyMessage]
def log_truncation_marker(max_bytes: int) -> str:
"""Return the in-band marker for a log ledger that exhausted its budget.
Byte-identical text on both sides of the wire so a truncated run reads the
same however the cap was hit.
"""
return f"[dsh-ptc-runtime-python] log capture truncated at {max_bytes} bytes"
|