fix: emit clean client error instead of bricking the session when the streaming validator detects a corrupted outbound stream

[scottwofford]

Summary

COE follow-up for PR #356, escalating the PR #426 streaming protocol validator from log-and-warn to enforcement. Trello: COE audit: Emit a clean client error instead of bricking the session when the streaming validator detects a corrupted outbound stream

• New StreamingProtocolValidator class (incremental: observe() per event + finalize() at end of stream) in stream_protocol_validator.py. The existing batch validate_anthropic_event_ordering() is now a thin wrapper over it, so there is exactly one rule set; all 21 pre-existing validator tests pass unchanged.
• _handle_execution_streaming validates each outbound event BEFORE forwarding it. On a violation (out-of-order events, the PR fix: safety judge returns 400 on parallel MCP tool use #356 bug class) the pipeline:
  1. withholds the corrupting event (it never reaches the client),
  2. emits a structured SSE error event ({"type": "error", "error": {"type": "api_error", "message": "Luthien proxy detected a corrupted response stream (protocol violation: <rules>) ... (transaction: <call_id>)"}}) that Anthropic-protocol clients parse as a normal API error,
  3. ends the stream and closes the policy emission generator.
• The client fails the turn cleanly instead of reconstructing a malformed assistant message, so the session stays usable (no persistent 400s, no /rewind needed).
• Recorded as streaming.protocol_violation policy event with aborted: true; webhook fires with success=False, http_status=500 (same accounting as the mid-stream exception path).
• End-of-stream rules (message_stop last, all blocks closed, non-empty) stay advisory log-and-warn: they are only decidable after every event has already been forwarded, so they cannot gate forwarding.

Definition of done (from the Trello card)

• When the streaming protocol validator detects out-of-order events, the proxy emits a structured error event to the client (not just a log warning)
• The client session remains usable after receiving the error (no bricked state): the corrupting event is withheld, so the client never reconstructs a malformed assistant message; the error event fails the turn the same way the existing mid-stream exception path does
• Covered by a unit test with a deliberately corrupted stream fixture (TestProtocolViolationAbortsStream, using the exact PR fix: safety judge returns 400 on parallel MCP tool use #356 corruption shape)

Design tradeoff (load-bearing)

Enforcement means a validator false positive would now abort a healthy stream (one failed turn, client retries) instead of just logging. I judged this acceptable because the enforced rules are exact Anthropic protocol invariants with dedicated unit coverage, and the failure mode of a false positive (one clean failed turn) is strictly better than the failure mode it prevents (silently corrupted session). If we want a kill switch, a config field can gate enforcement back to log-and-warn in a follow-up; I kept this PR to the simplest enforcement per repo convention.

RCA / COE

Root cause: PR #426 deliberately wired the protocol validator as log-and-warn, and it validated only after the stream completed, when every event had already been forwarded to the client. A corrupted stream (e.g. content blocks emitted after message_delta) was therefore delivered intact; the client (Claude Code) reconstructed a malformed assistant message, resent it as history on every subsequent request, and the API returned 400 on each: a bricked session recoverable only via /rewind.

Why it wasn't caught: it was, by design review rather than by an incident. PR #426's own description flagged the escalation path ("If we see violations in practice, we can escalate to aborting the stream to prevent session-bricking"), and the 2026-03-25 COE audit of PR #356 turned that into this tracked card. The gap was architectural: post-hoc batch validation cannot protect the client, because detection happens after forwarding. No test asserted what the client receives when the outbound stream is corrupted, only that violations were logged.

Why it won't recur: validation is now structural, on the forwarding path itself: every event passes StreamingProtocolValidator.observe() before it is yielded to the client, so any future policy bug in this class (PR #134 and PR #329/#356 were both this class) is contained at the proxy boundary regardless of whether the offending policy has test coverage. The batch validator is a wrapper over the same incremental class, so test-fixture rules and runtime-enforcement rules cannot drift apart. TestProtocolViolationAbortsStream locks in the client-observable contract: corrupted stream produces exactly one well-formed error event, corrupting events withheld, healthy streams untouched.

Test plan

• 8 new unit tests for the incremental validator (violation flagged on the offending event, finalize rules, ping tolerance, empty stream)
• 3 new pipeline tests: PR fix: safety judge returns 400 on parallel MCP tool use #356 corruption fixture yields a single structured error event and withholds the corrupted tail; corrupted first event yields exactly one error event (empty-stream error suppressed); healthy stream forwarded unchanged with no error event
• All 21 pre-existing validator tests pass unchanged against the refactored implementation
• Full unit suite: 2851 passed
• scripts/dev_checks.sh clean (ruff format + lint, pyright 0 errors)
• Mock e2e tier passed, sqlite e2e tier passed (client-side stream validation in test_mock_simple_llm_parallel_tools.py exercises the enforced path end to end)

🤖 Generated with Claude Code

[claude]

Code Review — PR #800

Overall: strong COE follow-up. Structural fix (validation on the forwarding path, not post-hoc) is the right escalation. The batch/incremental unification prevents test-fixture rules from drifting from runtime-enforcement rules — that's the load-bearing correctness property here. Comments justify their length: the streaming state machine has enough interacting flags (stream_completed, emitted_any, cancelled, caught_exception, protocol_aborted, final_status) that leaving out the WHY would make future edits hazardous.

Correctness

• Ordering of emitted_any = True (anthropic_processor.py:803) is now after the violation check. Correct — this is what makes is_empty_stream = not emitted_any and ... and not protocol_aborted and the test_corrupted_first_event_yields_exactly_one_error_event invariant coexist: a first-event violation keeps emitted_any=False, and the not protocol_aborted clause suppresses the empty-stream branch so the client sees exactly one error event.
• stream_completed = not protocol_aborted (line 807) correctly keeps the finalize() end-of-stream check gated behind final_status == 200 — no double-recording of the same failure.
• aclose() on the emissions generator (lines 812-820) is the right resource cleanup; catching bare Exception and logging is defensible since we're already on the abort path.
• Rule dedup via ", ".join(sorted({v.rule for v in violations})) in _build_protocol_violation_error_event — clean, deterministic error message.

Small suggestions

1. Test brittleness (tests/luthien_proxy/unit_tests/pipeline/test_anthropic_processor.py:716, 758):

   recorded_event_types = [call.args[1] for call in emitter.record.call_args_list]

   Positional indexing on emitter.record calls will break silently if the emitter signature ever gains a keyword arg or the caller switches to kwargs. Consider call.args[1] if len(call.args) > 1 else call.kwargs.get("event_type") or matching on emitter.record.call_args_list more explicitly.

2. Defensive getattr check (anthropic_processor.py:812):

   aclose = getattr(emissions, "aclose", None)
   if aclose is not None:

   emissions is typed as an async iterator produced by policy execution — aclose should always be present on the async-generator objects flowing here. Not harmful, but the if aclose is not None branch may be effectively dead code. If policies could ever emit a non-generator async iterator (e.g., a manually-implemented __aiter__), then this is warranted — worth a one-line comment either way.

3. Pre-existing quirk carried over (stream_protocol_validator.py, observe() in content_block_start branch): when idx <= self._highest_start_index fires (monotonicity violation), the code still runs self._highest_start_index = idx when idx >= 0, which can move the marker backwards. This is inherited from the batch validator, not introduced here — noting it so it's tracked. Not blocking for this PR; would be worth a follow-up.

4. ensure_request_recorded() is called before the violation check (line 762), so a first-event violation still records the request. That's the pre-existing invariant and probably what you want (the request happened, we just refused to forward the response) — worth a one-line note if this is an intentional preserved behavior rather than incidental.

Test coverage

Excellent — three complementary pipeline tests (PR #356 fixture, first-event violation, healthy stream unaffected) plus per-rule incremental-validator tests. The _make_event() update in the webhook-gate tests to yield message_start first is a necessary and correct consequence of enforcement — good catch that keeps those tests exercising the intended flow.

Design

The tradeoff you called out in the PR body (false positive → one clean failed turn vs. false negative → session brick) is correctly reasoned. Given the enforced rules are exact Anthropic protocol invariants with dedicated tests, log-and-warn on the enforceable rules would just be paying the false-negative cost forever. A future config gate is easy to add if needed; not adding one now is the right call per repo conventions.

Nits

• Some comment blocks are 10+ lines. Justified given the state-machine complexity, but consider whether the abort-path comment (lines 764-774) could be trimmed since the _build_protocol_violation_error_event docstring restates most of it. Non-blocking.

Ship it — the correctness properties matter more than the small polish notes above.

[claude]

Code review

Excellent PR overall: the incremental-validator refactor is clean, the batch API is preserved as a one-line wrapper, the enforcement path is minimally invasive, and the test coverage locks in the client-observable contract (corrupted event withheld, one well-formed error event, transaction id included). The COE section is exemplary — it names the class of bug, the architectural gap that caused it, and the structural fix.

A few small observations, none blocking:

1. transaction.streaming_response_recorded still fires with partial data on protocol abort

src/luthien_proxy/pipeline/anthropic_processor.py:1024 records original_response / final_response from the reconstructed clean prefix even when protocol_aborted=True. Downstream consumers of the transaction event stream will now see a streaming_response_recorded immediately followed by a streaming.protocol_violation with aborted: true, but the recorded response is partial (everything up to but not including message_delta, if the violation was mid-stream). Existing consumers that treat streaming_response_recorded as "response fully delivered" may be surprised.

Not necessarily a bug — the streaming.protocol_violation event with aborted: true is the authoritative signal — but worth confirming whether any downstream consumer (history, activity monitor, webhook aggregation) treats streaming_response_recorded as terminal success. If yes, a small guard like if reconstructed is not None and not protocol_aborted: around line 1011 would keep the two paths semantically consistent.

2. User-facing error message leaks internal rule names

src/luthien_proxy/pipeline/anthropic_processor.py:1313 embeds the raw rule slug (content_before_message_delta, block_start_monotonic, etc.) into the client-visible error message. This is useful for triage when a user reports the transaction id, but it does expose internal validator vocabulary — I'd consider whether a friendlier hard-coded string ("streaming event ordering") plus the transaction id would be enough, with the specific rule names living only in the log/policy event. Judgment call; either is defensible.

3. "Please retry the request" may mislead on deterministic policy bugs

src/luthien_proxy/pipeline/anthropic_processor.py:1318-1322 advises the client to retry. If the corruption is caused by a deterministic policy bug (which is exactly the PR #356 failure mode), the retry will hit the same violation, and the client burns a turn each time. The stronger phrasing might be "the proxy stopped this response before it reached your client — retry may or may not succeed depending on the underlying cause; please report the transaction id to your administrator if the failure persists."

4. Narrow edge: CancelledError during emissions.aclose() after protocol abort

src/luthien_proxy/pipeline/anthropic_processor.py:815: await aclose() is wrapped in except Exception, but if the client disconnects at exactly the wrong moment and aclose() raises CancelledError (BaseException, not Exception), the outer except asyncio.CancelledError catches it and sets cancelled=True, final_status=499, overriding the protocol_aborted accounting. Result: the webhook would fire with http_status=499 for a request whose actual failure was a protocol violation. Very narrow window (aclose is fast because the generator is suspended awaiting the next event), and either final_status is defensible for observability, but noting for completeness.

5. Minor: batch validator now emits violations in a different order

validate_anthropic_event_ordering previously grouped violations by rule (all message_start_first, then all message_stop_last, then all content_before_message_delta, then block-lifecycle). It now interleaves per-event violations then appends finalize violations. Since the existing 21 tests use any(v.rule == ...) rather than positional assertions, all pass — but this is a subtle behavior change for any external caller that relied on the old grouping. Probably fine, worth a mention in the changelog if you want to be defensive.

6. StreamingProtocolValidator state note

stream_protocol_validator.py:265: finalize() treats "no message_stop last" as a violation. When called on a protocol-aborted stream (where the corrupting event caused an early break), that would always fire because the corrupting event or a message_stop never landed. The caller currently gates finalize() on final_status == 200, so this is fine — but the class-level docstring could note that finalize() is only well-defined for streams intended to complete, not for streams that aborted mid-flight.

---

Nothing here blocks merging. The core mechanism is right, the tests are precise, and the design tradeoff (enforcement vs log-and-warn) is called out honestly in the PR body.

[claude]

Review

Overall, this is a well-scoped and well-tested COE follow-up. The refactor of the batch validator into an incremental one shared with the pipeline (single rule set, single class) is the right shape, and the design tradeoff is clearly reasoned in the PR body. I focused on correctness of the mid-stream enforcement path.

Strengths

• Single source of truth for rules. validate_anthropic_event_ordering is now a thin wrapper over StreamingProtocolValidator, so test fixtures and runtime enforcement cannot drift apart. All 21 pre-existing tests pass unchanged.
• Clean-error contract enforced by tests. TestProtocolViolationAbortsStream locks in exactly what the client sees on the PR fix: safety judge returns 400 on parallel MCP tool use #356 fixture (event: error at the tail, corrupting block-1 events withheld, no message_stop forwarded, exactly one error event). The test_corrupted_first_event_yields_exactly_one_error_event case explicitly guards the interaction with the empty-stream branch — that was a real trap and it's tested.
• Bookkeeping is coherent under abort. protocol_aborted correctly gates: (a) the empty-stream error suppression (is_empty_stream = ... and not protocol_aborted), (b) transaction.streaming_response_recorded suppression (partial reconstruction is not recorded as delivered), (c) finalize() being skipped (final_status != 200, so its unclosed/unstopped rules don't double-report), (d) webhook fires with success=False, http_status=500 — same accounting as the mid-stream exception path.
• Generator cleanup on abort. aclose() on the emissions generator is guarded with getattr (protecting against manually-implemented __aiter__) and wrapped in try/except so a failure in cleanup can't crash the response.
• The end-of-stream rules stay advisory, correctly — they're only decidable post-hoc, so they can't gate forwarding. The reasoning is written into the code comments and matches the runtime behavior.

Observations / Minor Suggestions

1. call_id in client-facing error message (_build_protocol_violation_error_event, stream_protocol_validator.py/anthropic_processor.py:1329-1332). The message ends with report transaction {call_id} to your administrator, which exposes the internal transaction UUID to the client. This matches the debuggability intent and is defensible (the UUID isn't sensitive), but note that mid-stream error messages elsewhere in this file (_build_error_event) don't include it. Consider whether that inconsistency is intentional or worth aligning.

2. _get_block_index accepts negative indices in _started_blocks. If an index is negative, block_index_non_negative fires but the index is still added to _started_blocks (line 200). A subsequent content_block_delta(-1) would then find the block "started" and skip the delta_after_start check. Not a real issue in practice — the negative index is already flagged — but slightly permissive. Fine to leave as-is.

3. chunk_count vs. client-received events. response_span.set_attribute(\"streaming.chunk_count\", chunk_count) only counts successfully-forwarded events; on abort, the client also receives the structured error event, which is not counted. Arguably correct (chunk_count = successful chunks), but worth noting for anyone reading span data.

4. Multiple message_start events aren't caught by the validator (pre-existing rule set, not this PR). message_start_first only fires at i == 0. A stream with a spurious second message_start at index 5 would pass. Consider adding a message_start_once rule in a follow-up if this failure mode is worth guarding against.

5. Nit — comment slightly misleading. Line 875: # protocol_aborted is excluded: the abort path already yielded its own structured error event — worth noting that the abort path also runs BEFORE the empty-stream check runs, so the ordering guarantee is what actually prevents a double error. The comment is right, just worth making explicit that both emitted_any and protocol_aborted gate this independently.

Test Coverage

• 8 new unit tests for the incremental validator directly assert per-event behavior (violation flagged on the offending event, not just at end).
• 3 pipeline tests cover the client-visible contract end-to-end.
• Healthy-stream regression test (test_healthy_stream_is_unaffected) confirms no false positives.
• Interaction with the empty-stream error path is explicitly tested (both first-event corruption and the pre-existing empty-stream branch).
• Good test hygiene: fixtures build events at the SDK type level; assertions target observable client behavior + emitted event types, not internal state.

Verdict

LGTM. The design correctly moves validation from post-hoc to structural (on the forwarding path), which is what prevents recurrence of the PR #356 bug class regardless of individual policy test coverage. The COE analysis is clear about why this needed enforcement rather than just better logging, and the failure mode of a false positive (one clean failed turn) is genuinely better than the failure mode it prevents (bricked session).

🤖 Generated with Claude Code

[scottwofford]

---

Post-review note (adversarial verification pass, Jul 7): merges cleanly with #799, but this PR's _build_protocol_violation_error_event() does not route through #799's append_advice, so protocol-violation errors would ship without the Suggestion: line that #799 adds everywhere else. Cosmetic inconsistency, not a bug; reconcile whichever lands second.

[scottwofford]

Claude-generated merge-queue triage of all open Luthien PRs, requested by Scott (Jul 7, 2026). Advisory only; Scott has not yet acted on these recommendations.

Recommendation: request changes (one item), then merge.

The fix direction is right: validate-before-forward with a single rule source, converting a bricked-session failure (the #134 / #329 class) into a clean, retryable client error. The item: enforcement is default-on with no configuration flag. The previous behavior was log-and-warn; with this PR a validator false positive aborts a healthy turn, and the heaviest traffic on this proxy is high-volume agent swarms, where any edge case shows up in absolute numbers. A config field that can drop enforcement back to log-and-warn without a redeploy is cheap insurance and makes this safe to land. One human read before merge is warranted since this changes runtime behavior on production traffic.