fix(session): admit a conversation append on the stream port's capability

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

@@ -35,11 +35,25 @@ impl DriverConversationStore {
         for spec in Self::append_collection_specs() {
             self.docs.validate_collection_spec(&spec).await?;
         }
-        for capability in [
-            Capability::ValidatedBatch,
-            Capability::ConditionalStreamAppend,
+        // Each capability is checked on the port that uses it: the intent and
+        // metadata commit is a checked document batch, the message write a
+        // conditional stream append. A backend may advertise them on separate
+        // ports (SQLite does), so neither port speaks for the other.
+        for (capability, available) in [
+            (
+                Capability::ValidatedBatch,
+                self.docs
+                    .capabilities()
+                    .contains(Capability::ValidatedBatch),
+            ),
+            (
+                Capability::ConditionalStreamAppend,
+                self.streams
+                    .capabilities()
+                    .contains(Capability::ConditionalStreamAppend),
+            ),
         ] {
-            if !self.docs.capabilities().contains(capability) {
+            if !available {
                 return Err(StorageError::unsupported(
                     capability,
                     "conversation append requires checked documents and conditional streams",

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

@@ -22,6 +22,11 @@ pub(super) enum Action {
     FailBefore,
     Unsupported,
     Barrier,
+    /// The document port does not advertise conditional stream appends, as
+    /// SQLite documents do not (only its stream port does).
+    DocsLackStreamAppend,
+    /// The stream port does not advertise conditional stream appends.
+    StreamsLackStreamAppend,
 }
 #[derive(Debug)]
 pub(super) struct Fault {
@@ -94,7 +99,12 @@ struct Documents {
 #[async_trait]
 impl DocumentStore for Documents {
     fn capabilities(&self) -> Capabilities {
-        self.inner.capabilities()
+        let capabilities = self.inner.capabilities();
+        if matches!(self.fault.action, Action::DocsLackStreamAppend) {
+            capabilities.without(tinystoragedrivers_types::Capability::ConditionalStreamAppend)
+        } else {
+            capabilities
+        }
     }
     async fn ensure_collection(&self, _: &CollectionSpec) -> tinystoragedrivers_types::Result<()> {
         panic!("actor attempted DDL")
@@ -198,7 +208,12 @@ struct Streams {
 #[async_trait]
 impl StreamStore for Streams {
     fn capabilities(&self) -> tinystoragedrivers_types::Capabilities {
-        self.inner.capabilities()
+        let capabilities = self.inner.capabilities();
+        if matches!(self.fault.action, Action::StreamsLackStreamAppend) {
+            capabilities.without(tinystoragedrivers_types::Capability::ConditionalStreamAppend)
+        } else {
+            capabilities
+        }
     }
 
     async fn append(&self, s: &str, v: Value) -> tinystoragedrivers_types::Result<u64> {

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

@@ -247,6 +247,47 @@ async fn unobserved_ordinary_reservation_returns_typed_uncertainty_with_original
         0
     );
 }
+/// SQLite's shape: its document port does not advertise conditional stream
+/// appends, only its stream port does. Append must check the stream port.
+#[tokio::test]
+async fn an_append_admits_a_stream_capability_its_document_port_does_not_list() {
+    let store = actor(&MemoryStorage::new()).await;
+    let split = faulted(
+        &store,
+        Fault::new(Phase::Probe, Action::DocsLackStreamAppend),
+    );
+    assert!(
+        !split
+            .docs
+            .capabilities()
+            .contains(tinystoragedrivers_types::Capability::ConditionalStreamAppend)
+    );
+    split
+        .append_message("same", message("agent:a", "kept"))
+        .await
+        .expect("the stream port carries the capability the append uses");
+    let messages = store.get_messages("same").await.unwrap();
+    assert_eq!(messages.len(), 1);
+    assert_eq!(messages[0].content, "kept");
+}
+
+#[tokio::test]
+async fn an_append_is_refused_when_the_stream_port_lacks_conditional_appends() {
+    let store = actor(&MemoryStorage::new()).await;
+    let split = faulted(
+        &store,
+        Fault::new(Phase::Probe, Action::StreamsLackStreamAppend),
+    );
+    let error = split
+        .append_message("same", message("agent:a", "refused"))
+        .await
+        .unwrap_err();
+    assert!(
+        matches!(error, ConversationError::Driver(ref e) if matches!(e.kind(), ErrorKind::Unsupported(tinystoragedrivers_types::Capability::ConditionalStreamAppend)))
+    );
+    assert!(store.get_messages("same").await.unwrap().is_empty());
+}
+
 #[tokio::test]
 async fn counterfeit_document_capabilities_do_not_reserve_when_actual_stream_probe_refuses() {
     let store = actor(&MemoryStorage::new()).await;