Gateway: retry pre-stream ECONNRESET, synthesize SSE error on mid-stream reset

[james-in-a-box]

Fixes #1907. Makes the gateway's /v1/messages proxy resilient to upstream Anthropic TCP resets.

• (A) Pre-stream retry: if client.send() or the first iter_bytes() prime raises httpx.ReadError/RemoteProtocolError before any downstream byte has flowed, transparently re-issue the upstream request once. Downstream SDK never sees the error.
• (B) Mid-stream synthetic error: if a reset arrives after bytes have already streamed, catch it inside generate(), yield a well-formed SSE event: error frame, and close the stream cleanly so the agent's SDK fails gracefully instead of dying on a truncated socket.

Distinct from #1883 (gateway pod restart); this covers the gateway-healthy/upstream-unhealthy case where the fix belongs inside the gateway.

Test Plan

• Automated: new tests in tests/gateway/test_anthropic_proxy.py::TestStreamingResponse — send() reset → retry success, first-chunk reset → retry success, mid-stream reset → synthetic error frame. Existing streaming tests continue to pass.
• Manual: run pytest tests/gateway/test_anthropic_proxy.py -v and confirm all green.

Manual Steps

Pre-merge: none beyond CI.
Post-merge: observe gateway logs for logger.warning("upstream reset", ...) entries over the next 24h to confirm the code path is exercising under real traffic and not spuriously triggering on healthy streams.

Pipeline Context

Pipeline: issue-1907-v2
Issue: #1907

Per-phase BRC transcripts: implement.

Authored-by: egg

[egg-reviewer]

No agent-mode design concerns.

This PR is pure gateway infrastructure — TCP reset resilience for the /v1/messages proxy. The gateway uses httpx in its designated role as the credential-injection proxy layer (not making its own LLM calls), no prompts are constructed or agent workflows altered, and the synthetic SSE error frame targets the downstream SDK (machine consumer) in Anthropic's documented event shape. All changes are well within the gateway's architectural role.

— Authored by egg

[egg-reviewer]

Solid, well-scoped change. The two-pronged resilience strategy (pre-stream retry + mid-stream synthetic error) is the right design — it avoids the impossible problem of resuming a stateless stream while still giving the SDK a clean failure signal. Tests cover the four key scenarios thoroughly. A few non-blocking observations below.

Non-blocking suggestions

1. _send_and_prime() — resource leak on unexpected exception types

gateway/gateway.py ~L5278-5292

The except clause in _send_and_prime() only catches (ReadError, RemoteProtocolError). If next(iterator) raised any other exception (e.g. httpx.StreamClosed, a hypothetical RuntimeError), upstream_resp would not be closed.

In practice this is extremely unlikely for iter_bytes() iteration, but a broader except BaseException (close-and-reraise pattern) would be more defensive:

upstream_resp = client.send(http_req, stream=True)
try:
    iterator = upstream_resp.iter_bytes()
    try:
        first = next(iterator)
    except StopIteration:
        first = None
    return upstream_resp, iterator, first
except BaseException:
    try:
        upstream_resp.close()
    except Exception:
        pass
    raise

This ensures any exception from iter_bytes() or the first next() cleans up the upstream connection.

2. Documentation: "gated on bytes_seen == 0" is misleading

Both docs/architecture/credential-injection.md and gateway/README.md describe the pre-stream retry as "gated on bytes_seen == 0". The code doesn't actually check bytes_seen — the invariant is structural: _send_and_prime() raises before any byte reaches generate(), so the retry loop can only fire before downstream streaming begins. The bytes_seen variable is always 0 at that point because _consume_chunk only runs inside generate().

Consider rewording to something like "gated on the first chunk not yet having been yielded downstream" to match the actual mechanism. The current phrasing could mislead a reader into looking for an explicit if bytes_seen == 0 guard that doesn't exist.

3. Vestigial nonlocal bytes_seen in generate()

gateway/gateway.py L5348

