feat(filter): resume IRR iterations within streaming responses

docs/filters/http/traffic_management/iterative_request_router.md

@@ -9,13 +9,16 @@ Framework-level filter for iterative sub-request execution.
 
 Holds named steps, each backed by a pre-built sub-pipeline. During request processing, runs an iteration loop: execute each step's request filters, make the HTTP call via Pingora's `Connector`, execute its response filters, evaluate transition rules, and continue or return the final response.
 
+Streaming steps remain pull-based. Header-safe failover rules run before any bytes are exposed; all other `on_result` rules run after clean EOF and may resume another step inside the same committed downstream response.
+
 ## Configuration
 
 | Field | Type | Required | Description |
 |-------|------|---------|-------------|
 | `initial_step` | string | yes | Name of the first step to execute. |
 | `max_iterations` | integer | no | Maximum iterations before aborting (default 10, max 100). |
 | `max_response_bytes` | integer | no | Maximum response body bytes per sub-request. |
+| `max_stream_response_bytes` | integer | no | Optional cumulative byte ceiling for one logical streamed response. This is intentionally distinct from buffered per-step response limits. |
 | `max_state_bytes` | integer | no | Maximum accumulated iteration state bytes. |
 | `step_timeout_ms` | integer | no | Per-step timeout in milliseconds. Defaults to `timeout_ms`. |
 | `steps` | StepConfig[] | yes | Named steps, each with filters and transition rules. |
@@ -37,7 +40,7 @@ Holds named steps, each backed by a pre-built sub-pipeline. During request proce
 | `steps[].filters[].name` | string | no | Optional user-assigned name for this filter entry. Used as a rejoin target by branch chains. |
 | `steps[].filters[].response_conditions` | ResponseCondition[] | no | Ordered conditions that gate whether this filter runs on responses. Evaluated against the upstream response (status, headers). Empty means the filter always runs on responses. |
 | `steps[].filters[].failure_mode` | `closed` \| `open` | no | Per-filter failure behaviour (`open` or `closed`). |
-| `steps[].on_result` | StepTransition[] | no | Transition rules evaluated after the sub-request response. First match wins. |
+| `steps[].on_result` | StepTransition[] | no | Transition rules evaluated in order. Streaming header-safe failovers run before body exposure; remaining rules run after step completion. |
 | `steps[].on_result[].default` | bool | no | If true, this is the default (always-match) rule. |
 | `steps[].on_result[].filter` | string | no | Filter name whose results to check. |
 | `steps[].on_result[].key` | string | no | Result key to match. |

filter/src/actions.rs

@@ -8,7 +8,7 @@ use std::fmt;
 use async_trait::async_trait;
 use bytes::Bytes;
 
-use crate::FilterError;
+use crate::{FilterError, RequestExtensions};
 
 // -----------------------------------------------------------------------------
 // Streaming terminal response
@@ -37,6 +37,15 @@ pub trait StreamingResponseBody: Send + 'static {
     ///
     /// This operation must be idempotent.
     async fn cancel(&mut self);
+
+    /// Exchange request extensions with the protocol lifecycle owner.
+    ///
+    /// Most streaming bodies do not own filter extensions and use this
+    /// default no-op. Iterative sessions override it so the same extension
+    /// set can move between step filters and parent response filters without
+    /// cloning type-erased values.
+    #[doc(hidden)]
+    fn swap_extensions(&mut self, _extensions: &mut RequestExtensions) {}
 }
 
 /// A terminal response whose body is delivered incrementally.

filter/src/builtins/http/traffic_management/iterative_request_router/config.rs

@@ -74,6 +74,11 @@ pub(crate) struct IterativeRequestRouterConfig {
     #[serde(default = "default_max_response_bytes")]
     pub(crate) max_response_bytes: usize,
 
+    /// Optional cumulative byte ceiling for one logical streamed response.
+    /// This is intentionally distinct from buffered per-step response limits.
+    #[serde(default)]
+    pub(crate) max_stream_response_bytes: Option<usize>,
+
     /// Maximum accumulated iteration state bytes.
     #[serde(default = "default_max_state_bytes")]
     pub(crate) max_state_bytes: usize,
@@ -95,7 +100,7 @@ pub(crate) struct IterativeRequestRouterConfig {
 }
 
 /// A named step within the iterative router.
-#[derive(Debug, Deserialize)]
+#[derive(Clone, Debug, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub(crate) struct StepConfig {
     /// Step name (must be unique within the router).
@@ -104,14 +109,14 @@ pub(crate) struct StepConfig {
     /// Filters to execute for this step's sub-request.
     pub(crate) filters: Vec<crate::FilterEntry>,
 
-    /// Transition rules evaluated after the sub-request
-    /// response. First match wins.
+    /// Transition rules evaluated in order. Streaming header-safe failovers
+    /// run before body exposure; remaining rules run after step completion.
     #[serde(default)]
     pub(crate) on_result: Vec<StepTransition>,
 }
 
 /// A transition rule evaluated after a step completes.
-#[derive(Debug, Deserialize)]
+#[derive(Clone, Debug, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub(crate) struct StepTransition {
     /// If true, this is the default (always-match) rule.
@@ -227,6 +232,14 @@ pub(crate) fn validate(cfg: &IterativeRequestRouterConfig) -> Result<(), FilterE
             .into());
     }
 
+    if cfg.max_stream_response_bytes == Some(0) {
+        return Err(
+            "iterative_request_router: max_stream_response_bytes must be > 0 when configured"
+                .to_owned()
+                .into(),
+        );
+    }
+
     if cfg.max_response_bytes == 0 {
         return Err("iterative_request_router: max_response_bytes must be > 0"
             .to_owned()