From 8c06369521a89a33a19bd12fd2871261398076b0 Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Sat, 24 Oct 2026 04:34:03 +0530 Subject: [PATCH 1/2] fix(session): admit a conversation append on the stream port's capability `admit_append` required ValidatedBatch and ConditionalStreamAppend from the document port. The append writes messages through the stream port, and a backend may advertise the two on separate ports: SQLite's document store lists ValidatedBatch but not ConditionalStreamAppend, which only its stream store does. So every append on a SQLite-backed conversation store was refused with "unsupported capability conditional_stream_append" before writing anything, while memory-backed tests (whose documents list everything) passed. Check each capability on the port that uses it: ValidatedBatch on the documents, ConditionalStreamAppend on the streams. Tests cover both shapes with the append fixtures: a document port without the stream capability plus a stream port with it appends, and a stream port without it is still refused. --- .../src/threads/driver/append.rs | 22 +++++++++--- .../src/threads/driver_append_fixtures.rs | 19 ++++++++-- .../src/threads/driver_append_tests.rs | 35 +++++++++++++++++++ 3 files changed, 70 insertions(+), 6 deletions(-) diff --git a/crates/tinyagents-session/src/threads/driver/append.rs b/crates/tinyagents-session/src/threads/driver/append.rs index c23312571..5841e413e 100644 --- a/crates/tinyagents-session/src/threads/driver/append.rs +++ b/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", diff --git a/crates/tinyagents-session/src/threads/driver_append_fixtures.rs b/crates/tinyagents-session/src/threads/driver_append_fixtures.rs index 53d0465f6..ed12f49e0 100644 --- a/crates/tinyagents-session/src/threads/driver_append_fixtures.rs +++ b/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 { diff --git a/crates/tinyagents-session/src/threads/driver_append_tests.rs b/crates/tinyagents-session/src/threads/driver_append_tests.rs index 2e9b916cf..93b759ad9 100644 --- a/crates/tinyagents-session/src/threads/driver_append_tests.rs +++ b/crates/tinyagents-session/src/threads/driver_append_tests.rs @@ -247,6 +247,41 @@ 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; From 751a2d786c576073e1e946c75f60ab2cdb5aca02 Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Sat, 24 Oct 2026 04:43:02 +0530 Subject: [PATCH 2/2] style(session): rustfmt the split-capability append tests --- .../src/threads/driver_append_tests.rs | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/crates/tinyagents-session/src/threads/driver_append_tests.rs b/crates/tinyagents-session/src/threads/driver_append_tests.rs index 93b759ad9..dc5f4c002 100644 --- a/crates/tinyagents-session/src/threads/driver_append_tests.rs +++ b/crates/tinyagents-session/src/threads/driver_append_tests.rs @@ -252,7 +252,10 @@ async fn unobserved_ordinary_reservation_returns_typed_uncertainty_with_original #[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)); + let split = faulted( + &store, + Fault::new(Phase::Probe, Action::DocsLackStreamAppend), + ); assert!( !split .docs @@ -271,7 +274,10 @@ async fn an_append_admits_a_stream_capability_its_document_port_does_not_list() #[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 split = faulted( + &store, + Fault::new(Phase::Probe, Action::StreamsLackStreamAppend), + ); let error = split .append_message("same", message("agent:a", "refused")) .await