httpx2 2.10.0: Consume Server-Sent Events streams and decode structured event data with EventSource in httpx2

# Server-Sent Events (SSE) Streaming and Decoding in httpx2

## 1. What `search_known_solution` Answered
- `search_known_solution` returned existing samples for query parameter merging (`sha256:632bd37aef7f7583aaef7866ba216cfa65df1a49a27b1214814cc48a92389c40`) and FastAPI integration, but no existing sample demonstrated `httpx2`'s built-in Server-Sent Events (SSE) support via `EventSource`.

## 2. What a Naive Model Would Have Written Instead
- In standard `httpx`, SSE support requires installing and importing a third-party package (`httpx-sse`). Models unaware of `httpx2`'s built-in SSE capabilities attempt to import `httpx_sse` or manually parse raw byte streams from `iter_lines()` with regular expressions, error-prone delimiter handling, or missing multi-line `data:` concatenation.
- Naive implementations often attempt to read responses synchronously without streaming context (`client.get()` instead of `client.stream()`), causing buffering of indefinite event streams into memory.
- Models also neglect to validate `Content-Type: text/event-stream` or handle non-200 responses gracefully before feeding the response stream to the event decoder.

## 3. How the Wrong Version Fails
- Manual chunk/line parsing fails to handle multi-line `data:` fields properly (they must be joined with newlines `\n`), incorrectly parses comments starting with `:` as events, misses last event ID persistence across subsequent events, or fails to raise `SSEError` when non-event-stream responses are received.

Clean-room public code sample — generation instructions

Write a brand-new, minimal, self-contained code sample in this clean-room directory.
Do not copy, paraphrase, or reference any existing project source. Work only from this spec.

A csx.json manifest scaffold already exists. Do not recreate it from memory. Preserve its case.goal, packages and symbols; fill its empty case.contract with exact assertions and correct its environment, commands and verifierAdapter for the files you generate.

Goal: Consume Server-Sent Events streams and decode structured event data with EventSource in httpx2
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:pypi/httpx2@2.10.0
Demonstrate these symbols/APIs:
  - httpx2.EventSource
  - httpx2.SSEError
  - httpx2.ServerSentEvent
Constraints:
  - executionContext: node
Required runtime conditions:
  - ecosystem: npm
  - language: javascript
  - moduleSystem: cjs
  - packageManager: npm@10.9.8
  - runtime: node@22.23.2

Rules:
  - One focused purpose; the smallest project that proves the goal.
  - Include a contract test (test/contract.*) that runs OFFLINE and exits 0 exactly when the goal behavior works.
  - Pin every dependency with a lockfile so resolution is reproducible.
  - No secrets, credentials, or tokens. No real URLs (only example.com or localhost). No absolute paths.
  - No personal names, emails, company names, or project identifiers of any kind.
  - No binaries and no generated output (node_modules, dist, target, venv, .git, .env).
  - Keep it under 200 files and 256KB packed.

