fix(usage): preserve full bodies before queue truncation

crates/aether-usage/runtime/src/queue.rs

@@ -104,6 +104,13 @@ impl UsageQueue {
 
     pub async fn enqueue(&self, event: &UsageEvent) -> Result<String, DataLayerError> {
         let encoded = self.encode_event(event)?;
+        self.enqueue_encoded(encoded).await
+    }
+
+    pub(crate) async fn enqueue_encoded(
+        &self,
+        encoded: EncodedUsageEvent,
+    ) -> Result<String, DataLayerError> {
         self.runner
             .append_fields_with_maxlen(
                 &self.stream,
@@ -117,7 +124,10 @@ impl UsageQueue {
         self.encode_event(event).map(|_| ())
     }
 
-    fn encode_event(&self, event: &UsageEvent) -> Result<EncodedUsageEvent, DataLayerError> {
+    pub(crate) fn encode_event(
+        &self,
+        event: &UsageEvent,
+    ) -> Result<EncodedUsageEvent, DataLayerError> {
         let encoded = match event.to_bounded_stream_fields(self.config.queue_payload_max_bytes) {
             Ok(encoded) => encoded,
             Err(error) => {

crates/aether-usage/runtime/src/runtime.rs

@@ -4919,7 +4919,7 @@ impl UsageRuntime {
         &self,
         data: &T,
         queue: UsageQueue,
-        event: UsageEvent,
+        mut event: UsageEvent,
     ) -> TerminalPersistenceOutcome
     where
         T: UsageRuntimeAccess,
@@ -4952,7 +4952,20 @@ impl UsageRuntime {
                 .await;
         };
 
-        if let Err(err) = queue.enqueue(&event).await {
+        let enqueue_result = match queue.encode_event(&event) {
+            Ok(encoded) => {
+                if encoded.diagnostics_omitted
+                    && self
+                        .try_write_terminal_direct_fallback(data, &mut event, "queue_wire_limit")
+                        .await
+                {
+                    return TerminalPersistenceOutcome::PersistedDirectly;
+                }
+                queue.enqueue_encoded(encoded).await
+            }
+            Err(err) => Err(err),
+        };
+        if let Err(err) = enqueue_result {
             drop(_guard);
             if is_permanent_enqueue_error(&err) {
                 return self
@@ -12725,6 +12738,159 @@ mod tests {
         assert_eq!(queued.data.total_cost_usd, None);
     }
 
+    fn oversized_full_terminal_event(max_bytes: usize) -> (serde_json::Value, UsageEvent) {
+        let body = json!({"content": "full-body".repeat(max_bytes / 16)});
+        let mut event = UsageEvent::new(
+            UsageEventType::Completed,
+            "oversized-full-capture",
+            UsageEventData {
+                provider_name: "openai".to_string(),
+                model: "gpt-5".to_string(),
+                status_code: Some(200),
+                total_tokens: Some(12),
+                request_body: Some(body.clone()),
+                provider_request_body: Some(body.clone()),
+                response_body: Some(body.clone()),
+                client_response_body: Some(body.clone()),
+                ..UsageEventData::default()
+            },
+        );
+        apply_usage_body_capture_policy_to_event(
+            UsageBodyCapturePolicy {
+                record_level: UsageRequestRecordLevel::Full,
+            },
+            &mut event,
+        );
+        (body, event)
+    }
+
+    #[tokio::test]
+    async fn oversized_full_terminal_capture_is_persisted_without_queue_truncation() {
+        let config = UsageRuntimeConfig {
+            enabled: true,
+            queue_terminal_events: true,
+            consumer_block_ms: 1,
+            ..UsageRuntimeConfig::default()
+        };
+        let queue_runner: Arc<dyn RuntimeQueueStore> =
+            Arc::new(RuntimeState::memory(MemoryRuntimeStateConfig::default()));
+        let queue = UsageQueue::new(Arc::clone(&queue_runner), config.clone()).unwrap();
+        let store = EnrichmentCountingQueueStore {
+            records: Mutex::new(Vec::new()),
+            queue: queue_runner,
+            enrich_calls: AtomicUsize::new(0),
+        };
+        let runtime = UsageRuntime::new(config.clone()).unwrap();
+        let (body, event) = oversized_full_terminal_event(config.queue_payload_max_bytes);
+        assert!(
+            event
+                .to_bounded_stream_fields(config.queue_payload_max_bytes)
+                .unwrap()
+                .diagnostics_omitted
+        );
+
+        let outcome = runtime.enqueue_or_write_terminal(&store, event).await;
+
+        assert_eq!(
+            outcome,
+            super::TerminalPersistenceOutcome::PersistedDirectly
+        );
+        assert_eq!(store.enrich_calls.load(Ordering::Acquire), 1);
+        assert_eq!(queue.stats().await.unwrap().stream_length, 0);
+        let records = store.records.lock().unwrap();
+        assert_eq!(records.len(), 1);
+        for captured in [
+            &records[0].request_body,
+            &records[0].provider_request_body,
+            &records[0].response_body,
+            &records[0].client_response_body,
+        ] {
+            assert_eq!(captured.as_ref(), Some(&body));
+        }
+    }
+
+    #[tokio::test]
+    async fn oversized_full_terminal_capture_keeps_bounded_queue_fallback() {
+        for unavailable in ["writer", "write_failure", "worker_gate", "fallback_gate"] {
+            let config = UsageRuntimeConfig {
+                enabled: true,
+                queue_terminal_events: true,
+                consumer_block_ms: 1,
+                worker_record_concurrency_limit: Some(1),
+                ..UsageRuntimeConfig::default()
+            };
+            let queue_runner: Arc<dyn RuntimeQueueStore> =
+                Arc::new(RuntimeState::memory(MemoryRuntimeStateConfig::default()));
+            let queue = UsageQueue::new(Arc::clone(&queue_runner), config.clone()).unwrap();
+            queue.ensure_consumer_group().await.unwrap();
+            let store = FailingWriteQueueConfiguredUsageStore {
+                queue: Arc::clone(&queue_runner),
+                upsert_attempts: Arc::new(AtomicUsize::new(0)),
+            };
+            let queue_only = QueueOnlyUsageStore {
+                queue: queue_runner,
+                upsert_attempts: Arc::clone(&store.upsert_attempts),
+            };
+            let runtime = UsageRuntime::new(config.clone()).unwrap();
+            let worker_permit = (unavailable == "worker_gate").then(|| {
+                runtime
+                    .worker_record_gate
+                    .as_ref()
+                    .unwrap()
+                    .try_acquire()
+                    .unwrap()
+            });
+            let fallback_permit = (unavailable == "fallback_gate").then(|| {
+                runtime
+                    .terminal_direct_fallback_state
+                    .try_acquire()
+                    .unwrap()
+            });
+            let (_, event) = oversized_full_terminal_event(config.queue_payload_max_bytes);
+
+            let outcome = if unavailable == "writer" {
+                runtime.enqueue_or_write_terminal(&queue_only, event).await
+            } else {
+                runtime.enqueue_or_write_terminal(&store, event).await
+            };
+
+            assert_eq!(
+                outcome,
+                super::TerminalPersistenceOutcome::Queued,
+                "{unavailable}"
+            );
+            assert_eq!(
+                store.upsert_attempts.load(Ordering::Acquire),
+                usize::from(unavailable == "write_failure")
+            );
+            let entries = queue
+                .read_group("oversized-capture-consumer")
+                .await
+                .unwrap();
+            assert_eq!(entries.len(), 1);
+            assert!(entries[0].fields["payload"].len() <= config.queue_payload_max_bytes);
+            let queued = UsageEvent::from_stream_fields(&entries[0].fields).unwrap();
+            assert_eq!(queued.data.total_tokens, Some(12));
+            for (body, state) in [
+                (&queued.data.request_body, queued.data.request_body_state),
+                (
+                    &queued.data.provider_request_body,
+                    queued.data.provider_request_body_state,
+                ),
+                (&queued.data.response_body, queued.data.response_body_state),
+                (
+                    &queued.data.client_response_body,
+                    queued.data.client_response_body_state,
+                ),
+            ] {
+                assert!(body.is_none());
+                assert_eq!(state, Some(UsageBodyCaptureState::Truncated));
+            }
+            assert_eq!(runtime.metrics_snapshot().terminal_enqueue_failed_total, 0);
+            drop((worker_permit, fallback_permit));
+        }
+    }
+
     #[tokio::test]
     async fn terminal_enqueue_failure_uses_bounded_direct_database_fallback() {
         let config = UsageRuntimeConfig {