feat: add resumable Responses streams - #15820
Open
richpoirier wants to merge 3 commits into
Open
Conversation
alt-glitch
added
type/feature
4 tasks
When a streaming /v1/responses request with store=True has its client
disconnect mid-stream, the backend used to interrupt the agent and cancel
the task, which meant the response was lost and could not be recovered via
GET /v1/responses/{id}.
This change introduces a _safe_write() helper that silently absorbs
disconnect errors (ConnectionResetError, ConnectionAbortedError,
BrokenPipeError, OSError) when store=True. The agent keeps running, text
deltas and tool events continue to accumulate in final_text_parts /
emitted_items, and the normal persistence branch writes the completed
response into ResponseStore. A later GET /v1/responses/{id} returns the
full snapshot as if the client had stayed connected.
For store=False the legacy behavior is preserved: disconnect errors
propagate to the outer exception handler, the agent is interrupted, and
the task is cancelled. This avoids spending upstream tokens on an ephemeral
response that no one will read.
Tests:
- test_stream_continues_after_client_disconnect_when_stored
- test_stream_interrupts_agent_on_disconnect_when_not_stored
This is the backend change required for Sage's Phase 1 resumable responses
recovery (see sage-app/docs/plans/2026-04-23-resumable-responses-streaming-implementation-plan.md
Task 4).
Adds Phase 2 resumable-streaming support for /v1/responses: a persistent
per-response event log plus a new GET /v1/responses/{id}/events endpoint
so clients can reconnect after a dropped stream and continue from the
last applied sequence_number, instead of re-POSTing the user turn or
waiting only for the final snapshot.
Changes:
- ResponseStore: new response_events table keyed by
(response_id, sequence_number) with append_event / get_events /
has_events / latest_event_sequence / clear_events helpers. LRU
eviction on the parent responses row cascades to its event log, and
DELETE /v1/responses/{id} also clears the log so replay can't leak
rows for an id the client deleted.
- API server adapter: new in-memory pub/sub state
(_active_responses, _response_event_subscribers) so an active /events
subscriber can live-tail events published by the same-process
streaming handler. _write_event now persists + fans out to
subscribers before the live-client write, so a client disconnect in
the middle of a write never costs subscribers (or the replay cursor)
an event they should have received.
- _write_sse_responses wraps the stream loop in a try/finally that
pushes a terminal sentinel to every /events subscriber, so the
replay handler settles cleanly even if the agent task crashed hard.
- New handler _handle_get_response_events registered at
GET /v1/responses/{response_id}/events?after=<sequence_number>.
Replays stored events > after, then tails live events while the
response is active, with defensive post-finalize drain + stored-row
sweep to close the "subscribed after the last event fired" race.
Returns 404 for unknown ids, 400 for non-integer after.
- Tests: 10 new tests covering ResponseStore event-log CRUD, cursor
semantics (empty catchup, after=0, after=max_seq), completed-replay,
live-tail-then-close, 404 on unknown id, 400 on bad query, legacy
stored-without-events = empty 200.
- Docs: api-server.md gets a section on
GET /v1/responses/{id}/events describing ?after semantics, replay +
live-tail behavior, and how it closes the loop with "continue after
disconnect" for resumable Responses streams.
Runs on top of the continue-after-disconnect change: the POST handler
keeps running after a client drop (store=true), so the event log is
fully populated by the time any /events replay arrives.
Test evidence:
$ pytest tests/gateway/test_api_server.py -k 'responses and (disconnect or events)'
11 passed
The one pre-existing CORS baseline failure in this file
(test_cors_headers_for_origin_disabled_by_default) is unrelated and
reproduces on main before this change.
richpoirier
force-pushed
the
responses-continue-after-disconnect
branch
from
April 26, 2026 19:08
eb25925 to
ab8e43d
Compare
teknium1
reviewed
Jul 12, 2026
teknium1
left a comment
teknium1
left a comment
Contributor
There was a problem hiding this comment.
Thanks for implementing a concrete recovery path for stored Responses streams. The premise is still valid on current main: gateway/platforms/api_server.py:3137-3153 still interrupts and cancels a /v1/responses agent after a client disconnect, while the existing reconnectable SSE work is limited to /v1/runs/{id}/events (72cf4d8).
Problems
gateway/platforms/api_server.py:2300creates an unbounded queue for every/eventssubscriber, and_write_eventfans every event into every queue at lines 1491-1493. A slow client can retain an unbounded stream of event payloads in memory.ResponseStore.append_event()persists and commits every event at lines 425-431, with no per-response event/byte retention bound. The response-count LRU does not bound a single long stream.- The PR's
ResponseStore.delete()omits current main's conversation-mapping cleanup (gateway/platforms/api_server.py:503-513), so a salvage must retain that newer behavior.
Suggested changes
- Port the bounded retained-log approach used by current run-event replay, including explicit retention-gap behavior and slow-subscriber coverage.
- Salvage onto current
api_server.pyrather than carrying forward the olderResponseStoreimplementation.
Automated hermes-sweeper review.
| # otherwise be lost. Since stored + live are deduplicated client- | ||
| # side by sequence_number (and we dedupe via a local "seen set" | ||
| # below), a temporary double-delivery is fine. | ||
| live_q: "asyncio.Queue" = asyncio.Queue() |
Contributor
There was a problem hiding this comment.
asyncio.Queue() is unbounded, while _write_event puts every event into every subscriber queue. A slow connected /events client can retain the entire response stream in memory. Please use a bounded/shared retained-log cursor design (as current /v1/runs/{id}/events does) and define the replay-gap behavior.
| "VALUES (?, ?, ?, ?, ?)", | ||
| (response_id, int(sequence_number), event_type, payload, time.time()), | ||
| ) | ||
| self._conn.commit() |
Contributor
There was a problem hiding this comment.
This commits synchronously once per emitted event and the new table has no per-response event or byte bound. A single long stored stream can grow disk usage without bound and repeatedly block the asyncio loop. Add bounded retention and batch/offload persistence, with coverage for the retention limit.
| # Drop any event-log rows for this response too so we don't leak | ||
| # rows for an id that can no longer be looked up. | ||
| self._conn.execute( | ||
| "DELETE FROM response_events WHERE response_id = ?", (response_id,) |
Contributor
There was a problem hiding this comment.
Current main's ResponseStore.delete() also deletes conversations rows referencing this response before removing it. Preserve that cleanup when salvaging this method; otherwise named conversations can remain mapped to a deleted response ID.
teknium1
added
sweeper:risk-session-state
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What does this PR do?
I have my own front end client that uses the API server. I wanted it to work similar to messaging platforms - shoot off a message, come back to it in the middle of the agent response or after it’s done and see everything it did. As well as having multiple sessions going at once, that I can pop in and out of without any connection issues. These changes achieve that.
Adds backend support for resumable
/v1/responsesstreams so clients can recover after dropped SSE connections, page reloads, or returning to an in-progress response.For
store=truestreaming responses, the agent task now continues running after a client disconnect, records the response event stream, and persists the completed response for later recovery viaGET /v1/responses/{id}. Forstore=false, the existing token-saving behavior is preserved: disconnecting the client interrupts and cancels the agent task.This approach keeps non-stored streams inexpensive while enabling messaging-style clients to safely reconnect, replay missed events, and live-tail active stored responses without losing response state.
Related Issue
N/A — no linked issue.
Type of Change
Changes Made
gateway/platforms/api_server.pystore=truestreaming responses alive after client disconnects.ResponseStore.GET /v1/responses/{id}/events?after=<sequence_number>.store=falsestreams.tests/gateway/test_api_server.pyAPI_SERVER_CORS_ORIGINSenvironment variable.website/docs/user-guide/features/api-server.mdHow to Test
python -m pytest tests/gateway/test_api_server.py -k 'ResponseStoreEventLog or ResponsesEventsReplay or stream_continues_after_client_disconnect_when_stored or stream_interrupts_agent_on_disconnect_when_not_stored or stream_cancelled_persists_incomplete_snapshot'Checklist
Code
fix(scope):,feat(scope):, etc.)pytest tests/ -qand all tests pass — targeted gateway/API server tests were run insteadDocumentation & Housekeeping
docs/, docstrings) —website/docs/user-guide/features/api-server.mdcli-config.yaml.exampleif I added/changed config keys — N/ACONTRIBUTING.mdorAGENTS.mdif I changed architecture or workflows — N/AScreenshots / Logs
Backend/API change; no screenshots.