use super::*;
use anda_engine::memory::ConversationRef;
use futures::stream::BoxStream;
use object_store::{
CopyOptions, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta,
PutMultipartOptions, PutOptions, PutPayload, PutResult, Result as StoreResult, path::Path,
};
use std::sync::atomic::AtomicBool;
#[derive(Debug, Default)]
struct BlockAfterPut {
inner: InMemory,
path: parking_lot::Mutex<String>,
armed: AtomicBool,
only_when_readonly: parking_lot::Mutex<Option<Weak<AndaDB>>>,
entered: tokio::sync::Notify,
}
impl std::fmt::Display for BlockAfterPut {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str("BlockAfterPut")
}
}
impl BlockAfterPut {
fn arm(&self, path: &str) {
*self.path.lock() = path.into();
self.armed.store(true, Ordering::SeqCst);
}
}
#[async_trait::async_trait]
impl ObjectStore for BlockAfterPut {
async fn put_opts(
&self,
path: &Path,
payload: PutPayload,
options: PutOptions,
) -> StoreResult<PutResult> {
let phase_matches = self
.only_when_readonly
.lock()
.as_ref()
.is_none_or(|db| db.upgrade().is_some_and(|db| db.is_read_only()));
let block = phase_matches
&& path.as_ref().contains(self.path.lock().as_str())
&& self.armed.swap(false, Ordering::SeqCst);
let result = self.inner.put_opts(path, payload, options).await?;
if block {
self.entered.notify_one();
std::future::pending::<()>().await;
}
Ok(result)
}
async fn put_multipart_opts(
&self,
path: &Path,
options: PutMultipartOptions,
) -> StoreResult<Box<dyn MultipartUpload>> {
self.inner.put_multipart_opts(path, options).await
}
async fn get_opts(&self, path: &Path, options: GetOptions) -> StoreResult<GetResult> {
self.inner.get_opts(path, options).await
}
fn delete_stream(
&self,
paths: BoxStream<'static, StoreResult<Path>>,
) -> BoxStream<'static, StoreResult<Path>> {
self.inner.delete_stream(paths)
}
fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, StoreResult<ObjectMeta>> {
self.inner.list(prefix)
}
async fn list_with_delimiter(&self, prefix: Option<&Path>) -> StoreResult<ListResult> {
self.inner.list_with_delimiter(prefix).await
}
async fn copy_opts(&self, from: &Path, to: &Path, options: CopyOptions) -> StoreResult<()> {
self.inner.copy_opts(from, to, options).await
}
}
async fn fixture(name: &str) -> (AppState, Arc<Space>, Arc<BlockAfterPut>) {
let store = Arc::new(BlockAfterPut::default());
let app = crate::testkit::app_state_core(name, Arc::new(Models::default()), vec![], "test", 0)
.fork_with_store(store.clone());
let space = crate::testkit::create_loaded_space(&app, name).await;
(app, space, store)
}
async fn record(collection: &Collection, status: ConversationStatus, label: &str) -> u64 {
collection
.add_from(&Conversation {
user: SELF_USER_ID,
status,
label: Some(label.into()),
..Default::default()
})
.await
.unwrap()
}
#[tokio::test]
async fn close_recovers_a_cancelled_conversation_put_and_persists_unknown_terminal_state() {
let (app, space, store) = fixture("close_conversation").await;
let id = record(
&space.recall.conversations_collection,
ConversationStatus::Working,
"recall",
)
.await;
space
.recall
.conversations_collection
.update(
id,
BTreeMap::from([(
"failed_reason".into(),
Fv::Text("prior provider failure".into()),
)]),
)
.await
.unwrap();
let completed = record(
&space.recall.conversations_collection,
ConversationStatus::Completed,
"complete",
)
.await;
let queued = record(
&space.conversations,
ConversationStatus::Submitted,
"formation",
)
.await;
space.flush().await.unwrap();
let original = space.recall.conversations_collection.clone();
let writer = original.clone();
store.arm("close_conversation/recall/data/");
space.tasks.spawn(async move {
let _ = writer
.update(
id,
BTreeMap::from([("label".into(), Fv::Text("durable-before-cancel".into()))]),
)
.await;
});
tokio::time::timeout(Duration::from_secs(10), store.entered.notified())
.await
.unwrap();
tokio::time::timeout(Duration::from_secs(10), space.close())
.await
.unwrap()
.unwrap();
assert!(
original.is_poisoned(),
"the fault must cancel a real mutating future"
);
space.close().await.unwrap();
let reopened = app
.fork_with_store(store)
.load_space_with("close_conversation", false, false)
.await
.unwrap();
let restored = reopened
.recall
.conversations
.get_conversation(id)
.await
.unwrap();
assert_eq!(restored.label.as_deref(), Some("durable-before-cancel"));
assert_eq!(restored.status, ConversationStatus::Cancelled);
let reason = restored.failed_reason.unwrap();
assert!(reason.contains("outcome_unknown"));
assert!(reason.contains("prior provider failure"));
assert_eq!(
reopened
.recall
.conversations
.get_conversation(completed)
.await
.unwrap()
.status,
ConversationStatus::Completed
);
assert_eq!(
reopened
.memory
.get_conversation(queued)
.await
.unwrap()
.status,
ConversationStatus::Submitted
);
reopened.close().await.unwrap();
}
#[tokio::test]
async fn close_recovers_a_durable_native_commit_without_replaying_the_formation() {
let (app, space, store) = fixture("close_native_commit").await;
let id = space
.memory
.add_conversation(ConversationRef::from(&Conversation {
user: SELF_USER_ID,
status: ConversationStatus::Working,
label: Some("formation".into()),
..Default::default()
}))
.await
.unwrap();
space.flush().await.unwrap();
let original = space.memory.nexus().store.commit_log();
let nexus = space.memory.nexus().clone();
store.arm("close_native_commit/kip_commit_log/data/");
space.tasks.spawn(async move {
let _ = execute_request(
nexus.as_ref(),
&kip::request(
r#"MUTATE {
CREATE CONCEPT ?p {TYPE "Person" NAME "effect committed before close"}
}"#,
),
)
.await;
});
tokio::time::timeout(Duration::from_secs(10), store.entered.notified())
.await
.unwrap();
tokio::time::timeout(Duration::from_secs(10), space.close())
.await
.unwrap()
.unwrap();
assert!(original.is_poisoned());
let reopened = app
.fork_with_store(store)
.load_space_with("close_native_commit", false, false)
.await
.unwrap();
let response = reopened
.execute_kip_readonly(kip::request(
r#"FIND(?p) WHERE {
?p CONCEPT {name: "effect committed before close"}
}"#,
))
.await
.unwrap();
assert_eq!(
crate::assess::single_read_result(&response)
.unwrap()
.as_array()
.unwrap()
.len(),
1
);
assert_eq!(
reopened.memory.get_conversation(id).await.unwrap().status,
ConversationStatus::Cancelled
);
assert!(
reopened.restart_formation(SELF_USER_ID, id).await.is_err(),
"unknown work must not automatically rerun"
);
reopened.close().await.unwrap();
}
#[derive(Debug)]
struct BlockedModel;
impl anda_engine::model::CompletionFeaturesDyn for BlockedModel {
fn model_name(&self) -> String {
"blocked-close".into()
}
fn completion(
&self,
_: anda_core::CompletionRequest,
) -> anda_core::BoxPinFut<Result<AgentOutput, BoxError>> {
Box::pin(std::future::pending())
}
}
#[tokio::test]
async fn close_can_retry_after_its_own_checkpoint_put_was_cancelled() {
let (app, space, store) = fixture("close_retry").await;
let id = record(
&space.recall.conversations_collection,
ConversationStatus::Working,
"recall",
)
.await;
space.flush().await.unwrap();
*store.only_when_readonly.lock() = Some(Arc::downgrade(&space.db));
store.arm("close_retry/recall/meta.cbor");
let closing = space.clone();
let task = tokio::spawn(async move { closing.close().await });
tokio::time::timeout(Duration::from_secs(10), store.entered.notified())
.await
.unwrap();
assert!(
space.db.is_read_only(),
"interrupt the storage-close phase after reconciliation"
);
task.abort();
assert!(task.await.unwrap_err().is_cancelled());
assert!(space.recall.conversations_collection.is_poisoned());
tokio::time::timeout(Duration::from_secs(10), space.close())
.await
.unwrap()
.unwrap();
space.close().await.unwrap();
let reopened = app
.fork_with_store(store)
.load_space_with("close_retry", false, false)
.await
.unwrap();
assert_eq!(
reopened
.recall
.conversations
.get_conversation(id)
.await
.unwrap()
.status,
ConversationStatus::Cancelled
);
reopened.close().await.unwrap();
}
#[tokio::test]
async fn close_captures_an_active_submitted_formation_before_the_guard_is_dropped() {
let app = crate::testkit::app_state_core(
"close_submitted",
crate::testkit::models_with_completer(BlockedModel),
vec![],
"test",
0,
)
.fork_with_store(Arc::new(InMemory::new()));
let space = crate::testkit::create_loaded_space(&app, "close_submitted").await;
let output = space
.ingest(SELF_USER_ID, StringOr::String("remember this".into()))
.await
.unwrap();
let id = output.conversation.unwrap();
assert_eq!(space.formation.processing_id(), id);
space.close().await.unwrap();
let reopened = app
.fork_with_store(app.object_store.clone())
.load_space_with("close_submitted", false, false)
.await
.unwrap();
let conversation = reopened.memory.get_conversation(id).await.unwrap();
assert_eq!(conversation.status, ConversationStatus::Cancelled);
assert!(
conversation
.failed_reason
.unwrap()
.contains("outcome_unknown")
);
assert!(reopened.restart_formation(SELF_USER_ID, id).await.is_err());
reopened.close().await.unwrap();
}