fix(conversations): fail closed on append-back item loss (#837)

apis/src/openai/conversations/filter.rs

@@ -92,6 +92,22 @@ impl OpenaiConversationsFilter {
         }
     }
 
+    /// Build a filter around a pre-initialized store for tests.
+    ///
+    /// Pre-seeding the `OnceCell` lets tests inject a fault-injecting store
+    /// (e.g. one whose `create_conversation_items` fails) without standing up a
+    /// real database, so append-back error handling can be exercised directly.
+    #[cfg(test)]
+    pub(super) fn with_store_for_test(config: ConversationsConfig, store: Arc<dyn ConversationItemStore>) -> Self {
+        Self {
+            config,
+            // Pre-initialize the cell so `get_or_init_store` returns this store
+            // without touching a real backend. The outer `Some` marks the cell
+            // initialized; the inner `Some` is the stored (available) store.
+            store: OnceCell::new_with(Some(Some(store))),
+        }
+    }
+
     /// Build the configured store backend.
     async fn build_store(&self) -> Result<Arc<dyn ConversationItemStore>, StoreError> {
         let responses_table = self.config.responses_table();
@@ -474,9 +490,22 @@ impl HttpFilter for OpenaiConversationsFilter {
         };
 
         let conv_id = items.conversation_id;
-        if let Err(e) = self.append_items_blocking(&items.tenant_id, &conv_id, ctx, items.all_items) {
-            warn!(error = %e, conversation_id = %conv_id, "conversation append-back failed");
-        }
+        // Fail closed on lost items. Append-back runs at end-of-stream while the
+        // completed response body is still buffered (StreamBuffer), before any
+        // byte is released downstream. Under the default `failure_mode: closed`,
+        // the buffered body is never released after a persistence failure, so the
+        // client cannot observe a clean success that hides items which never
+        // persisted (#837). The exact downstream outcome is Pingora-timing-dependent
+        // — a not-yet-flushed header yields a clean 500, an already-committed one
+        // yields a 2xx followed by a reset — but either way the body is withheld.
+        // `failure_mode: open` is an explicit operator opt-out of that guarantee:
+        // the pipeline logs this error and converts it to Continue, releasing the
+        // body even though items were lost. Only pre-commit failures (item insertion
+        // and earlier) reach this `?`: a post-commit message-cache refresh failure
+        // is tolerated inside `persist_items` because the cache is a self-healing
+        // projection (see `refresh_message_cache`).
+        self.append_items_blocking(&items.tenant_id, &conv_id, ctx, items.all_items)
+            .inspect_err(|e| warn!(error = %e, conversation_id = %conv_id, "conversation append-back failed"))?;
 
         Ok(FilterAction::Continue)
     }