After extracting accumulator bookkeeping into _consume_chunk(), generate() no longer writes to bytes_seen — it only reads it (in the logger.warning kwargs). The nonlocal bytes_seen declaration is unnecessary (reading from an enclosing scope doesn't require nonlocal in Python). Harmless but worth cleaning up to avoid confusion about which function owns the mutation.

4. Test: _iter_then_raise as @staticmethod

The _iter_then_raise helper on TestStreamingResponse is a generator function used as a @staticmethod. This works, but since it has no coupling to the test class, it could also live as a module-level helper. Minor style preference — no action needed.

Verified correct

• _SSEAccumulator handles the synthetic error frame correctly. The event: error line is ignored by _process_line (doesn't start with "data: "), and the data: {"type":"error",...} line is parsed as an error event, added to _error_blocks, and sets _stop_reason = "error". Transcript capture works end-to-end.
• No information leakage in the synthetic error payload. The error message is a fixed string ("upstream connection reset"), not the raw exception details. Exception details are logged server-side only.
• request_body is reusable across retries. It's raw bytes from request.get_data(), not a consumed file-like object.
• The outer except handlers (ConnectError, TimeoutException, generic Exception) correctly catch retry-exhausted failures because the re-raise at attempt == 1 propagates to the enclosing try.
• No retry for non-transport errors. Application-level errors (4xx/5xx with body) return from _send_and_prime() normally — the retry only fires on the two transport-level exception types.

— Authored by egg

[james-in-a-box]

Contract Verification — PR #1913

Pipeline: issue-1907-v2 | Issue: #1907 | Phase: Implement (phase-1)

Task-by-Task Verification

[task-1-1] Pre-stream retry — VERIFIED

Contract: Add bounded (1x) retry around client.send() and first-chunk prime on httpx.ReadError/RemoteProtocolError before any downstream byte has been yielded. Close the failed upstream, rebuild the request, retry once. On second failure, fall through to the existing error-return path.

Evidence:

• gateway/gateway.py defines _send_and_prime() (lines 5261–5292) encapsulating build_request → send(stream=True) → iter_bytes() → next(iterator). Catches (httpx.ReadError, httpx.RemoteProtocolError), closes failed upstream, re-raises for retry.
• Outer for attempt in range(2) loop (lines 5297–5312): attempt 0 logs and continues; attempt 1 re-raises to existing except Exception handler → 502.
• Request is fully rebuilt each attempt via client.build_request — no stale connection-pool reuse.
• StopIteration on empty upstream correctly sets first_chunk = None.
• Failed upstream is closed before retry to prevent connection-pool leak.

Acceptance criteria: "When client.send() raises ReadError once, the retry succeeds and downstream sees a clean 200 SSE response. When the first iter_bytes() call raises ReadError, the gateway re-primes and produces a normal stream. Both verified by new unit tests." — MET. Tests test_streaming_send_reset_retries_once and test_streaming_first_chunk_reset_retries_once both pass.

---

[task-1-2] Mid-stream synthetic SSE error — VERIFIED

Contract: In generate(), wrap iter_bytes() loop with try/except for httpx.ReadError/RemoteProtocolError. Yield a well-formed synthetic SSE event: error frame with Anthropic-style payload, feed through accumulator, log warning with container_id and bytes_seen, return cleanly. Preserve finally: upstream.close() and _capture_streaming_response.

Evidence:

• generate() (lines 5349–5401) wraps yield first_chunk + for chunk in primed_iterator in inner try/except (httpx.ReadError, httpx.RemoteProtocolError).
• On catch: logs logger.warning with container_id, bytes_seen, error — all required fields present.
• Builds synthetic payload {"type": "error", "error": {"type": "api_error", "message": "upstream connection reset"}} — correct Anthropic error shape.
• Encodes as b"event: error\ndata: <json>\n\n" — well-formed SSE frame.
• Feeds through _consume_chunk() (extracted accumulator helper) before yielding — transcript capture records the failure.
• finally block (lines 5386–5401) preserved unchanged — upstream.close() and _capture_streaming_response() still run.

Acceptance criteria: "When iter_bytes() raises after one chunk has been yielded, the downstream body contains the original chunk followed by a well-formed event: error SSE frame, the stream closes without raising, and _capture_streaming_response still runs." — MET. Test test_streaming_midstream_reset_yields_synthetic_error_frame validates all of these assertions including JSON structure, ordering, and SSE terminator.

---

[task-1-3] Three new tests — VERIFIED

Contract: Extend TestStreamingResponse with tests covering: (a) client.send() raises ReadError once then succeeds on retry; (b) iter_bytes() raises on first iteration then succeeds on retry; (c) iter_bytes() raises RemoteProtocolError after one chunk — downstream body ends with synthetic event: error frame. Use a small helper to wrap an iterator so it raises after N yielded chunks.

Evidence:

• _iter_then_raise(chunks, exc) static helper (lines 570–581): yields chunks then raises — matches the "small helper" requirement.
• Test (a) test_streaming_send_reset_retries_once (line 583): send raises ReadError → retry succeeds → asserts send.call_count == 2, clean 200, no event: error.
• Test (b) test_streaming_first_chunk_reset_retries_once (line 632): iter_bytes raises on first pull → retry succeeds → asserts send.call_count == 2, bad_response.close called, clean stream.
• Test (c) test_streaming_midstream_reset_yields_synthetic_error_frame (line 693): iter_bytes raises after one chunk → asserts send.call_count == 1 (no retry), body contains original chunk then event: error, JSON payload parsed and validated, frame terminator correct.
• Bonus test test_streaming_send_reset_retry_exhausted_returns_502 (line 782): both attempts raise → 502.
• Existing tests test_streaming_request_detected and test_streaming_content_type_forwarded preserved and passing.

Acceptance criteria: "All three new tests pass. Existing tests in TestStreamingResponse continue to pass." — MET. All 6 tests in TestStreamingResponse pass (2 existing + 4 new).

Test Results

tests/gateway/test_anthropic_proxy.py::TestStreamingResponse::test_streaming_request_detected PASSED
tests/gateway/test_anthropic_proxy.py::TestStreamingResponse::test_streaming_content_type_forwarded PASSED
tests/gateway/test_anthropic_proxy.py::TestStreamingResponse::test_streaming_send_reset_retries_once PASSED
tests/gateway/test_anthropic_proxy.py::TestStreamingResponse::test_streaming_first_chunk_reset_retries_once PASSED
tests/gateway/test_anthropic_proxy.py::TestStreamingResponse::test_streaming_midstream_reset_yields_synthetic_error_frame PASSED
tests/gateway/test_anthropic_proxy.py::TestStreamingResponse::test_streaming_send_reset_retry_exhausted_returns_502 PASSED
6 passed in 1.63s

Phase Consistency

• Phase-1 status: complete — all 3 tasks marked complete ✓
• No orphaned code outside contract scope — docs/README updates are supplementary documentation aligned with the feature
• No contract violations found

Supplementary Changes (not in contract, no issues)

• docs/architecture/credential-injection.md: New "Upstream Stream Resilience" subsection with reset-timing table, design rationale, and scope separation from #1883/#1873.
• gateway/README.md: Updated /v1/messages endpoint description + new design decision item #11.
• _consume_chunk() helper extracted from inline accumulator logic in generate() — cleaner, reusable for synthetic frame feeding. Functionally equivalent to the original inline code.

Verdict

All acceptance criteria verified. All tasks complete. No contract violations. Approving.

Note: egg-contract verify-criterion could not be run because the orchestrator is unreachable. Criteria verification is documented above based on code inspection and test execution.

— Authored by egg

[james-in-a-box]

Agreed. Addressed suggestions 1–3 from the code review in a4a1c06:

1. _send_and_prime() resource leak: Broadened the except to BaseException with close-and-reraise, so any unexpected exception from iter_bytes() / next() cleans up the upstream connection.

2. "gated on bytes_seen == 0" wording: Reworded in both docs/architecture/credential-injection.md and gateway/README.md to "gated on the first chunk not yet having been yielded downstream" — matches the actual structural enforcement via _send_and_prime().

3. Vestigial nonlocal bytes_seen: Removed from generate() since it only reads bytes_seen (which doesn't require nonlocal).

Suggestion 4 (_iter_then_raise as @staticmethod) — no change per reviewer's note that no action is needed.

— Authored by egg

[egg-reviewer]

No agent-mode design concerns (re-review).

The delta since 0166bec is a single commit (a4a1c06) addressing code review feedback — broadened exception cleanup in _send_and_prime(), doc wording fix, and removed a vestigial nonlocal. All mechanical; no change to agent-mode assessment.

— Authored by egg

[egg-reviewer]

Re-reviewed the delta from 0166bec to a4a1c06. All three feedback items from the prior review are properly addressed.

Verification of addressed feedback

1. _send_and_prime() resource leak (suggestion #1) — Fixed correctly

The except was broadened from (httpx.ReadError, httpx.RemoteProtocolError) to BaseException with close-and-reraise. This ensures upstream_resp.close() runs on any exception from iter_bytes() / next(), not just the two expected transport errors. The outer retry loop still narrows to just those two exception types, so only transport errors trigger a retry — other exceptions propagate (after cleanup) to the enclosing error handlers. Correct.

2. "gated on bytes_seen == 0" wording (suggestion #2) — Fixed correctly

Both docs/architecture/credential-injection.md and gateway/README.md now read "gated on the first chunk not yet having been yielded downstream" with a parenthetical noting the structural enforcement via _send_and_prime(). Matches the actual mechanism.

3. Vestigial nonlocal bytes_seen (suggestion #3) — Fixed correctly

Removed from generate(). _consume_chunk() retains the nonlocal bytes_seen, capture_truncated declaration and owns the mutation; generate() only reads bytes_seen in the logger.warning kwargs, which doesn't require nonlocal.

No new issues found

The delta introduces no new code paths, no behavioral changes, and no regressions. The three changes are exactly scoped to the feedback.

— Authored by egg

[james-in-a-box]

Contract Re-verification — PR #1913 (incremental)

Pipeline: issue-1907-v2 | Issue: #1907 | Phase: Implement (phase-1)

Delta Since Last Review (0166bec → a4a1c06)

Single commit addressing three code review suggestions. No new features, no behavioral changes.

Verification of Delta Changes

1. _send_and_prime() resource leak fix — Correct

except (httpx.ReadError, httpx.RemoteProtocolError) → except BaseException with close-and-reraise. The outer retry loop at lines 5300–5315 still narrows to only (ReadError, RemoteProtocolError), so retry behavior is unchanged — only cleanup scope is broadened. Any unexpected exception from iter_bytes() / next() now properly closes the upstream connection before propagating.

2. Doc wording fix — Correct

Both docs/architecture/credential-injection.md and gateway/README.md updated from "gated on bytes_seen == 0" to "gated on the first chunk not yet having been yielded downstream". Matches the actual structural enforcement via _send_and_prime().

3. Vestigial nonlocal bytes_seen removal — Correct

Removed from generate(). The nonlocal declaration remains in _consume_chunk() where mutation occurs. generate() only reads bytes_seen in logger.warning kwargs, which does not require nonlocal in Python.

Regression Check

All previously verified acceptance criteria still hold:

Task	Criterion	Status
task-1-1	Pre-stream retry on ReadError/RemoteProtocolError, bounded 1x	Still met — retry loop unchanged
task-1-2	Mid-stream synthetic SSE error frame, stream closes cleanly	Still met — generate() logic unchanged except nonlocal removal
task-1-3	All tests pass, existing tests preserved	Still met — 50/50 tests pass

Test Results

tests/gateway/test_anthropic_proxy.py — 50 passed in 1.61s

All 6 TestStreamingResponse tests pass (2 existing + 4 new).

Verdict

No regressions. Delta correctly addresses all prior review feedback. All contract tasks remain verified.

egg-contract verify-criterion could not be run — orchestrator unreachable. Contract has no top-level acceptance_criteria entries; per-task criteria verified via code inspection and test execution above.

— Authored by egg

[james-in-a-box]

egg is addressing review feedback...

[james-in-a-box]

egg feedback addressed. View run logs

10 previous review(s) hidden.