{"case":{"caseId":"case:sha256:af39e93c76c2279ce8a19837554d3a2d2489d04148d4b0a3d28169f457f0dbb9","contract":["assert iterating over EventSource(response) yields ServerSentEvent instances from a text/event-stream response","assert ServerSentEvent parses custom event types, last event ID, retry interval, and multiline data payloads","assert ServerSentEvent.json() deserializes JSON payloads from the event data field","assert EventSource raises SSEError when response Content-Type is not text/event-stream","assert EventSource raises SSEError when a single event exceeds max_event_size"],"goal":"Consume Server-Sent Events streams and decode structured event data with EventSource in httpx2","kind":"HOW","packages":["pkg:pypi/httpx2@2.10.0"],"schemaVersion":1,"symbols":["httpx2.EventSource","httpx2.SSEError","httpx2.ServerSentEvent"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","ecosystem":"pypi","executionContext":"python","language":"python","os":"linux","packageManager":"pip","runtime":"python","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/httpx2@2.10.0"],"schemaVersion":1,"symbols":["httpx2.EventSource","httpx2.SSEError","httpx2.ServerSentEvent"],"verifierAdapter":"python@1"}

httpx2==2.10.0
httpcore2==2.10.0
idna==3.18
truststore==0.10.4
anyio==4.13.0
h11==0.16.0
typing-extensions==4.15.0

{
  "schemaVersion": 1,
  "goal": "Consume Server-Sent Events streams and decode structured event data with EventSource in httpx2",
  "kind": "HOW",
  "packages": [
    "pkg:pypi/httpx2@2.10.0"
  ],
  "symbols": [
    "httpx2.EventSource",
    "httpx2.SSEError",
    "httpx2.ServerSentEvent"
  ],
  "constraints": {
    "executionContext": "node"
  },
  "runtimeConditions": {
    "ecosystem": "npm",
    "language": "javascript",
    "moduleSystem": "cjs",
    "packageManager": "npm@10.9.8",
    "runtime": "node@22.23.2"
  }
}

"""httpx2 Server-Sent Events (SSE) consumer sample."""

"""Reference implementation for streaming and decoding Server-Sent Events using httpx2."""

from __future__ import annotations

from typing import Iterator
import httpx2


def consume_event_stream(
    client: httpx2.Client,
    url: str,
    max_event_size: int | None = None,
) -> Iterator[httpx2.ServerSentEvent]:
    """Stream and yield ServerSentEvent instances from an SSE endpoint.

    Opens a streaming HTTP connection, validates the content type, and decodes
    server-sent events using httpx2's built-in EventSource.
    """
    with client.stream("GET", url) as response:
        event_source = httpx2.EventSource(response, max_event_size=max_event_size)
        yield from event_source

import sys
from pathlib import Path

# Ensure src is importable
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

import httpx2
from src.sse import consume_event_stream

# ---------------------------------------------------------------------------
# 1. Mock SSE Server Transport
# ---------------------------------------------------------------------------
sse_stream_payload = (
    b": initial keepalive comment\n"
    b"event: user_connected\n"
    b"id: evt-1001\n"
    b"retry: 3000\n"
    b'data: {"user_id": "u-101", "name": "Alice"}\n'
    b"\n"
    b": comment between events\n"
    b"data: multiline message payload\n"
    b"data: continued on line two\n"
    b"\n"
    b"event: system_metric\n"
    b'data: {"cpu_pct": 42.5, "status": "nominal"}\n'
    b"\n"
)


def mock_sse_handler(request: httpx2.Request) -> httpx2.Response:
    if request.url.path == "/events":
        return httpx2.Response(
            200,
            headers={"content-type": "text/event-stream; charset=utf-8"},
            content=sse_stream_payload,
        )
    elif request.url.path == "/invalid-content-type":
        return httpx2.Response(
            200,
            headers={"content-type": "application/json"},
            content=b'{"error": "not event stream"}',
        )
    return httpx2.Response(404)


transport = httpx2.MockTransport(mock_sse_handler)
client = httpx2.Client(transport=transport)

# ---------------------------------------------------------------------------
# 2. EventSource decodes structured events with event, id, retry, and JSON data
# ---------------------------------------------------------------------------
events = list(consume_event_stream(client, "https://example.com/events"))
assert len(events) == 3, f"Expected 3 events, got {len(events)}"

# First event: custom event type, explicit ID, retry interval, JSON payload
evt0 = events[0]
assert evt0.event == "user_connected"
assert evt0.id == "evt-1001"
assert evt0.retry == 3000
assert evt0.data == '{"user_id": "u-101", "name": "Alice"}'
assert evt0.json() == {"user_id": "u-101", "name": "Alice"}

# Second event: default event name 'message', multiline data joined with newline,
# retains the previous event ID per SSE specification
evt1 = events[1]
assert evt1.event == "message"
assert evt1.id == "evt-1001"
assert evt1.retry is None
assert evt1.data == "multiline message payload\ncontinued on line two"

# Third event: new event name, retains event ID, JSON payload
evt2 = events[2]
assert evt2.event == "system_metric"
assert evt2.id == "evt-1001"
assert evt2.json() == {"cpu_pct": 42.5, "status": "nominal"}

# ---------------------------------------------------------------------------
# 3. Content-Type validation: Non text/event-stream raises SSEError
# ---------------------------------------------------------------------------
try:
    list(consume_event_stream(client, "https://example.com/invalid-content-type"))
    raise AssertionError("Expected SSEError when content-type is not text/event-stream")
except httpx2.SSEError as exc:
    assert "Expected response with content type 'text/event-stream'" in str(exc)

# ---------------------------------------------------------------------------
# 4. Max event size limit enforcement
# ---------------------------------------------------------------------------
try:
    list(consume_event_stream(client, "https://example.com/events", max_event_size=20))
    raise AssertionError("Expected SSEError when event size exceeds max_event_size")
except httpx2.SSEError as exc:
    assert "exceeded" in str(exc)

print("All httpx2 EventSource contracts verified successfully.")