@@ -590,6 +619,12 @@ async fn persist_items(
         .await
         .map_err(|e| -> FilterError { Box::new(e) })?;
 
+    // Item rows are durable past this point. The message cache is a self-healing
+    // projection rebuilt from the items table on the next successful sync, so a
+    // refresh failure is not data loss and must not fail the turn: propagating it
+    // would abort a request whose items already committed and drive a client retry
+    // that re-appends the same (typically id-less) input items as duplicates
+    // (#837). Log and continue; a later successful cache-refresh re-syncs the cache.
     refresh_message_cache(store, tenant_id, conversation_id).await;
     debug!(
         conversation_id,
@@ -600,6 +635,18 @@ async fn persist_items(
 }
 
 /// Refresh the denormalized conversation message cache after item mutation.
+///
+/// The cache is a projection of the items table that `openai_responses_rehydrate`
+/// replays on the next turn. It runs *after* the item rows have committed, so it
+/// is deliberately best-effort: the sync is idempotent — it rebuilds `messages`
+/// from the durable items table — so a later successful sync re-syncs whatever this
+/// attempt left behind (a later *append* is not sufficient: its own refresh may
+/// also fail). Failing the turn here would abort a request whose
+/// items already persisted and push the client into a retry that re-appends the
+/// same items as duplicates, so a refresh failure is logged and swallowed rather
+/// than propagated (#837). A transient stale-cache window remains until the next
+/// successful sync; closing it fully (atomic item-insert + projection) is tracked
+/// separately.
 async fn refresh_message_cache(store: &dyn ConversationItemStore, tenant_id: &str, conversation_id: &str) {
     let record = ConversationRecord {
         conversation_id: conversation_id.to_owned(),
@@ -609,7 +656,12 @@ async fn refresh_message_cache(store: &dyn ConversationItemStore, tenant_id: &st
         messages: Value::Null,
     };
     if let Err(e) = handlers::sync_conversation_messages(store, record).await {
-        warn!(error = %e, conversation_id, "conversation message sync failed after append-back");
+        warn!(
+            error = %e,
+            conversation_id,
+            "conversation message cache refresh failed after append-back; items are durable and the \
+             cache re-syncs on a later successful refresh (#837)"
+        );
     }
 }
 

apis/src/openai/conversations/tests.rs

@@ -1,7 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 // Copyright (c) 2026 Praxis Contributors
 
-use std::collections::BTreeMap;
+use std::{collections::BTreeMap, sync::Arc};
 
 use bytes::Bytes;
 use http::Method;
@@ -16,6 +16,7 @@ use super::{
 };
 use crate::{
     openai::responses::state::ResponsesState,
+    store::{ConversationItemRecord, ConversationItemStore, ConversationRecord, StoreError},
     test_utils::{make_filter_context, make_request, make_response},
 };
 
@@ -3425,6 +3426,89 @@ async fn on_response_body_appends_completed_response() {
     assert_eq!(items.len(), 2, "append-back should persist both input and output items");
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn on_response_body_surfaces_item_insert_failure() {
+    let filter = build_failing_filter(FailingItemStore {
+        fail_create_items: true,
+        fail_message_sync: false,
+    });
+
+    let req = make_request(Method::POST, "/v1/responses");
+    let mut ctx = make_filter_context(&req);
+    ctx.current_filter_id = Some(0);
+    set_append_back_metadata(&mut ctx);
+    ctx.extensions.insert(ResponsesState {
+        input: vec![serde_json::json!({"type": "message", "role": "user", "content": "hello from append"})],
+        ..ResponsesState::default()
+    });
+
+    let mut resp = make_response();
+    resp.headers
+        .insert(http::header::CONTENT_TYPE, "application/json".parse().unwrap());
+    ctx.response_header = Some(&mut resp);
+    drop(filter.on_response(&mut ctx).await.unwrap());
+
+    // A completed response whose items cannot be persisted must not be swallowed:
+    // the filter propagates the append-back failure as an error (#837). The
+    // client-visible outcome is then the pipeline's failure-mode decision — a
+    // withheld body under the default `failure_mode: closed`, or a logged release
+    // under `open`.
+    let response_json = serde_json::json!({
+        "status": "completed",
+        "output": [{"type": "message", "role": "assistant", "content": "hi from model"}]
+    });
+    let mut body = Some(Bytes::from(serde_json::to_vec(&response_json).unwrap()));
+    let result = filter.on_response_body(&mut ctx, &mut body, true);
+
+    assert!(
+        result.is_err(),
+        "item-insert failure during append-back must propagate as an error, not a silent success"
+    );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn on_response_body_tolerates_message_cache_failure() {
+    let filter = build_failing_filter(FailingItemStore {
+        fail_create_items: false,
+        fail_message_sync: true,
+    });
+
+    let req = make_request(Method::POST, "/v1/responses");
+    let mut ctx = make_filter_context(&req);
+    ctx.current_filter_id = Some(0);
+    set_append_back_metadata(&mut ctx);
+    ctx.extensions.insert(ResponsesState {
+        input: vec![serde_json::json!({"type": "message", "role": "user", "content": "hello from append"})],
+        ..ResponsesState::default()
+    });
+
+    let mut resp = make_response();
+    resp.headers
+        .insert(http::header::CONTENT_TYPE, "application/json".parse().unwrap());
+    ctx.response_header = Some(&mut resp);
+    drop(filter.on_response(&mut ctx).await.unwrap());
+
+    // Items committed, but the denormalized message-cache refresh failed. The
+    // cache is a self-healing projection of the durable items table, so this is
+    // not data loss: failing the turn would drive a client retry that re-appends
+    // the same items as duplicates. The turn must succeed (Continue); the cache
+    // re-syncs on a later successful cache-refresh/sync attempt — a later append is
+    // not sufficient, since its own refresh may also fail (#837, durability boundary).
+    let response_json = serde_json::json!({
+        "status": "completed",
+        "output": [{"type": "message", "role": "assistant", "content": "hi from model"}]
+    });
+    let mut body = Some(Bytes::from(serde_json::to_vec(&response_json).unwrap()));
+    let action = filter
+        .on_response_body(&mut ctx, &mut body, true)
+        .expect("message-cache refresh failure must not fail the turn");
+
+    assert!(
+        matches!(action, FilterAction::Continue),
+        "message-cache refresh failure must be tolerated (Continue), not surfaced as an error"
+    );
+}
+
 #[test]
 fn conformance_conversations_routes_match_runtime_registry() {
     let spec = generated_openapi_spec();
@@ -3530,6 +3614,132 @@ async fn create_test_conversation(filter: &dyn HttpFilter, metadata: Value) -> S
     resp["id"].as_str().unwrap().to_owned()
 }
 
+/// Build a conversations filter backed by a fault-injecting store.
+fn build_failing_filter(store: FailingItemStore) -> OpenaiConversationsFilter {
+    let yaml: serde_yaml::Value = serde_yaml::from_str(
+        r#"
+        backend: sqlite
+        database_url: "sqlite::memory:"
+        conversations_table: test_conversations
+        items_table: test_items
+        "#,
+    )
+    .unwrap();
+    let cfg: ConversationsConfig = parse_filter_config("openai_conversations", &yaml).unwrap();
+    OpenaiConversationsFilter::with_store_for_test(cfg, Arc::new(store))
+}
+
+/// A [`ConversationItemStore`] that fails selected operations on demand.
+///
+/// Benign methods return empty/default results; the two boolean knobs force the
+/// item-insert and message-cache-sync paths to error so append-back failure
+/// handling can be tested without a real database.
+struct FailingItemStore {
+    /// Force `create_conversation_items` to return a database error.
+    fail_create_items: bool,
+    /// Force the message-cache sync (`update_conversation_messages`) to error.
+    fail_message_sync: bool,
+}
+
+#[async_trait::async_trait]
+impl ConversationItemStore for FailingItemStore {
+    async fn upsert_conversation(&self, _record: &ConversationRecord) -> Result<(), StoreError> {
+        Ok(())
+    }
+
+    async fn update_conversation_messages(
+        &self,
+        _tenant_id: &str,
+        _conversation_id: &str,
+        _messages: &Value,
+    ) -> Result<bool, StoreError> {
+        if self.fail_message_sync {
+            return Err(StoreError::Database("mock message sync failure".to_owned()));
+        }
+        Ok(true)
+    }
+
+    async fn compare_and_swap_conversation_messages(
+        &self,
+        _tenant_id: &str,
+        _conversation_id: &str,
+        _expected_messages: &Value,
+        _messages: &Value,
+    ) -> Result<bool, StoreError> {
+        Ok(true)
+    }
+
+    async fn get_conversation(
+        &self,
+        _tenant_id: &str,
+        _conversation_id: &str,
+    ) -> Result<Option<ConversationRecord>, StoreError> {
+        Ok(None)
+    }
+
+    async fn delete_conversation(&self, _tenant_id: &str, _conversation_id: &str) -> Result<bool, StoreError> {
+        Ok(false)
+    }
+
+    async fn create_conversation_items(&self, _items: &[ConversationItemRecord]) -> Result<(), StoreError> {
+        if self.fail_create_items {
+            return Err(StoreError::Database("mock item insert failure".to_owned()));
+        }
+        Ok(())
+    }
+
+    async fn list_conversation_items(
+        &self,
+        _tenant_id: &str,
+        _conversation_id: &str,
+        _after_item_id: Option<&str>,
+        _limit: u32,
+        _ascending: bool,
+    ) -> Result<Vec<ConversationItemRecord>, StoreError> {
+        Ok(Vec::new())
+    }
+
+    async fn get_existing_conversation_item_ids(
+        &self,
+        _tenant_id: &str,
+        _conversation_id: &str,
+        _item_ids: &[&str],
+    ) -> Result<Vec<String>, StoreError> {
+        Ok(Vec::new())
+    }
+
+    async fn get_conversation_item(
+        &self,
+        _tenant_id: &str,
+        _conversation_id: &str,
+        _item_id: &str,
+    ) -> Result<Option<ConversationItemRecord>, StoreError> {
+        Ok(None)
+    }
+
+    async fn delete_conversation_item(
+        &self,
+        _tenant_id: &str,
+        _conversation_id: &str,
+        _item_id: &str,
+    ) -> Result<bool, StoreError> {
+        Ok(false)
+    }
+
+    async fn conversation_item_position(
+        &self,
+        _tenant_id: &str,
+        _conversation_id: &str,
+        _item_id: &str,
+    ) -> Result<Option<i64>, StoreError> {
+        Ok(None)
+    }
+
+    async fn max_item_position(&self, _tenant_id: &str, _conversation_id: &str) -> Result<i64, StoreError> {
+        Ok(0)
+    }
+}
+
 #[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
 struct OperationKey {
     method: String,