From 3d506c77ac67101abe672f14e1ab0641d58af067 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 23 Oct 2026 12:02:07 +0000 Subject: [PATCH 1/2] test(session): cover conditional stream append capability on stream port Add a fixture that hides the conditional stream append capability on the document port and a test asserting that append still succeeds when the stream port advertises it, so capability checks are verified to consult the stream port rather than the document port. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- .../src/threads/driver_append_fixtures.rs | 27 +++++++++++++++++-- .../src/threads/driver_append_tests.rs | 15 +++++++++++ 2 files changed, 40 insertions(+), 2 deletions(-) diff --git a/crates/tinyagents-session/src/threads/driver_append_fixtures.rs b/crates/tinyagents-session/src/threads/driver_append_fixtures.rs index 53d0465f6..7ed802323 100644 --- a/crates/tinyagents-session/src/threads/driver_append_fixtures.rs +++ b/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, fault: Arc, + 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(); diff --git a/crates/tinyagents-session/src/threads/driver_append_tests.rs b/crates/tinyagents-session/src/threads/driver_append_tests.rs index 2e9b916cf..738995fa8 100644 --- a/crates/tinyagents-session/src/threads/driver_append_tests.rs +++ b/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}; From 4da7a23fdddb8f52c9afbbca9c041c1b3f1b8163 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 23 Oct 2026 12:03:49 +0000 Subject: [PATCH 2/2] fix(session): check stream capabilities against the stream store The append preflight validated both ValidatedBatch and ConditionalStreamAppend against the document store, so a backend whose stream store supports conditional appends could be rejected. Each capability is now checked against the store that provides it, with a separate error message for each. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- .../src/threads/driver/append.rs | 32 ++++++++++++------- 1 file changed, 21 insertions(+), 11 deletions(-) diff --git a/crates/tinyagents-session/src/threads/driver/append.rs b/crates/tinyagents-session/src/threads/driver/append.rs index c23312571..e3545ed22 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", + ) + .into()); + } + if !self + .streams + .capabilities() + .contains(Capability::ConditionalStreamAppend) + { + return Err(StorageError::unsupported( + Capability::ConditionalStreamAppend, + "conversation append requires conditional streams", + ) + .into()); } Ok(()) }