Abort session on client disconnect (HTTP server)

[AgrawalAmey]

Problem

When an HTTP client disconnects mid-generation, Vajra continues running the session to completion, wasting GPU cycles on output nobody is listening for. The only current disconnect signal is the SIGPIPE-to-EPIPE trick in server.py:167-172, which fires only when the C++ writer next attempts a pipe write — on slow-generating sessions that can mean seconds of wasted compute per dropped client.

AsyncEngine.abort() at vajra/entrypoints/common/async_engine.py:180-181 is literally a no-op:

async def abort(self, request_id: str) -> None:
    """Abort a request (not implemented in base engine yet)."""

Current state

• No path from FastAPI noticing TCP close → telling the C++ engine to stop.
• request.is_disconnected() is unreliable anyway because AuthMiddleware inherits from BaseHTTPMiddleware, which breaks the receive() contract (see sibling issue on middleware refactor).
• Non-streaming handlers have no cancellation wrapper.

Desired behavior

1. Client TCP close → engine stops generating within ~one scheduler tick.
2. KV-cache blocks held by the session are freed promptly.
3. Works for streaming (SSE pipe reads), non-streaming (FastAPI handlers), and the Responses API (which also cleans up its in-memory conversation state placeholder).

Implementation sketch

1. C++ side: add InferenceEngine::AbortSession(session_id) that enqueues an abort request into the ComputeManager. On the next scheduler tick, the session is dropped and its KV blocks are returned to the allocator.
2. Pybind: expose the method from vajra/_native/engine.
3. AsyncEngine.abort(): replace the no-op with await asyncio.to_thread(self.engine.abort_session, request_id).
4. Streaming path (async_engine.py:generate_sse_stream): wrap the pipe-read loop in a try/finally that calls abort() on cancellation.
5. Non-streaming path: add a with_cancellation decorator to routes, modeled on vLLM's pattern at vllm/entrypoints/utils.py:34-72 — races the handler against a listen_for_disconnect task via asyncio.wait(..., return_when=FIRST_COMPLETED) and cancels the loser.
6. Responses API: on abort, also call _remove_conversation_state(request_id) so we don't leave orphan placeholders.

Reference

• vLLM disconnect-to-abort: vllm/entrypoints/utils.py:34-72 (decorator), vllm/v1/engine/async_llm.py:450-456, 516-524 (streaming abort propagation).
• SGLang: StreamingResponse(..., background=create_abort_task(...)) hook at python/sglang/srt/entrypoints/http_server.py:1171 plus explicit is_disconnected() polling in tokenizer_manager.py:1141-1148.

Acceptance criteria

• AsyncEngine.abort() actually stops the engine — verifiable by test that drops a streaming connection and asserts the engine session is dead within N seconds.
• A slow-generating request whose client drops does not monopolize GPU for the remainder of the session.
• Unit tests cover: streaming disconnect, non-streaming disconnect, disconnect during Responses conversation state placeholder.

Blocks / related

• Depends on auth middleware moving off BaseHTTPMiddleware for reliable request.receive().