Hi, just want to follow up on this PR. I know it is not a small one, so I totally understand if it's not easy to review quickly. I didn't break it down because all changes are made for one goal: the compelete background tasks running.

Following are changes that this PR made and are delibrated in the rest of this comment:

1. ResponseStore schema/conversation changes
2. ResponseRun lifecycle
3. ResponseRunManager
4. SSE writer / lifecycle policy
5. /v1/responses handler changes
6. GET / DELETE recovery and cleanup
7. Tests

1. ResponseStore: durable response state + safer conversation pointers

ResponseStore already stored completed responses for previous_response_id chaining. This PR extends its conversation bookkeeping.

Previously, a conversation mapped to a single response_id.

That was ambiguous because the mapped response could be:

• completed
• active/in-progress
• failed
• incomplete
• deleted/evicted

This PR changes the conversation schema to separate:

latest_completed_response_id
active_response_id

Design intent:

• latest_completed_response_id is the checkpoint used for future chaining.
• active_response_id is only a guard that prevents concurrent writes to the same named conversation.
• failed/incomplete responses should not become the latest conversation checkpoint.
• deleting or evicting a response should remove any conversation references to it.

Important methods:

• _migrate_conversations_schema()
  • migrates old conversations(name, response_id TEXT NOT NULL) tables
  • rebuilds the table if legacy response_id exists
  • this is important because SQLite cannot simply drop the old NOT NULL column
• get_conversation_latest_completed(...)
• get_conversation_active(...)
• set_conversation_completed(...)
• set_conversation_active(...)
• clear_conversation_active(...)
• clear_conversation_references(...)
• delete(...)
• LRU eviction path in put(...)

2. ResponseRun: the core execution abstraction

ResponseRun is the main new abstraction.

It represents one /v1/responses streaming execution and owns:

• response id / model / created_at
• agent task
• streaming callbacks
• emitted output items
• current response snapshot
• final status
• persistence when store=True
• conversation active/completed updates when applicable
• subscriber queues
• cancellation / interruption

The important design point is:

ResponseRun is independent of any single SSE client connection.

So for background runs, if an SSE client disconnects, the ResponseRun can continue.

Important methods:

start()

Starts the background asyncio task for the run.

subscribe(...)

Creates a per-subscriber queue and yields SSE events.

This is where a new subscriber can receive:

• response.created
• response.snapshot
• response.completed
• response.failed
• response.incomplete
• keepalive comments via __keepalive__

Design intent:

• active reattach uses response.snapshot, not fake replay
• completed/failed/incomplete runs emit final event immediately
• subscribers are removed cleanly in finally

_run()

Main lifecycle:

1. mark status in_progress
2. optionally persist an initial snapshot
3. optionally mark conversation active
4. broadcast response.created
5. run the agent task
6. drain tool/text callback events
7. finalize as completed/failed/incomplete
8. notify subscribers
9. unregister from manager

_callbacks()

Converts agent callbacks into internal run events:

• text deltas
• tool starts
• tool completions

These are queued into the event loop with _put_event_threadsafe(...).

_dispatch_event(...), _emit_text_delta(...), _emit_tool_started(...), _emit_tool_completed(...)

These convert internal callback events into Responses API-style SSE events.

_finalize_completed(...)

Builds final response object, persists it when store=True, and updates conversation checkpoint.

Key rule:

Only completed responses call set_conversation_completed(...)

_finalize_failed(...) and _finalize_incomplete(...)

Build failed/incomplete response objects and clear active conversation state, but do not advance latest_completed_response_id.

cancel(...)

Used by DELETE, shutdown, and ephemeral client disconnect.

Important detail:

• waits briefly for agent_ref to become available before cancelling the task
• this avoids a race where cancellation arrives before the agent object has been installed

3. ResponseRunManager: active durable run registry

ResponseRunManager tracks only active durable/recoverable runs.

It provides:

• register(run)
• unregister(response_id)
• get(response_id)
• can_start()
• shutdown()

Design intent:

• store=True streaming runs are registered
• store=False ephemeral runs are not registered
• GET can recover active durable runs
• shutdown can cancel all active durable runs
• concurrency limit applies to durable background runs

4. Unified SSE writer: _stream_response_run(...)

_stream_response_run(..., cancel_on_disconnect: bool)

The lifecycle policy is controlled by cancel_on_disconnect.

Durable background streaming

cancel_on_disconnect=False

Used for:

stream=true, store=true

