Epic: stream subrequest responses through iterative filter pipelines

[leseb]

Summary

Add end-to-end response streaming for Praxis subrequests, including long-lived
SSE responses, without replacing or weakening the buffered SubResponse API.

This epic builds on:

• iterative_request_router: failover and agentic loop support #849, which introduced the Pingora-native iterative subrequest path.
• refactor(core): move SubRequest and SubResponse to core crate #868, which moved SubRequest and SubResponse into praxis-core.
• core: add a hardened SubRequestClient over SubRequestConnector #826, which exposed the hardened buffered SubRequestClient.

The streaming API should be additive:

• SubResponse remains buffered, cloneable, and suitable for ordinary
  subrequests.
• A separate streaming response owns the active transport session, admission
  permit, cancellation state, limits, and response body.
• Buffered execution should eventually be implemented by collecting the
  streaming response through the same hardened transport lifecycle.

A representative core interface is:

pub struct StreamingSubResponse {
    pub status: u16,
    pub headers: HeaderMap,
    pub body: SubResponseBody,
}

impl SubResponseBody {
    pub async fn next_chunk(
        &mut self,
    ) -> Result<Option<Bytes>, SubRequestError>;

    pub async fn cancel(self);
}

impl StreamingSubResponse {
    pub async fn collect(
        self,
        max_bytes: usize,
    ) -> Result<SubResponse, SubRequestError>;
}

impl SubRequestClient {
    pub async fn send_streaming(
        &self,
        peer: HttpPeer,
        request: SubRequest,
        limits: StreamLimits,
    ) -> Result<StreamingSubResponse, SubRequestError>;
}

StreamLimits should distinguish bounded memory from total response length:

• Maximum buffered/in-flight bytes.
• Optional maximum total bytes.
• Idle timeout.
• Optional overall timeout.

Whether a subrequest response is buffered or streamed is selected by filters,
not by Praxis configuration. The response mode defaults to buffered. A filter
that constructs the final outbound request may select streaming through a
typed HttpFilterContext API after deciding the effective request body and
whether the response can be exposed directly. Filters that may make this
selection should declare that capability so IRR can validate their step
pipelines conservatively.

For the OpenAI Responses API, the AI filters own this decision. They may use
the client request's stream value and orchestration state as inputs, but the
filter that constructs the effective outbound request must keep its serialized
"stream": true value and the typed Praxis response-mode selection aligned.
Praxis core and IRR must not parse OpenAI JSON or depend on AI metadata keys.

Streaming must preserve the Praxis filter lifecycle. The iterative request
router must run response-header filters before deciding whether to transition
or expose the response. Once downstream headers are committed, failover is no
longer possible; later transport or filter failures terminate the stream.

Streaming should initially be supported only for a response that is terminal
once selected. IRR may still perform header-safe failover before exposing that
response. Intermediate rounds remain buffered because their bodies may be
needed by later transitions or steps.

Body-dependent transitions and filters requiring BodyMode::StreamBuffer must
be rejected during configuration validation when a step pipeline declares
that it may select streaming. Equivalent runtime guards must remain in place
for dynamically registered filters. Praxis must not silently turn a
filter-selected stream back into a buffered response.

Core owns transport mechanics, backpressure, resource cleanup, and typed
transport errors. The filter and protocol layers continue to own filter
execution, transition policy, downstream framing, and HTTP semantics.

Non-goals

• Streaming request bodies.
• Streaming intermediate IRR rounds whose response bodies affect control flow.
• Retrying or failing over after downstream headers are committed.
• A Praxis YAML option that decides whether a client requested streaming.
• Parsing provider-specific request bodies or metadata in Praxis core or IRR.
• SSE parsing or provider-specific transformation in praxis-core.
• Routing, SSRF policy, retry policy, or HTTP-status interpretation in
  SubRequestClient.
• WebSocket or HTTP upgrade tunnelling.
• Replacing SubResponse with a buffered/streaming enum.
• Silently buffering streams for incompatible filters.

Success Criteria

• praxis-core exposes an opaque streaming response API that does not expose
  Pingora session mechanics.
• Streaming and buffered requests share the same hardened connection, framing,
  sanitization, admission, deadline, and error-handling implementation.
• Response chunks reach the downstream client incrementally and are subject to
  real backpressure.
• Long-lived SSE responses are not constrained by the buffered-response ceiling
  unless an explicit total limit is configured.
• Idle and overall deadlines have distinct, documented behavior.
• Cancellation, timeout, downstream disconnect, early drop, and body errors
  release the admission permit and safely close or reuse the upstream
  connection.
• HTTP/1.1 incomplete responses are never returned to the pool; HTTP/2
  cancellation resets only the affected stream.
• The filter API exposes a typed, per-subrequest buffered/streaming selection;
  buffered remains the default.
• IRR uses the filter-selected mode without interpreting OpenAI request bodies
  or AI metadata keys.
• Streaming IRR transitions can depend only on information available before the
  body is exposed, such as status, origin, transport error, or default rules.
• Step response-header filters run before transition evaluation.
• Step and parent response-body filters run once per chunk with valid owned
  state across the stream lifetime.
• Response-body completion hooks run exactly once.
• No transition or fallback occurs after downstream response headers are
  committed.
• Protocol handling correctly implements HEAD, 204, and 304 body suppression
  and HTTP/1.1 versus HTTP/2 framing.
• Incompatible body-buffering filters and body-dependent transitions fail
  configuration validation for pipelines that declare streaming capability,
  with runtime guards for dynamic cases.
• Integration tests use a test-only mode-selecting filter and prove that SSE
  data arrives before the upstream response completes.
• The public filter contract is documented sufficiently for Praxis AI to
  select streaming without a separate callout client.
• Praxis AI provides the functional Responses SSE example when it adopts the
  stable API; Praxis does not add a production streaming-mode configuration
  filter solely for an example.

Sub-Tasks

• Complete the buffered SubRequestClient tracked by core: add a hardened SubRequestClient over SubRequestConnector #826.
• Add the core streaming response, body, limits, and cancellation API.
• Implement bounded backpressure and correct Pingora session/permit
  ownership.
• Reimplement buffered response collection on top of the streaming
  transport path.
• Add a typed filter-to-IRR response-mode contract and streaming-capability
  declaration.
• Add a terminal streaming action to the filter API.
• Add protocol support for incrementally forwarding terminal response
  bodies.
• Introduce owned IRR response-phase state that can safely outlive
  on_request().
• Run IRR step and parent response filters incrementally over streamed
  chunks.
• Read the filter-selected response mode after step request-body filters
  finish constructing the effective outbound request.
• Restrict streaming transitions to response-header-safe predicates and
  add static plus runtime compatibility validation.
• Add tracing and metrics for stream duration, bytes, cancellation,
  timeouts, and termination causes.
• Add core tests for backpressure, limits, timeout behavior, cancellation,
  cleanup, and HTTP/1.1 and HTTP/2 reuse.
• Add filter tests for per-chunk execution, state preservation, completion
  hooks, and incompatible body modes.
• Add integration tests with a test-only mode selector covering SSE
  latency, framing, header-based failover, downstream disconnects, and late
  errors.
• Document the public streaming-selection contract and lifecycle.
• Migrate Praxis AI to the stable API, keeping internal body-dependent
  rounds buffered and adding the functional Responses SSE example there.

Area

Filter Pipeline

[leseb]

cc @shaneutt @usize

[leseb]

I initially self-assigned but thought it'd be better for someone from core to take it, otherwise I can still do it :)

[shaneutt]

cc @praxis-proxy/core-reviewers