From 0a52a13f1aab925aca41287dfd23581c40bf3019 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <enamakel@tinyhumans.ai>
Date: Fri, 23 Oct 2026 02:37:43 +0000
Subject: [PATCH] fix(session): check append capabilities on owning ports

Co-authored-by: Medulla <medulla@tinyhumans.ai>
---
 .../src/threads/driver/append.rs              | 32 +++++++++++------
 .../src/threads/driver_append_fixtures.rs     | 19 +++++++++-
 .../src/threads/driver_append_tests.rs        | 35 ++++++++++++++++++-
 3 files changed, 73 insertions(+), 13 deletions(-)

diff --git a/crates/tinyagents-session/src/threads/driver/append.rs b/crates/tinyagents-session/src/threads/driver/append.rs
index c23312571..33ea70430 100644
--- a/crates/tinyagents-session/src/threads/driver/append.rs
+++ b/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 and conditional streams",
+            )
+            .into());
+        }
+        if !self
+            .streams
+            .capabilities()
+            .contains(Capability::ConditionalStreamAppend)
+        {
+            return Err(StorageError::unsupported(
+                Capability::ConditionalStreamAppend,
+                "conversation append requires checked documents and conditional streams",
+            )
+            .into());
         }
         Ok(())
     }
diff --git a/crates/tinyagents-session/src/threads/driver_append_fixtures.rs b/crates/tinyagents-session/src/threads/driver_append_fixtures.rs
index 53d0465f6..63a1b88b9 100644
--- a/crates/tinyagents-session/src/threads/driver_append_fixtures.rs
+++ b/crates/tinyagents-session/src/threads/driver_append_fixtures.rs
@@ -90,11 +90,13 @@ impl Fault {
 struct Documents {
     inner: Arc<dyn DocumentStore>,
     fault: Arc<Fault>,
+    capabilities: Option<Capabilities>,
 }
 #[async_trait]
 impl DocumentStore for Documents {
     fn capabilities(&self) -> Capabilities {
-        self.inner.capabilities()
+        self.capabilities
+            .unwrap_or_else(|| self.inner.capabilities())
     }
     async fn ensure_collection(&self, _: &CollectionSpec) -> tinystoragedrivers_types::Result<()> {
         panic!("actor attempted DDL")
@@ -267,6 +269,7 @@ pub(super) fn faulted(
         docs: Arc::new(Documents {
             inner: store.docs.clone(),
             fault: fault.clone(),
+            capabilities: None,
         }),
         streams: Arc::new(Streams {
             inner: store.streams.clone(),
@@ -275,6 +278,20 @@ pub(super) fn faulted(
         validate_only: true,
     }
 }
+pub(super) fn without_document_capability(
+    store: &DriverConversationStore,
+    capability: tinystoragedrivers_types::Capability,
+) -> DriverConversationStore {
+    DriverConversationStore {
+        docs: Arc::new(Documents {
+            inner: store.docs.clone(),
+            fault: Fault::new(Phase::Probe, Action::LoseAck),
+            capabilities: Some(store.docs.capabilities().without(capability)),
+        }),
+        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();
diff --git a/crates/tinyagents-session/src/threads/driver_append_tests.rs b/crates/tinyagents-session/src/threads/driver_append_tests.rs
index 2e9b916cf..b64bbcaa8 100644
--- a/crates/tinyagents-session/src/threads/driver_append_tests.rs
+++ b/crates/tinyagents-session/src/threads/driver_append_tests.rs
@@ -131,7 +131,40 @@ async fn reconstructed_memory_prepared_append_reopens_and_preserves_classic_dupl
 
 #[path = "driver_append_fixtures.rs"]
 mod fixtures;
-use fixtures::{Action, Fault, Phase, actor, faulted};
+use fixtures::{Action, Fault, Phase, actor, faulted, without_document_capability};
+
+#[tokio::test]
+async fn append_requires_conditional_stream_support_from_the_stream_port() {
+    let store = actor(&MemoryStorage::new()).await;
+    let store = without_document_capability(
+        &store,
+        tinystoragedrivers_types::Capability::ConditionalStreamAppend,
+    );
+    assert!(
+        store
+            .docs
+            .capabilities()
+            .contains(tinystoragedrivers_types::Capability::ValidatedBatch)
+    );
+    assert!(
+        !store
+            .docs
+            .capabilities()
+            .contains(tinystoragedrivers_types::Capability::ConditionalStreamAppend)
+    );
+    assert!(
+        store
+            .streams
+            .capabilities()
+            .contains(tinystoragedrivers_types::Capability::ConditionalStreamAppend)
+    );
+
+    let message = message("agent:stream-capability", "append succeeds");
+    assert_eq!(
+        store.append_message("same", message.clone()).await.unwrap(),
+        message
+    );
+}
 
 async fn canceled(backend: &dyn StorageBackend, phase: Phase) {
     let store = actor(backend).await;