fix(session): check conditional append capability on streams

crates/tinyagents-session/src/threads/driver/append.rs

@@ -35,17 +35,27 @@ impl DriverConversationStore {
         for spec in Self::append_collection_specs() {
             self.docs.validate_collection_spec(&spec).await?;
         }
-        for capability in [
-            Capability::ValidatedBatch,
-            Capability::ConditionalStreamAppend,
-        ] {
-            if !self.docs.capabilities().contains(capability) {
-                return Err(StorageError::unsupported(
-                    capability,
-                    "conversation append requires checked documents and conditional streams",
-                )
-                .into());
-            }
+        if !self
+            .docs
+            .capabilities()
+            .contains(Capability::ValidatedBatch)
+        {
+            return Err(StorageError::unsupported(
+                Capability::ValidatedBatch,
+                "conversation append requires checked documents",
+            )
+            .into());
+        }
+        if !self
+            .streams
+            .capabilities()
+            .contains(Capability::ConditionalStreamAppend)
+        {
+            return Err(StorageError::unsupported(
+                Capability::ConditionalStreamAppend,
+                "conversation append requires conditional streams",
+            )
+            .into());
         }
         Ok(())
     }

crates/tinyagents-session/src/threads/driver_append_fixtures.rs

@@ -1,7 +1,8 @@
 use super::*;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use tinyagents_storage_kit::{
-    Capabilities, Page, Query, ReadCheck, Sort, StreamEntry, Version, WriteOp, WriteResult,
+    Capabilities, Capability, Page, Query, ReadCheck, Sort, StreamEntry, Version, WriteOp,
+    WriteResult,
 };
 use tinystoragedrivers_api::StorageBackend;
 use tokio::sync::{Barrier, Notify};
@@ -90,11 +91,17 @@ impl Fault {
 struct Documents {
     inner: Arc<dyn DocumentStore>,
     fault: Arc<Fault>,
+    hide_conditional_stream_append: bool,
 }
 #[async_trait]
 impl DocumentStore for Documents {
     fn capabilities(&self) -> Capabilities {
-        self.inner.capabilities()
+        let capabilities = self.inner.capabilities();
+        if self.hide_conditional_stream_append {
+            capabilities.without(Capability::ConditionalStreamAppend)
+        } else {
+            capabilities
+        }
     }
     async fn ensure_collection(&self, _: &CollectionSpec) -> tinystoragedrivers_types::Result<()> {
         panic!("actor attempted DDL")
@@ -267,6 +274,7 @@ pub(super) fn faulted(
         docs: Arc::new(Documents {
             inner: store.docs.clone(),
             fault: fault.clone(),
+            hide_conditional_stream_append: false,
         }),
         streams: Arc::new(Streams {
             inner: store.streams.clone(),
@@ -275,6 +283,21 @@ pub(super) fn faulted(
         validate_only: true,
     }
 }
+
+pub(super) fn hide_document_conditional_stream_append(
+    store: &DriverConversationStore,
+) -> DriverConversationStore {
+    DriverConversationStore {
+        docs: Arc::new(Documents {
+            inner: store.docs.clone(),
+            fault: Fault::new(Phase::Probe, Action::Unsupported),
+            hide_conditional_stream_append: true,
+        }),
+        streams: store.streams.clone(),
+        validate_only: true,
+    }
+}
+
 pub(super) async fn actor(backend: &dyn StorageBackend) -> DriverConversationStore {
     let scope = Scope::new("alice").unwrap();
     let raw = backend.for_scope(&scope).unwrap();

crates/tinyagents-session/src/threads/driver_append_tests.rs

@@ -129,6 +129,21 @@ async fn reconstructed_memory_prepared_append_reopens_and_preserves_classic_dupl
     prepared_append(&backend).await;
 }
 
+#[tokio::test]
+async fn append_checks_conditional_stream_capability_on_the_stream_port() {
+    let store = actor(&MemoryStorage::new()).await;
+    let stream_capability_only = fixtures::hide_document_conditional_stream_append(&store);
+    let message = message("agent:capability", "conditional stream append is supported");
+
+    assert_eq!(
+        stream_capability_only
+            .append_message("same", message.clone())
+            .await
+            .unwrap(),
+        message
+    );
+}
+
 #[path = "driver_append_fixtures.rs"]
 mod fixtures;
 use fixtures::{Action, Fault, Phase, actor, faulted};