rohitg00/ai-engineering-from-scratch · error · ProtocolError

event arrived after message_stop

Error message

event arrived after message_stop

What it means

Raised by collect_stream_text when a stream event appears after message_stop was already received. message_stop is terminal in the Messages streaming protocol; any subsequent event means the event sequence is malformed or events from two streams were concatenated.

Source

Thrown at certifications/claude/lessons/08-messages-api-and-application-lifecycle/code/main.py:166

    if not isinstance(blocks, list) or not blocks:
        raise ProtocolError("response content must be a non-empty block list")
    if not all(isinstance(block, dict) and isinstance(block.get("type"), str) for block in blocks):
        raise ProtocolError("every content block needs a type")
    return blocks


def _text_from_blocks(blocks: list[dict[str, Any]]) -> str:
    return "".join(str(block.get("text", "")) for block in blocks if block["type"] == "text")


def collect_stream_text(events: Iterable[dict[str, Any]]) -> str:
    """Collect only text deltas while checking that a stream terminates."""
    chunks: list[str] = []
    stopped = False
    for event in events:
        event_type = event.get("type")
        if stopped:
            raise ProtocolError("event arrived after message_stop")
        if event_type == "content_block_delta":
            delta = event.get("delta", {})
            if delta.get("type") == "text_delta":
                chunks.append(str(delta.get("text", "")))
        elif event_type == "message_stop":
            stopped = True
    if not stopped:
        raise ProtocolError("stream ended without message_stop")
    return "".join(chunks)


def batch(items: list[Any], size: int) -> list[list[Any]]:
    if size < 1:
        raise ValueError("batch size must be positive")
    return [items[index : index + size] for index in range(0, len(items), size)]


def stable_cache_key(model: str, stable_prefix: str) -> str:

View on GitHub (pinned to 39ea8a1c6d)

Solutions

1. Trim the event list so message_stop is last
2. Clear/rotate stream buffers between requests so events never bleed across messages
3. If aggregating multiple responses, call collect_stream_text once per response

Example fix

// before
events = resp1_events + resp2_events
collect_stream_text(events)
// after
text1 = collect_stream_text(resp1_events)
text2 = collect_stream_text(resp2_events)

def stop_is_last(events):
    stops = [i for i, e in enumerate(events) if e.get("type") == "message_stop"]
    return len(stops) <= 1 and (not stops or stops[0] == len(events) - 1)

def is_terminated_stream(events: list) -> bool:
    seen_stop = False
    for e in events:
        if seen_stop:
            return False
        if e.get("type") == "message_stop":
            seen_stop = True
    return seen_stop

try:
    text = collect_stream_text(events)
except ProtocolError as exc:
    if "after message_stop" in str(exc):
        split_events_and_reprocess()

When it happens

Trigger: Passing an event list like [{"type":"message_stop"}, {"type":"content_block_delta"}] where deltas or any event follow message_stop (test_stream_collector_requires_stop).

Common situations: Concatenating chunks from separate streamed responses, reusing a buffer without clearing it between requests, or replaying captured SSE events out of order.

Related errors

• stream ended without message_stop
• tool_use requires name and object input
• response content must be a non-empty block list
• every content block needs a type
• batch size must be positive

AI-assisted analysis of rohitg00/ai-engineering-from-scratch@39ea8a1c6d (2026-08-26). Data as JSON: /api/errors/5563c8da30b6a2a3. Report an issue: GitHub.