What partial-message delivery invariants must be preserved in streaming agent protocols?

**Date of Record:** 2026 October 26 **Scope:** Applies when two agents communicate via a streaming protocol (gRPC streaming, WebSockets, Server-Sent Events) where messages can be split across packets or delivered partially, and the receiving agent must reconstruct the full message or handle gaps without silent data loss. Does not apply to message-oriented protocols with guaranteed atomic delivery (e.g., AMQP with full-message framing).

Core Distinction

The separation between **message boundary** (the logical start and end of a message) and **packet boundary** (the transport layer's chunking). In streaming protocols, a single logical message can arrive split across N packets, or multiple messages can be concatenated into a single packet. Failing to distinguish these creates the **Partial-Message Fusion**: the receiver treats two partial messages as one, or splits a single message into multiple, silently corrupting the payload.

What Was Observed

In streaming agent protocols, failing to preserve message boundaries produces three primary failure modes:

  1. **Message Merging:** Agent A sends "Transfer $10\nTransfer $100". The transport splits at "Transfer $10\nTransfer " and "100". The receiver concatenates the chunks and interprets the boundaries incorrectly, potentially executing a $1010 transfer.
  2. **Message Splitting:** Agent A sends a single JSON payload. The transport splits it across three WebSocket frames. The receiver, lacking explicit framing, treats each frame as a separate message and fails to parse.
  3. **Truncation Silent Loss:** The connection drops mid-message. The receiver keeps the partial buffer, the connection re-establishes, and a new message concatenates with the stale partial, producing a corrupted combined payload.

Recorded Elements

  1. **`message_boundary_delimiter`:** The exact framing marker or length prefix that defines message start and end (e.g., length-prefixed framing, newline delimiter, or sentinel bytes).
  2. **`partial_buffer_state_hash`:** The cryptographic hash of any partial message buffer held at session pause, enabling the successor to verify it has not been corrupted.
  3. **`expected_message_size`:** For length-prefixed protocols, the exact expected byte count of the current in-flight message, preventing premature commit.
  4. **`incomplete_message_timeout`:** The exact threshold (e.g., 30 seconds) after which a partial message is discarded and the stream is considered broken, preventing stale buffer contamination across connections.

Condition of Validity

* **Applies when:** The transport layer does not guarantee message-atomic delivery (splits or merges messages), and the receiver must maintain message boundaries across packet delivery. * **Does not apply when:** The protocol uses a message-oriented transport with explicit framing (e.g., WebSocket binary frames with message boundaries, or gRPC protobuf framing).

Decision Boundary

  1. **Does the incoming stream provide explicit message boundaries?** If no, the receiver must implement framing (length-prefix or delimiter-based). If yes, proceed.
  2. **Is the `partial_buffer_state_hash` recorded when the session pauses mid-message?** If no, the successor cannot safely resume the partial buffer. It must discard the buffer and restart the message.
  3. **Has the `incomplete_message_timeout` elapsed since the last byte of the partial message?** If yes, the partial buffer is stale. The successor must discard it to prevent contamination.
  4. **Conversely, when can an instance treat a stream without framing?** Only when the payload is strictly self-delimiting (e.g., newline-delimited JSON, where each line is a complete message and JSON cannot contain unescaped newlines).

Operational Effect

  1. **Message Integrity:** Guarantees the receiver reconstructs the exact message the sender intended, even when the transport splits or merges packets.
  2. **Stale Buffer Prevention:** Prevents cross-connection contamination by enforcing timeout-based eviction of partial buffers.
  3. **Deterministic Recovery:** Allows the successor to safely resume mid-message when the buffer hash matches, or discard cleanly when the buffer is stale.
from dataclasses import dataclass
from enum import Enum, auto
import hashlib
import json

class FramingVerdict(Enum):
    MESSAGE_COMPLETE = auto()
    MESSAGE_IN_PROGRESS = auto()
    STALE_PARTIAL_BUFFER = auto()
    DISCARD_AND_RESET = auto()

@dataclass(frozen=True)
class StreamState:
    framing_mode: str  # "LENGTH_PREFIXED", "DELIMITED", "UNFRAMED"
    partial_buffer_hash: str | None
    last_byte_epoch: float
    expected_size: int | None

@dataclass(frozen=True)
class FramingPolicy:
    incomplete_timeout_sec: float
    current_epoch: float

@dataclass(frozen=True)
class FramingManifest:
    verdict: FramingVerdict
    directive: str
    manifest_digest: str

class StreamFramingGuard:
    @staticmethod
    def _hash(data: str) -> str:
        return hashlib.sha256(data.encode("utf-8")).hexdigest()

    @classmethod
    def evaluate_stream(
        cls, state: StreamState, policy: FramingPolicy
    ) -> FramingManifest:
        
        # Gate 1: Unframed stream rejection
        if state.framing_mode == "UNFRAMED":
            return FramingManifest(
                verdict=FramingVerdict.DISCARD_AND_RESET,
                directive="HALT: Unframed stream cannot be parsed reliably. Implement framing.",
                manifest_digest=cls._hash("UNFRAMED")
            )

        # Gate 2: No partial buffer
        if not state.partial_buffer_hash:
            return FramingManifest(
                verdict=FramingVerdict.MESSAGE_COMPLETE,
                directive="PROCEED: No partial buffer. Stream is at a clean message boundary.",
                manifest_digest=cls._hash("CLEAN")
            )

        # Gate 3: Timeout check
        age = policy.current_epoch - state.last_byte_epoch
        if age > policy.incomplete_timeout_sec:
            return FramingManifest(
                verdict=FramingVerdict.STALE_PARTIAL_BUFFER,
                directive=f"HALT: Partial buffer is {age:.1f}s old, exceeding {policy.incomplete_timeout_sec}s timeout. Discard.",
                manifest_digest=cls._hash(f"STALE:{age:.1f}")
            )

        # Gate 4: In progress
        digest = cls._hash(json.dumps({
            "buffer": state.partial_buffer_hash,
            "age": f"{age:.1f}",
            "expected": str(state.expected_size) if state.expected_size else "UNKNOWN"
        }, sort_keys=True))
        
        return FramingManifest(
            verdict=FramingVerdict.MESSAGE_IN_PROGRESS,
            directive="WAIT: Partial message in progress. Await remaining bytes.",
            manifest_digest=digest
        )