fix(token_usage): recover from SSE overflow instead of silently dropping usage

docs/filters/reference.md

@@ -91,4 +91,4 @@ see the [Praxis core filter reference][core-ref].
 | Filter | Description |
 |--------|-------------|
 | [`token_count`](token_count.md) | Extracts token usage from AI inference responses and writes unified counts to [`filter_metadata`]. |
-| [`token_usage_headers`](token_usage_headers.md) | Injects `Praxis-Token-Input`, `Praxis-Token-Output`, and `Praxis-Token-Total` headers into downstream responses when token usage data is present in [`filter_metadata`]. |
+| [`token_usage_headers`](token_usage_headers.md) | Injects `Praxis-Token-Input`, `Praxis-Token-Output`, and `Praxis-Token-Total` headers into downstream responses when token usage data is present in [`filter_metadata`]. Also injects `Praxis-Token-Status` when usage capture failed (e.g. overflow), so an unavailable count is never silently indistinguishable from a genuine zero. |

docs/filters/token_count.md

@@ -14,10 +14,14 @@ Supports both streaming (SSE) and non-streaming (JSON) responses across five pro
 | Field | Type | Required | Description |
 |-------|------|---------|-------------|
 | `provider` | `openai` \| `anthropic` \| `google` \| `bedrock` \| `bedrock_invoke_model` \| `azure` | yes | AI provider whose response format to parse. |
+| `max_body_bytes` | integer | no | Maximum bytes to buffer for a non-streaming JSON response before giving up on locating its usage field. Must be greater than 0 and at most 64 MiB. |
+| `max_scratch_bytes` | integer | no | Maximum scratch bytes (buffered line + in-progress event data) for the SSE scanner before an event is discarded as oversized. Must be greater than 0 and at most 64 MiB. |
 
 ## Example
 
 ```yaml
 filter: token_count
 provider: openai   # openai | anthropic | google | bedrock | bedrock_invoke_model | azure
+max_body_bytes: 1048576    # optional, JSON capture limit
+max_scratch_bytes: 65536   # optional, SSE per-event capture limit
 ```

docs/filters/token_usage_headers.md

@@ -3,7 +3,7 @@
 
 # `token_usage_headers`
 
-Injects `Praxis-Token-Input`, `Praxis-Token-Output`, and `Praxis-Token-Total` headers into downstream responses when token usage data is present in [`filter_metadata`].
+Injects `Praxis-Token-Input`, `Praxis-Token-Output`, and `Praxis-Token-Total` headers into downstream responses when token usage data is present in [`filter_metadata`]. Also injects `Praxis-Token-Status` when usage capture failed (e.g. overflow), so an unavailable count is never silently indistinguishable from a genuine zero.
 
 ## Configuration Notes
 

filters/src/agentic/a2a/mod.rs

@@ -845,16 +845,15 @@ fn process_sse_response_chunk(
         try_extract_task_from_sse_payload(payload, ctx, store, config);
     }
 
-    if result.overflowed {
+    if result.dropped_events > 0 {
         debug!(
-            scratch_bytes = state.scratch_bytes,
+            dropped_events = result.dropped_events,
             max_bytes = config.max_response_body_bytes,
-            "SSE scratch exceeds capture limit, disabling streaming capture"
+            "SSE event exceeded capture limit, discarding and resuming at next event boundary"
         );
-        clear_sse_capture_metadata(ctx);
-    } else {
-        save_sse_scan_state(ctx, &state);
     }
+
+    save_sse_scan_state(ctx, &state);
 }
 
 /// Drains any incomplete SSE event buffered in `filter_metadata` at
@@ -896,6 +895,7 @@ fn try_extract_task_from_sse_payload(
 /// Reconstructs scanner state from hex-encoded `filter_metadata` keys.
 /// Metadata bypasses the 256-byte dynamic-value helper because the
 /// scanner buffers raw SSE line/data bytes that can exceed that limit.
+#[expect(clippy::too_many_lines, reason = "sequential per-field metadata reads")]
 fn load_sse_scan_state(ctx: &HttpFilterContext<'_>) -> sse::SseScanState {
     let line_buf = ctx
         .filter_metadata
@@ -925,12 +925,21 @@ fn load_sse_scan_state(ctx: &HttpFilterContext<'_>) -> sse::SseScanState {
         .and_then(|v| v.parse().ok())
         .unwrap_or(0);
 
+    let skip = sse::SkipPhase::from_metadata_str(ctx.filter_metadata.get("a2a.response.sse_skip").map(String::as_str));
+    let tail = sse::ScanTail::from_metadata_str(
+        ctx.filter_metadata
+            .get("a2a.response.sse_dropped_tail")
+            .map(String::as_str),
+    );
+
     sse::SseScanState {
         line_buf,
         data_buf,
         has_data,
         prev_cr,
         scratch_bytes,
+        skip,
+        tail,
     }
 }
 
@@ -951,6 +960,13 @@ fn save_sse_scan_state(ctx: &mut HttpFilterContext<'_>, state: &sse::SseScanStat
         "a2a.response.sse_scratch_bytes".to_owned(),
         state.scratch_bytes.to_string(),
     );
+
+    ctx.filter_metadata
+        .insert("a2a.response.sse_skip".to_owned(), state.skip.as_str().to_owned());
+    ctx.filter_metadata.insert(
+        "a2a.response.sse_dropped_tail".to_owned(),
+        state.tail.as_str().to_owned(),
+    );
 }
 
 /// Hex-encodes raw bytes into a metadata value, or removes the key if empty.
@@ -975,10 +991,10 @@ fn clear_sse_capture_metadata(ctx: &mut HttpFilterContext<'_>) {
     ctx.filter_metadata.remove("a2a.response.sse_has_data");
     ctx.filter_metadata.remove("a2a.response.sse_prev_cr");
     ctx.filter_metadata.remove("a2a.response.sse_scratch_bytes");
+    ctx.filter_metadata.remove("a2a.response.sse_skip");
     ctx.filter_metadata.remove("a2a.response.cluster");
 }
 
-
 /// Build a `JsonRpcConfig` for the shared parser with A2A-appropriate defaults.
 fn build_json_rpc_config(max_body_bytes: usize) -> JsonRpcConfig {
     use praxis_filter::builtins::http::payload_processing::json_rpc::config::{BatchPolicy, JsonRpcHeaders};