Behavior:

• client disconnect only detaches that subscriber
• ResponseRun continues
• response can be recovered via GET

Ephemeral foreground streaming

cancel_on_disconnect=True

Used for:

stream=true, store=false

Behavior:

• same ResponseRun event generation
• same SSE formatting
• same keepalive handling
• but client disconnect cancels/interupts the run
• no persistence because ResponseRun(store=False)
• no recovery because it is not registered in ResponseRunManager

This was intentionally refactored this way to avoid two duplicated streaming implementations.

5. Streaming idempotency: _ResponseRunIdempotencyIndex

Streaming idempotency is separate from the existing non-streaming idempotency cache.

The new index maps:

Idempotency-Key + request fingerprint -> response_id

Used only for:

stream=true, store=true

Design intent:

• if same key/body is retried while run is active, attach to active run
• if same key/body is retried after completion, replay stored final response as SSE
• if same key/different body is retried, return conflict
• if response is deleted, remove idempotency mapping

Important methods:

• lookup(...)
• put(...)
• remove_response(...)

6. /v1/responses handler flow

The main request handler now does roughly this:

Step A: parse request and resolve conversation

If conversation is provided:

1. check active_response_id
2. reject if there is an active durable response
3. resolve latest_completed_response_id
4. use that as previous_response_id

Design intent:

• conversations cannot be concurrently advanced by multiple durable runs
• failed/incomplete responses do not poison future history
• stale deleted/evicted references are cleared

Step B: build conversation_history

Uses precedence:

explicit conversation_history > previous_response_id/conversation latest completed

For previous responses:

• only completed is valid
• legacy stored responses without status are treated as completed for compatibility

Step C: branch by stream

Non-streaming

Existing behavior mostly remains:

• run agent synchronously
• return JSON response
• persist if store=True
• update conversation completed if stored

Streaming + store=True

Creates a durable ResponseRun:

run = ResponseRun(..., store=True, conversation=conversation)
manager.register(run)
run.start()
return _stream_response_run(..., cancel_on_disconnect=False)

Also handles streaming idempotency before creating a new run.

Streaming + store=False

Creates an ephemeral ResponseRun:

run = ResponseRun(..., store=False, conversation=None)
run.start()
return _stream_response_run(..., cancel_on_disconnect=True)

Important detail:

• conversation_history may still be read from the named conversation before this
• but conversation=None is passed into ResponseRun
• therefore ephemeral streams cannot mutate conversation state

7. GET / DELETE behavior

GET /v1/responses/{response_id}

Lookup order:

1. ResponseStore
2. active ResponseRunManager
3. 404

Design intent:

• completed durable responses come from store
• active durable responses return live snapshot
• ephemeral responses return 404 because they are not stored or registered

DELETE /v1/responses/{response_id}

If response is active:

1. cancel run with delete=True
2. wait for completion
3. delete store entry
4. remove idempotency mapping

If response is stored:

1. delete store entry
2. remove idempotency mapping

8. Shutdown behavior

APIServerAdapter.disconnect() now shuts down active response runs.

Design intent:

• server shutdown should not leave durable background tasks running
• active runs are interrupted/cancelled cleanly

9. Test map

The tests in tests/gateway/test_api_server.py are intended to cover the main contracts.

Useful tests to inspect:

Background durable streaming

• client disconnect does not cancel durable run
• active GET snapshot works
• DELETE active run does not re-persist incomplete state
• shutdown cancels active response runs

Ephemeral streaming

• stream=true, store=false succeeds
• ephemeral response is not stored / GET returns 404
• ephemeral disconnect interrupts agent
• ephemeral stream with Idempotency-Key returns idempotency_requires_store

Idempotency

• same key/body reuses active run
• same key/body replays stored response
• same key/different body returns conflict
• delete removes mapping

Conversation safety

• non-completed previous_response_id rejected
• non-completed conversation checkpoint rejected
• active conversation blocks concurrent request
• failed durable stream does not poison latest completed checkpoint
• store=false conversation stream reads history but does not update checkpoint/active marker
• delete and LRU eviction clear conversation references

Migration

• legacy conversations.response_id TEXT NOT NULL schema is rebuilt
• legacy mapping is preserved as latest completed
• new active writes no longer fail with NOT NULL constraint

SSE details

• active reattach emits response.snapshot
• initial stream still emits normal events
• stored replay includes CORS
• keepalive is emitted
• cancellation race waits briefly for agent_ref