feat(instrument): Extract SSE parser to shared utility, port to agno + ms_agent_framework (cross-poll #9)

[mmercuri]

Summary

Cross-pollination item #9 from A:/tmp/adapter-cross-pollination-audit.md §2.10 / §3.1: extract Server-Sent Events handling from per-adapter ad-hoc code into a single shared utility, then wire it into the agno and ms_agent_framework streaming code paths so each chunk emits a discrete model.stream.chunk event instead of being collapsed into one accumulated emission.

What changed

Shared parser (src/layerlens/instrument/adapters/_base/sse_parser.py)

• SSEEvent (pydantic model): one dispatched event with data, event, id, retry, plus a done convenience flag for the OpenAI / Agentforce [DONE] sentinel.
• SSEParser: incremental, stateful parser. Spec-compliant per the WHATWG HTML "Server-sent events" section.
  • Partial events at chunk boundaries: held in a line-remainder buffer until a terminator arrives.
  • UTF-8 multi-byte codepoint splits across chunks: handled via codecs.getincrementaldecoder so a 3-byte ★ or 4-byte 🌍 reassembled correctly.
  • Mixed \n / \r / \r\n line terminators all parse correctly.
  • Multi-line data: joined with \n per spec; one leading space stripped after : per spec.
  • Comments (: lines), unknown fields, NUL in id:, non-digit retry: all tolerated.
  • 1 MiB MAX_LINE_BYTES guard against runaway buffers from malformed upstream.
• parse_event(text): convenience for a fully-framed single block.
• parse_stream(byte_iter): async helper for httpx.aiter_bytes-style sources. Drains a final flush() so an unterminated trailing event surfaces correctly.

agno port

• AgnoAdapter._create_traced_run / _create_traced_run_sync now detect stream=True calls. When the result is iterable, the adapter wraps it in a generator that emits one model.stream.chunk per yielded object — or, when chunks are bytes/str, runs them through SSEParser so multi-event chunks expand into per-event emissions and partial events across chunks are reassembled.
• Final on_run_end fires from the wrapper's finally block on stream exhaustion so duration measurement remains accurate for the streaming path.

ms_agent_framework port

• MSAgentAdapter._create_traced_invoke_stream now emits a discrete model.stream.chunk event per StreamingChatMessageContent (or per parsed SSE event when chunks arrive as raw bytes/str), rather than silently accumulating last_message and emitting only run start/end.

Capture-config plumbing

• CaptureConfig.is_layer_enabled gains a model.stream.chunk → l3_model_metadata mapping so the new event type is gated by the same layer flag as model.invoke. Strictly additive — no existing event-type or default behavior changes.

LangChain refactor

Not applicable on this branch — the LangChain adapter does not currently implement its own SSE parser (it consumes upstream LangChain callbacks rather than raw HTTP). Audit item §2.10 lists langchain as "none" for SSE parsing; only agentforce had bespoke SSE wire-format handling, which now has a shared replacement available for future adoption.

Test plan

• uv run pytest tests/instrument/adapters/_base/test_sse_parser.py -x → 56 / 56 passed
  • W3C field handling (data / event / id / retry, comments, unknown fields, leading-space stripping, NUL in id, non-digit retry)
  • Multi-line data: concatenation per spec
  • Mixed \n / \r / \r\n terminators
  • Partial event / chunk boundary handling (mid-line, mid-field-name, mid-block, byte-by-byte)
  • UTF-8 multi-byte codepoint splits (3-byte and 4-byte)
  • [DONE] sentinel surfaced via done flag
  • Async parse_stream() over an async byte iterator
  • Single-block convenience parse_event()
  • Malformed input tolerance, buffer overflow protection
  • Realistic OpenAI chat.completions and Agentforce text streams end-to-end
  • Parametrized stress: byte-by-byte through 256-byte chunks, all yield identical events
• uv run pytest tests/instrument/adapters/frameworks/test_agno_adapter.py -x → 17 / 17 passed (12 pre-existing + 5 new streaming tests)
• uv run pytest tests/instrument/adapters/frameworks/test_ms_agent_framework_adapter.py -x → 16 / 16 passed (12 pre-existing + 4 new streaming tests)
• uv run pytest tests/instrument/ --ignore=...known-pre-existing → 232 passed, 1 skipped (no regressions)
• uv run pytest tests/ --ignore=...known-pre-existing → 834 passed, 1 skipped (no regressions in resources / models / clients)
• uv run mypy --strict on every modified .py file → clean
• uv run ruff check on every modified .py file → clean

Files modified

• src/layerlens/instrument/adapters/_base/sse_parser.py (new, 372 lines)
• src/layerlens/instrument/adapters/_base/__init__.py (export new symbols)
• src/layerlens/instrument/adapters/_base/capture.py (one-line additive map entry)
• src/layerlens/instrument/adapters/frameworks/agno/lifecycle.py (streaming wrapper)
• src/layerlens/instrument/adapters/frameworks/ms_agent_framework/lifecycle.py (streaming wrapper)
• tests/instrument/adapters/_base/__init__.py (new package)
• tests/instrument/adapters/_base/test_sse_parser.py (new, 56 tests)
• tests/instrument/adapters/frameworks/test_agno_adapter.py (+5 streaming tests)
• tests/instrument/adapters/frameworks/test_ms_agent_framework_adapter.py (+4 streaming tests)