feat(routing): mid-stream fallback — resume a committed stream on fallback targets

crates/aisix-admin/src/openapi.rs

@@ -4343,6 +4343,19 @@ fn add_variant_titles(doc: &mut Value) {
             "/components/schemas/StreamDoneMarker/oneOf",
             &["Required", "Optional", "None"],
         ),
+        (
+            "/components/schemas/StreamFailureMode/oneOf",
+            &["Terminate", "Continue"],
+        ),
+        (
+            "/components/schemas/StreamFailureTrigger/oneOf",
+            &[
+                "Transport error",
+                "Read timeout",
+                "Upstream decode error",
+                "Upstream in-band error",
+            ],
+        ),
     ];
 
     for (pointer, titles) in variant_titles {

crates/aisix-core/src/lib.rs

@@ -49,8 +49,9 @@ pub use models::{
     GuardrailMonitorHit, KeywordConfig, KeywordPattern, McpAuthType, McpRateLimit, McpServer,
     McpServerType, McpTransport, Model, ObservabilityExporter, ParamConstraints, PolicyScope,
     PolicyWindow, ProviderKey, RateLimit, RateLimitPolicy, RequestOverrides, ResponseOverrides,
-    Routing, RoutingStrategy, RoutingTarget, SchemaError, StreamDoneMarker, TelemetryKind,
-    TelemetryTags, WhenAllUnavailablePolicy, DEFAULT_COOLDOWN_TRIGGER_STATUSES,
+    Routing, RoutingStrategy, RoutingTarget, SchemaError, StreamDoneMarker, StreamFailure,
+    StreamFailureMode, StreamFailureTrigger, TelemetryKind, TelemetryTags,
+    WhenAllUnavailablePolicy, DEFAULT_COOLDOWN_TRIGGER_STATUSES,
 };
 pub use resource::{Resource, ResourceEntry};
 pub use snapshot::{ResourceTable, SnapshotHandle};

crates/aisix-core/src/models/mod.rs

@@ -63,7 +63,10 @@ pub use provider_key::{
 };
 pub use rate_limit::{McpRateLimit, RateLimit};
 pub use rate_limit_policy::{PolicyScope, PolicyWindow, RateLimitPolicy};
-pub use routing::{Routing, RoutingStrategy, RoutingTarget, WhenAllUnavailablePolicy};
+pub use routing::{
+    Routing, RoutingStrategy, RoutingTarget, StreamFailure, StreamFailureMode,
+    StreamFailureTrigger, WhenAllUnavailablePolicy,
+};
 pub use schema::{
     validate_a2a_agent, validate_a2a_agent_lenient, validate_apikey, validate_apikey_lenient,
     validate_cache_policy, validate_cache_policy_lenient, validate_guardrail,

crates/aisix-core/src/models/routing.rs

@@ -184,6 +184,90 @@ pub struct Routing {
     /// default). Ignored by non-`weighted` strategies.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub sticky: Option<bool>,
+    /// What to do when a streaming response fails AFTER its first chunk was
+    /// already delivered to the client (the HTTP 200 is committed and cannot
+    /// be revised). Omitted keeps the historical behavior: terminate the
+    /// stream with an in-band error frame and no `[DONE]`.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub stream_failure: Option<StreamFailure>,
+}
+
+/// Mid-stream failure policy for streaming responses. Applies only to
+/// failures that occur after the response head (and possibly some
+/// chunks) reached the client; failures before the first chunk keep
+/// using the regular retry/failover loop.
+#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)]
+pub struct StreamFailure {
+    /// `terminate` (default) keeps the current behavior. `continue` lets
+    /// the router call the remaining fallback targets and resume the SAME
+    /// client stream with a best-effort continuation of the partial text.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub mode: Option<StreamFailureMode>,
+    /// Which mid-stream error classes trigger the fallback. Omitted =
+    /// all of them. Non-retryable errors (an in-band 4xx other than 429,
+    /// unless listed in `fallback_on_statuses`) never trigger regardless.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub on: Option<Vec<StreamFailureTrigger>>,
+    /// Max fallback targets tried for one mid-stream failure. Defaults
+    /// to 1 — mid-stream recovery burns client-visible latency per
+    /// attempt, so the default is deliberately tighter than the
+    /// pre-stream `max_fallbacks`.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub max_fallbacks: Option<u32>,
+}
+
+impl StreamFailure {
+    pub fn mode_or_default(&self) -> StreamFailureMode {
+        self.mode.unwrap_or_default()
+    }
+
+    pub fn max_fallbacks_or_default(&self) -> u32 {
+        self.max_fallbacks.unwrap_or(1)
+    }
+
+    /// Configured trigger classes; all classes when unset. `continue`
+    /// is itself the explicit opt-in, so the default set is the full
+    /// one rather than a conservative subset.
+    pub fn on_or_default(&self) -> &[StreamFailureTrigger] {
+        const ALL: &[StreamFailureTrigger] = &[
+            StreamFailureTrigger::TransportError,
+            StreamFailureTrigger::ReadTimeout,
+            StreamFailureTrigger::UpstreamDecodeError,
+            StreamFailureTrigger::UpstreamInBandError,
+        ];
+        self.on.as_deref().unwrap_or(ALL)
+    }
+}
+
+/// How a mid-stream failure is handled once the response is already
+/// streaming to the client.
+#[derive(
+    Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, schemars::JsonSchema,
+)]
+#[serde(rename_all = "snake_case")]
+pub enum StreamFailureMode {
+    /// Terminate the stream: in-band error frame, no `[DONE]` (the
+    /// historical behavior).
+    #[default]
+    Terminate,
+    /// Continue on a fallback target inside the same client stream.
+    Continue,
+}
+
+/// Mid-stream error classes eligible for [`StreamFailureMode::Continue`].
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
+#[serde(rename_all = "snake_case")]
+pub enum StreamFailureTrigger {
+    /// The upstream connection broke mid-stream (reset, premature close).
+    TransportError,
+    /// The gap between chunks exceeded the effective `stream_timeout`.
+    ReadTimeout,
+    /// A frame failed to parse as a chunk (and was not a recognizable
+    /// in-band error envelope).
+    UpstreamDecodeError,
+    /// The provider reported an error inside the committed 200 stream
+    /// (an SSE error frame / event-stream modeled exception).
+    UpstreamInBandError,
 }
 
 impl Routing {
@@ -271,6 +355,7 @@ mod tests {
             fallback_on_statuses: None,
             when_all_unavailable: None,
             sticky: None,
+            stream_failure: None,
         };
         assert_eq!(r.max_fallbacks_or_default(), 0);
     }
@@ -286,6 +371,7 @@ mod tests {
             fallback_on_statuses: None,
             when_all_unavailable: None,
             sticky: None,
+            stream_failure: None,
         };
         assert_eq!(r.max_fallbacks_or_default(), 0);
     }

crates/aisix-obs/src/metrics.rs

@@ -79,6 +79,12 @@ pub const M_DEPLOYMENT_STATE: &str = "aisix_deployment_state";
 pub const M_DEPLOYMENT_COOLED_DOWN_TOTAL: &str = "aisix_deployment_cooled_down_total";
 pub const M_ROUTING_SUCCESSFUL_FALLBACKS_TOTAL: &str = "aisix_routing_successful_fallbacks_total";
 pub const M_ROUTING_FAILED_FALLBACKS_TOTAL: &str = "aisix_routing_failed_fallbacks_total";
+/// Mid-stream fallback outcomes (`routing.stream_failure: continue`,
+/// AISIX-Cloud#1222). `outcome` is `recovered` (the client stream
+/// completed on a fallback target) or `failed` (every eligible fallback
+/// target also failed and the stream terminated). Labelled by the
+/// requested (routing) model.
+pub const M_MID_STREAM_FALLBACKS_TOTAL: &str = "aisix_mid_stream_fallbacks_total";
 pub const M_RATELIMIT_REMAINING_REQUESTS: &str = "aisix_ratelimit_remaining_requests";
 pub const M_RATELIMIT_REMAINING_TOKENS: &str = "aisix_ratelimit_remaining_tokens";
 pub const M_BUDGET_LIMIT_USD: &str = "aisix_budget_limit_usd";
@@ -1181,6 +1187,20 @@ impl Metrics {
         });
     }
 
+    /// One mid-stream fallback episode resolved (AISIX-Cloud#1222):
+    /// `recovered` when the client stream completed on a fallback
+    /// target, `failed` when the fallback chain was exhausted.
+    pub fn record_mid_stream_fallback(&self, model: &str, recovered: bool) {
+        metrics::with_local_recorder(&self.inner.recorder, || {
+            metrics::counter!(
+                M_MID_STREAM_FALLBACKS_TOTAL,
+                "model" => model.to_string(),
+                "outcome" => if recovered { "recovered" } else { "failed" },
+            )
+            .increment(1);
+        });
+    }
+
     pub fn set_rate_limit_remaining(
         &self,
         api_key_id: &str,

crates/aisix-obs/src/usage.rs

@@ -195,6 +195,22 @@ pub struct UsageEvent {
     #[serde(default, skip_serializing_if = "String::is_empty")]
     pub finish_reason: String,
 
+    /// Logical outcome of a STREAMING response, set on the serving
+    /// attempt's event only (AISIX-Cloud#1222). The HTTP status is
+    /// committed at 200 before the stream runs, so it cannot express a
+    /// mid-stream failure; this field can:
+    /// - `success` — the stream completed normally;
+    /// - `partial_failed` — the stream terminated after the 200 with an
+    ///   in-band error frame (`[DONE]` withheld);
+    /// - `partial_recovered` — a mid-stream failure was recovered by
+    ///   `routing.stream_failure: continue` on a fallback target and the
+    ///   stream then completed normally.
+    ///
+    /// Empty on non-streaming events, failed-attempt events, and
+    /// client-abandoned streams (those keep `status_code: 499`).
+    #[serde(default, skip_serializing_if = "String::is_empty")]
+    pub stream_outcome: String,
+
     /// Cost the DP computed for this request in US dollars. Zero when
     /// the request never reached cost calculation (e.g. blocked by a
     /// guardrail before dispatch). cp-api recomputes this server-side