diff --git a/crates/astra-cli/src/cli/session/session_startup.rs b/crates/astra-cli/src/cli/session/session_startup.rs index 76869bc3c3..d7fc581d78 100644 --- a/crates/astra-cli/src/cli/session/session_startup.rs +++ b/crates/astra-cli/src/cli/session/session_startup.rs @@ -488,11 +488,7 @@ impl astra_runtime::turn::cloud::memoria_compact::MemoriaPort for CliSessionMemo if !status.is_success() { return Err(format!("memory store HTTP {status}")); } - let memory_id = payload - .get("memory_id") - .and_then(serde_json::Value::as_str) - .ok_or_else(|| "memory store missing memory_id".to_string())? - .to_string(); + let memory_id = astra_runtime::turn::cloud::memoria_compact::confirmed_memory_id(&payload)?; if memory_type == "working" && let Some(session_id) = session_id && let Ok(mut guard) = self.working_ids.lock() @@ -502,6 +498,66 @@ impl astra_runtime::turn::cloud::memoria_compact::MemoriaPort for CliSessionMemo Ok(memory_id) } + async fn store_episode(&self, session_id: &str, overview: &str) -> Result { + astra_runtime::turn::cloud::memoria_compact::MemoryScope::new( + &self.owner_user_id, + session_id, + )?; + if overview.trim().is_empty() { + return Err("store_episode: empty overview".into()); + } + let content = astra_runtime::turn::cloud::memoria_compact::encode_episode_memory( + session_id, overview, + ); + self.store(&content, "episodic", Some(session_id), Some("T3")) + .await + } + + async fn store_scene( + &self, + session_id: &str, + signal: &str, + summary: &str, + ) -> Result { + astra_runtime::turn::cloud::memoria_compact::MemoryScope::new( + &self.owner_user_id, + session_id, + )?; + if summary.trim().is_empty() { + return Err("store_scene: empty summary".into()); + } + let content = + astra_runtime::turn::cloud::memoria_compact::encode_scene_memory(signal, summary); + self.store(&content, "semantic", Some(session_id), Some("T4")) + .await + } + + async fn reflect_session( + &self, + session_id: &str, + force: bool, + ) -> Result { + astra_runtime::turn::cloud::memoria_compact::MemoryScope::new( + &self.owner_user_id, + session_id, + )?; + let token = self.fresh_token().await?; + let text = self + .api + .post_bearer_path_json_text( + &token, + "/memory/reflect", + &serde_json::json!({ + "session_id": session_id, "force": force, "mode": "candidates" + }), + ) + .await + .map_err(|error| format!("memory reflect failed: {error}"))?; + let response = serde_json::from_str(&text) + .map_err(|error| format!("memory reflect parse failed: {error}"))?; + Ok(astra_runtime::turn::cloud::memoria_compact::parse_reflect_summary(&response)) + } + async fn purge_working(&self, session_id: &str) -> Result { let Some(memory_id) = self .working_ids @@ -1486,6 +1542,149 @@ mod tests { ); } + #[serial_test::serial] + #[tokio::test] + async fn cli_exit_archives_before_cleanup_and_retains_unconfirmed_working() { + use std::sync::{Arc, Mutex}; + use wiremock::matchers::body_partial_json; + let _creds_guard = crate::tests::isolate_credentials(); + let (_sessions, _sessions_guard) = crate::tests::isolated_sessions_dir(); + write_profile_with_token("session-1"); + for receipt in [ + r#"{"memory_id":"episode-1"}"#, + "{}", + r#"{"memory_id":""}"#, + "not-json", + ] { + let server = MockServer::start().await; + let events = Arc::new(Mutex::new(Vec::<&str>::new())); + let sid = "cli-exit-archive"; + let snapshot = astra_runtime::session_memory::runner::encode_session_memory_entry( + sid, + "# Session Memory\n\n## Task Specification\nComplete durable archive\n\n## Current State\n- Changes verified", + ); + Mock::given(method("POST")) + .and(path("/memory/retrieve")) + .and(header_exists("authorization")) + .and(body_partial_json( + serde_json::json!({"session_id": sid, "session_scope": "only"}), + )) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "memories": [{"memory_id": "working-1", "user_id": "user-1", "session_id": sid, + "memory_type": "working", "content": snapshot}] + }))) + .expect(1) + .mount(&server) + .await; + let stored = Arc::clone(&events); + let response_body = receipt.to_string(); + Mock::given(method("POST")).and(path("/memory/store")) + .and(header_exists("authorization")) + .and(body_partial_json(serde_json::json!({"memory_type": "episodic", "trust_tier": "T3", "session_id": sid}))) + .respond_with(move |_: &wiremock::Request| { + stored.lock().unwrap().push("episode"); + ResponseTemplate::new(201).set_body_string(response_body.clone()) + }).expect(1).mount(&server).await; + let purged = Arc::clone(&events); + Mock::given(method("DELETE")) + .and(path("/memory/working-1")) + .and(header_exists("authorization")) + .respond_with(move |_: &wiremock::Request| { + purged.lock().unwrap().push("purge"); + ResponseTemplate::new(204) + }) + .expect(u64::from(receipt.contains("episode-1"))) + .mount(&server) + .await; + let reflected = Arc::clone(&events); + Mock::given(method("POST")) + .and(path("/memory/reflect")) + .and(header_exists("authorization")) + .and(body_partial_json( + serde_json::json!({"mode": "candidates", "session_id": sid, "force": false}), + )) + .respond_with(move |_: &wiremock::Request| { + reflected.lock().unwrap().push("reflect"); + ResponseTemplate::new(503).set_body_string("optional reflection unavailable") + }) + .expect(1) + .mount(&server) + .await; + let api = astra_thin_client::ThinClient::new(&server.uri(), None).unwrap(); + let mut state = SessionState { + session_id: Some(sid.into()), + turn: 3, + session_memory_port: Some(Arc::new(CliSessionMemoryMemoriaPort::new( + api, None, "user-1", + ))), + ..Default::default() + }; + crate::cli::session::session_cleanup::finalize_session_with_budget( + &mut state, + std::time::Duration::from_secs(2), + ) + .await + .unwrap(); + let expected = if receipt.contains("episode-1") { + vec!["episode", "purge", "reflect"] + } else { + vec!["episode", "reflect"] + }; + assert_eq!(*events.lock().unwrap(), expected, "receipt={receipt}"); + server.verify().await; + } + } + + #[serial_test::serial] + #[tokio::test] + async fn cli_session_memory_port_stores_typed_scenes_and_accepts_reflect_skips() { + use wiremock::matchers::body_partial_json; + let _creds_guard = crate::tests::isolate_credentials(); + write_profile_with_token("session-1"); + let server = MockServer::start().await; + let content = astra_runtime::turn::cloud::memoria_compact::encode_scene_memory( + "pattern", + "recurring evidence", + ); + Mock::given(method("POST")) + .and(path("/memory/store")) + .and(header_exists("authorization")) + .and(body_partial_json( + serde_json::json!({"content": content, "memory_type": "semantic", + "session_id": "session-1", "trust_tier": "T4"}), + )) + .respond_with( + ResponseTemplate::new(201) + .set_body_json(serde_json::json!({"memory_id": "scene-1"})), + ) + .expect(1) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/memory/reflect")) + .and(header_exists("authorization")) + .respond_with( + ResponseTemplate::new(200).set_body_json( + serde_json::json!({"skipped": true, "cooldown_remaining_s": 3600}), + ), + ) + .expect(1) + .mount(&server) + .await; + let api = astra_thin_client::ThinClient::new(&server.uri(), None).unwrap(); + let port = CliSessionMemoryMemoriaPort::new(api, None, "user-1"); + assert_eq!( + port.store_scene("session-1", "pattern", "recurring evidence") + .await + .unwrap(), + "scene-1" + ); + let summary = port.reflect_session("session-1", false).await.unwrap(); + assert_eq!(summary.candidates, 0); + assert!(summary.candidate_payloads.is_empty()); + server.verify().await; + } + #[serial_test::serial] #[tokio::test] async fn cli_session_memory_port_updates_working_snapshot_through_server_contract() { diff --git a/crates/astra-memoria/src/lib.rs b/crates/astra-memoria/src/lib.rs index 6efdc7ab31..761788d341 100644 --- a/crates/astra-memoria/src/lib.rs +++ b/crates/astra-memoria/src/lib.rs @@ -278,6 +278,25 @@ impl From<&str> for MemoriaOperationError { } } +/// A successful write must identify the durable memory it created or updated. +/// HTTP success alone does not authorize cleanup or a claim of storage. +pub fn confirmed_memory_id(response: &serde_json::Value) -> Result { + let id = response + .get("memory_id") + .and_then(serde_json::Value::as_str) + .ok_or_else(|| "memory backend did not return a valid memory_id receipt".to_string())?; + validate_memory_id(id)?; + Ok(id.to_string()) +} + +/// Validate a receipt returned by any transport, including non-HTTP ports. +pub fn validate_memory_id(id: &str) -> Result<(), String> { + if id.trim().is_empty() || id.trim() != id { + return Err("memory backend did not return a valid memory_id receipt".into()); + } + Ok(()) +} + /// Provider-neutral Memoria operations used by runtime orchestration. /// Reads resolve fresh authority within the actual operation and report disabled /// separately from successful empty results. Defaults only delegate; callers do @@ -394,7 +413,7 @@ pub trait MemoriaPort: Send + Sync { } async fn store_episode(&self, _session_id: &str, _overview: &str) -> Result { - Ok(String::new()) + Err("Memoria transport does not support episode persistence".into()) } async fn store_scene( @@ -403,7 +422,7 @@ pub trait MemoriaPort: Send + Sync { _signal: &str, _summary: &str, ) -> Result { - Ok(String::new()) + Err("Memoria transport does not support scene persistence".into()) } async fn reflect_session( @@ -475,6 +494,25 @@ pub fn parse_reflect_candidates(data: &Value) -> Vec { .collect() } +/// Decode the same reflection response for direct and Server-proxied transports. +/// A successful cooldown skip is an empty best-effort result, not a failure. +pub fn parse_reflect_summary(data: &Value) -> ReflectSummary { + let candidate_payloads = parse_reflect_candidates(data); + ReflectSummary { + synthesized: data + .get("synthesized") + .and_then(Value::as_bool) + .unwrap_or(false), + candidates: (candidate_payloads.len() as u64).max( + data.get("scenes_created") + .and_then(Value::as_u64) + .unwrap_or(0), + ), + candidate_payloads, + diagnostics: data.to_string().chars().take(200).collect(), + } +} + #[derive(Debug, Clone, Default, Serialize, Deserialize)] pub struct MemoriaMemory { pub memory_id: String, @@ -960,6 +998,79 @@ mod tests { use super::*; use serde_json::json; + struct UnsupportedPersistence; + #[async_trait::async_trait] + impl MemoriaPort for UnsupportedPersistence { + async fn retrieve_ext( + &self, + _: &str, + _: Option<&str>, + _: usize, + _: bool, + ) -> Result, MemoriaOperationError> { + Ok(Vec::new()) + } + async fn store( + &self, + _: &str, + _: &str, + _: Option<&str>, + _: Option<&str>, + ) -> Result { + unreachable!("unsupported persistence must not fabricate a write") + } + async fn purge_working(&self, _: &str) -> Result { + Ok(0) + } + } + + #[test] + fn unsupported_episode_and_scene_persistence_fail_explicitly() { + use std::task::{Context, Poll, Waker}; + let mut context = Context::from_waker(Waker::noop()); + assert!(matches!( + UnsupportedPersistence + .store_episode("session", "overview") + .as_mut() + .poll(&mut context), + Poll::Ready(Err(_)) + )); + assert!(matches!( + UnsupportedPersistence + .store_scene("session", "signal", "summary") + .as_mut() + .poll(&mut context), + Poll::Ready(Err(_)) + )); + } + + #[test] + fn reflect_cooldown_skip_is_a_successful_empty_optional_result() { + let summary = + parse_reflect_summary(&json!({"skipped": true, "cooldown_remaining_s": 3600})); + assert_eq!(summary.candidates, 0); + assert!(!summary.synthesized); + assert!(summary.candidate_payloads.is_empty()); + } + + #[test] + fn write_receipts_require_an_exact_nonempty_memory_id() { + for response in [ + json!({}), + json!({"memory_id": null}), + json!({"memory_id": 7}), + json!({"memory_id": ""}), + json!({"memory_id": " "}), + json!({"memory_id": " id"}), + ] { + assert!(confirmed_memory_id(&response).is_err(), "{response}"); + } + assert_eq!( + confirmed_memory_id(&json!({"memory_id": "id-1"})).unwrap(), + "id-1" + ); + } + #[test] fn memory_scope_keeps_owner_and_session_as_distinct_exact_identities() { let scope = MemoryScope::new("user-1", "session-1").unwrap(); diff --git a/crates/astra-tools/src/memoria.rs b/crates/astra-tools/src/memoria.rs index 666d89d7ba..3160115aa4 100644 --- a/crates/astra-tools/src/memoria.rs +++ b/crates/astra-tools/src/memoria.rs @@ -1433,6 +1433,20 @@ impl MemoriaToolGateway { } return decorated; } + if op == "remember" { + match serde_json::from_str::(&raw_text) { + Ok(response) if response.get("error").is_some_and(|error| !error.is_null()) => {} + Ok(response) => { + if let Err(error) = astra_memoria::confirmed_memory_id(&response) { + return json!({"error": error}).to_string(); + } + } + Err(_) => { + return json!({"error": "memory backend returned a malformed write receipt"}) + .to_string(); + } + } + } if op == "expand" { return Self::decorate_expand_response( &raw_text, @@ -1640,14 +1654,12 @@ impl MemoriaToolGateway { } } - fn empty_success_response(op: &str, args: &Value) -> Value { + fn empty_success_response(op: &str, _args: &Value) -> Value { match op { "recall" => json!([]), - "remember" => json!({ - "status": "completed", - "message": "memory stored", - "content": args.get("content").and_then(Value::as_str).unwrap_or(""), - }), + "remember" => { + json!({"error": "memory backend did not return a valid memory_id receipt"}) + } "forget" => json!({ "status": "completed", "message": "memory purge completed", @@ -2847,14 +2859,13 @@ mod tests { } #[test] - fn empty_success_response_remember_returns_ack() { + fn empty_success_response_remember_rejects_missing_receipt() { let value = MemoriaToolGateway::empty_success_response( "remember", &json!({"content": "prefers smoke tests"}), ); - assert_eq!(value["status"], "completed"); - assert_eq!(value["message"], "memory stored"); - assert_eq!(value["content"], "prefers smoke tests"); + assert!(value.get("error").is_some()); + assert!(value.get("status").is_none()); } // ── Session isolation via v2 `scope` → v1 `session_scope` ── @@ -3134,6 +3145,52 @@ mod tests { ); } + #[tokio::test] + async fn remember_requires_a_confirmed_http_write_receipt() { + use wiremock::matchers::{method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + for body in [ + "", + "{}", + r#"{"memory_id":""}"#, + r#"{"memory_id":" "}"#, + r#"{"memory_id":12}"#, + "not-json", + r#"{"error":null}"#, + r#"{"memory_id":"mem-ok"}"#, + ] { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/memory/store")) + .respond_with(ResponseTemplate::new(201).set_body_string(body)) + .expect(1) + .mount(&server) + .await; + let gateway = MemoriaToolGateway::new(Some(server.uri()), Some("test-token".into())); + let output = gateway + .call_with_timeout( + "remember", + &json!({ + "content": "A durable preference for small focused changes", + "memory_type": "semantic" + }), + Duration::from_secs(1), + ) + .await; + let response: Value = serde_json::from_str(&output).unwrap(); + if body.contains("mem-ok") { + assert_eq!(response["memory_id"], "mem-ok"); + } else { + assert!( + response.get("error").is_some(), + "body={body:?}, output={output}" + ); + assert!(response.get("status").is_none()); + } + server.verify().await; + } + } + #[tokio::test] #[serial] async fn direct_remember_writes_once_without_retrieve_preflight() { diff --git a/crates/runtime/src/server/session/session_handlers.rs b/crates/runtime/src/server/session/session_handlers.rs index 5896fc0489..325c28c1ae 100644 --- a/crates/runtime/src/server/session/session_handlers.rs +++ b/crates/runtime/src/server/session/session_handlers.rs @@ -1891,9 +1891,16 @@ fn schedule_session_end_governance(state: &AppState, owner_id: String, session_i purged = report.working_purged, episode_chars = report.episode_chars, scenes_stored = report.scenes_stored, - "session-end governance complete" + working_purge_failed = report.working_purge_failed, + episode_write_failed = report.working_retained_due_to_episode_failure, + scene_persist_failures = report.scene_persist_failures, + reflection_failed = report.reflection_failed, + consolidation_complete = report.consolidation_complete(), + "session-end governance finished" ); - governance_permit.complete(); + if report.consolidation_complete() { + governance_permit.complete(); + } } Ok(Err(error)) => { tracing::warn!( diff --git a/crates/runtime/src/turn/cloud/memoria_compact.rs b/crates/runtime/src/turn/cloud/memoria_compact.rs index 7fae7169f9..aabcf016ce 100644 --- a/crates/runtime/src/turn/cloud/memoria_compact.rs +++ b/crates/runtime/src/turn/cloud/memoria_compact.rs @@ -78,7 +78,8 @@ impl Default for MemoriaCompactParams { pub use astra_memoria::{ MemoriaMemory, MemoriaOperationError, MemoriaPort, MemoriaToolTransport, MemoryScope, - ReflectCandidate, ReflectSummary, parse_reflect_candidates, validate_strict_memories, + ReflectCandidate, ReflectSummary, confirmed_memory_id, parse_reflect_candidates, + parse_reflect_summary, validate_strict_memories, }; fn cross_session_abstract(prefix: &str, evidence: &str) -> String { @@ -105,7 +106,8 @@ fn cross_session_abstract(prefix: &str, evidence: &str) -> String { abstract_text } -fn encode_episode_memory(session_id: &str, overview: &str) -> String { +/// Encode the canonical typed episode for every Memoria transport. +pub fn encode_episode_memory(session_id: &str, overview: &str) -> String { let goal = overview .lines() .find_map(|line| line.trim().strip_prefix("Goal: ")) @@ -121,7 +123,8 @@ fn encode_episode_memory(session_id: &str, overview: &str) -> String { .encode() } -fn encode_scene_memory(signal: &str, summary: &str) -> String { +/// Encode the canonical typed semantic scene for every Memoria transport. +pub fn encode_scene_memory(signal: &str, summary: &str) -> String { let prefix = if signal.trim().is_empty() { "Recurring cross-session pattern".to_string() } else { @@ -867,10 +870,7 @@ impl MemoriaPort for HttpMemoriaPort { .await .map_err(|e| format!("Memoria store parse failed: {e}"))?; - data.get("memory_id") - .and_then(Value::as_str) - .map(String::from) - .ok_or_else(|| "Memoria store: no memory_id in response".to_string()) + astra_memoria::confirmed_memory_id(&data) } /// Purge working memories for a session. @@ -990,11 +990,7 @@ impl MemoriaPort for HttpMemoriaPort { .json() .await .map_err(|e| format!("Memoria store_episode parse failed: {e}"))?; - Ok(data - .get("memory_id") - .and_then(Value::as_str) - .map(String::from) - .unwrap_or_default()) + astra_memoria::confirmed_memory_id(&data) } /// Persist a reflect scene candidate as a semantic memory with a stable @@ -1037,11 +1033,7 @@ impl MemoriaPort for HttpMemoriaPort { .json() .await .map_err(|e| format!("Memoria store_scene parse failed: {e}"))?; - Ok(data - .get("memory_id") - .and_then(Value::as_str) - .map(String::from) - .unwrap_or_default()) + astra_memoria::confirmed_memory_id(&data) } async fn reflect_session( @@ -1079,22 +1071,9 @@ impl MemoriaPort for HttpMemoriaPort { text.chars().take(120).collect::() )); } - let data: Value = serde_json::from_str(&text).unwrap_or(Value::Null); - let candidate_payloads = parse_reflect_candidates(&data); - let candidate_count = candidate_payloads.len() as u64; - Ok(ReflectSummary { - synthesized: data - .get("synthesized") - .and_then(Value::as_bool) - .unwrap_or(false), - candidates: candidate_count.max( - data.get("scenes_created") - .and_then(Value::as_u64) - .unwrap_or(0), - ), - candidate_payloads, - diagnostics: text.chars().take(200).collect(), - }) + let data: Value = serde_json::from_str(&text) + .map_err(|error| format!("Memoria reflect parse failed: {error}"))?; + Ok(astra_memoria::parse_reflect_summary(&data)) } async fn feedback( @@ -2723,6 +2702,44 @@ mod tests { assert!(body.get("user_id").is_none()); } + #[tokio::test] + async fn all_store_paths_reject_invalid_http_write_receipts() { + for body in [ + "{}", + r#"{"memory_id":""}"#, + r#"{"memory_id":" "}"#, + r#"{"memory_id":12}"#, + "not-json", + ] { + for operation in ["store", "episode", "scene"] { + let (address, _, server) = + capture_one_http_request("201 Created", body.as_bytes()).await; + let client = + HttpMemoriaPort::new_master(format!("http://{address}"), "test-key".into()) + .with_owner_user_id("owner-7"); + let result = match operation { + "store" => { + client + .store("a durable fact", "semantic", Some("session-9"), Some("T3")) + .await + } + "episode" => { + client + .store_episode("session-9", "completed the migration") + .await + } + _ => { + client + .store_scene("session-9", "pattern", "a recurring pattern") + .await + } + }; + assert!(result.is_err(), "operation={operation}, body={body}"); + server.await.unwrap(); + } + } + } + #[tokio::test] async fn episode_store_matches_memoria_schema_and_owner_routing() { let response = br#"{"memory_id":"episode-1"}"#; diff --git a/crates/runtime/src/turn/cloud/session_end_governance.rs b/crates/runtime/src/turn/cloud/session_end_governance.rs index ff8c7e2b1f..78b6d485ee 100644 --- a/crates/runtime/src/turn/cloud/session_end_governance.rs +++ b/crates/runtime/src/turn/cloud/session_end_governance.rs @@ -7,9 +7,8 @@ //! learnings, and structured [`SessionFacts`] (deterministic, no LLM call). //! 3. **Purge working memory** tied to the session only after the snapshot read //! succeeds and any required episode is durable. -//! 4. **Trigger reflection** so Memoria's graph-consolidation picks up -//! recent memories into scene nodes. Respects the backend's cooldown -//! (v1 defaults to 1h), so hot sessions won't thrash. +//! 4. **Request reflection candidates** and persist optional semantic scenes. +//! Reflection is best-effort and does not control consolidation cooldown. //! //! Governance is best-effort. A failed snapshot read stops subsequent work; //! a failed episode write retains working memory while reflection may proceed. @@ -210,8 +209,16 @@ pub async fn run_session_end_governance( if let Some(overview) = build_episode_overview_with_narrative(facts, snapshot.as_ref().map(|s| &s.narrative)) { - match client.store_episode(session_id, &overview).await { - Ok(memory_id) if !memory_id.is_empty() => { + let episode_receipt = + client + .store_episode(session_id, &overview) + .await + .and_then(|memory_id| { + astra_memoria::validate_memory_id(&memory_id)?; + Ok(memory_id) + }); + match episode_receipt { + Ok(memory_id) => { report.episode_memory_id = Some(memory_id); report.episode_chars = overview.chars().count(); tracing::debug!( @@ -220,11 +227,6 @@ pub async fn run_session_end_governance( "session-end episode stored" ); } - Ok(_) => { - // store succeeded but response didn't include a memory_id; - // count it as a write but leave id blank. - report.episode_chars = overview.chars().count(); - } Err(e) => { safe_to_purge_working = false; report.working_retained_due_to_episode_failure = true; @@ -278,14 +280,17 @@ pub async fn run_session_end_governance( report.reflect_candidates = summary.candidates; report.reflect_synthesized = summary.synthesized; for cand in &summary.candidate_payloads { - match client + let scene_receipt = client .store_scene(session_id, &cand.signal, &cand.summary) .await - { - Ok(memory_id) if !memory_id.is_empty() => { + .and_then(|memory_id| { + astra_memoria::validate_memory_id(&memory_id)?; + Ok(memory_id) + }); + match scene_receipt { + Ok(_) => { report.scenes_stored += 1; } - Ok(_) => report.scenes_stored += 1, Err(e) => { report.scene_persist_failures += 1; tracing::warn!( @@ -298,8 +303,9 @@ pub async fn run_session_end_governance( } } Err(e) => { - // Cooldown rejection is expected under hot activity; keep it in - // structured debug logs and continue without user-facing noise. + report.reflection_failed = true; + // Reflection is optional. Preserve diagnostics without making a + // later close replay already completed episode/purge stages. tracing::debug!( session_id, error = %e, @@ -333,6 +339,17 @@ pub struct SessionEndReport { pub scenes_stored: usize, /// Scene candidates that could not be persisted after reflection. pub scene_persist_failures: usize, + /// Optional reflection failed; this does not invalidate durable consolidation. + pub reflection_failed: bool, +} + +impl SessionEndReport { + /// Only required episode persistence and working cleanup govern the success + /// cooldown. Repeating those destructive stages to retry optional reflection + /// or scene synthesis would replay already completed consolidation. + pub fn consolidation_complete(&self) -> bool { + !self.working_purge_failed && !self.working_retained_due_to_episode_failure + } } #[cfg(test)] @@ -523,6 +540,7 @@ mod tests { episodes: Mutex>, reflect_response: Mutex, fail_scene: bool, + empty_scene_receipt: bool, } #[async_trait::async_trait] @@ -571,6 +589,9 @@ mod tests { if self.fail_scene { return Err("scene store unavailable".into()); } + if self.empty_scene_receipt { + return Ok(String::new()); + } self.scenes.lock().unwrap().push(( sid.to_string(), signal.to_string(), @@ -580,6 +601,28 @@ mod tests { } } + #[tokio::test] + async fn governance_does_not_count_unconfirmed_scenes_as_stored() { + let client = SceneCaptureClient { + empty_scene_receipt: true, + ..Default::default() + }; + *client.reflect_response.lock().unwrap() = ReflectSummary { + candidate_payloads: vec![ReflectCandidate { + signal: "pattern".into(), + summary: "recurring evidence".into(), + importance: 1.0, + }], + ..Default::default() + }; + let report = run_session_end_governance(&SessionFacts::default(), "scene-receipt", &client) + .await + .unwrap(); + assert_eq!(report.scenes_stored, 0); + assert_eq!(report.scene_persist_failures, 1); + assert!(report.consolidation_complete()); + } + #[tokio::test] async fn governance_forward_feeds_reflect_candidates_as_scenes() { let client = Arc::new(SceneCaptureClient::default()); @@ -671,7 +714,9 @@ mod tests { read_error: Option, empty_snapshot: bool, fail_episode: bool, + episode_receipt: Option, fail_purge: bool, + fail_reflect: bool, } #[async_trait::async_trait] @@ -721,7 +766,10 @@ mod tests { if self.fail_episode { Err("episode store unavailable".into()) } else { - Ok("episode-1".into()) + Ok(self + .episode_receipt + .clone() + .unwrap_or_else(|| "episode-1".into())) } } @@ -740,6 +788,9 @@ mod tests { _force: bool, ) -> Result { self.operations.lock().unwrap().push("reflect".into()); + if self.fail_reflect { + return Err("reflection unavailable".into()); + } Ok(Default::default()) } } @@ -843,6 +894,90 @@ mod tests { ); assert_eq!(report.working_purged, 0); assert!(report.working_retained_due_to_episode_failure); + assert!(!report.consolidation_complete()); + } + + #[tokio::test] + async fn governance_retains_working_on_unconfirmed_episode_receipts() { + for receipt in ["", " ", " episode-1"] { + let client = LifecycleOrderClient { + episode_receipt: Some(receipt.into()), + ..Default::default() + }; + let report = + run_session_end_governance(&SessionFacts::default(), "session-lifecycle", &client) + .await + .unwrap(); + assert!(report.working_retained_due_to_episode_failure); + assert!(!report.consolidation_complete()); + assert!(report.episode_memory_id.is_none()); + assert_eq!(report.episode_chars, 0); + assert!( + !client + .operations + .lock() + .unwrap() + .iter() + .any(|operation| operation == "purge") + ); + } + } + + #[tokio::test] + async fn completed_consolidation_starts_cooldown_after_optional_reflection_failure() { + let client = LifecycleOrderClient { + fail_reflect: true, + ..Default::default() + }; + let debouncer = crate::turn::session_end_debounce::SessionEndDebouncer::new(); + let permit = debouncer.try_begin("owner", "session-lifecycle").unwrap(); + let report = + run_session_end_governance(&SessionFacts::default(), "session-lifecycle", &client) + .await + .unwrap(); + assert!(report.reflection_failed); + assert_eq!(report.episode_memory_id.as_deref(), Some("episode-1")); + assert_eq!(report.working_purged, 1); + assert!(report.consolidation_complete()); + permit.complete(); + assert!(debouncer.try_begin("owner", "session-lifecycle").is_none()); + assert_eq!( + client.operations.lock().unwrap().as_slice(), + ["retrieve", "store_episode", "purge", "reflect"] + ); + } + + #[test] + fn optional_scene_failures_do_not_replay_completed_consolidation() { + let report = SessionEndReport { + scene_persist_failures: 1, + ..Default::default() + }; + assert!(report.consolidation_complete()); + } + + #[test] + fn partial_governance_does_not_start_a_success_cooldown() { + use crate::turn::session_end_debounce::SessionEndDebouncer; + let debouncer = SessionEndDebouncer::new(); + for report in [ + SessionEndReport { + working_purge_failed: true, + ..Default::default() + }, + SessionEndReport { + working_retained_due_to_episode_failure: true, + ..Default::default() + }, + ] { + let permit = debouncer.try_begin("owner", "session").unwrap(); + assert!(!report.consolidation_complete()); + drop(permit); + } + let permit = debouncer.try_begin("owner", "session").unwrap(); + assert!(SessionEndReport::default().consolidation_complete()); + permit.complete(); + assert!(debouncer.try_begin("owner", "session").is_none()); } #[tokio::test] diff --git a/crates/runtime/src/turn/session_end_debounce.rs b/crates/runtime/src/turn/session_end_debounce.rs index 7c2aa5fc18..ec71f8342e 100644 --- a/crates/runtime/src/turn/session_end_debounce.rs +++ b/crates/runtime/src/turn/session_end_debounce.rs @@ -21,8 +21,8 @@ use std::time::{Duration, Instant}; /// Minimum gap between governance runs for the same session. Below this /// the subsequent run is considered "still active" and governance is -/// deferred. Aligns with Memoria's reflect 1h cooldown but is client- -/// side so even purge_working and store_episode are debounced. +/// deferred. This bounds repeated episode/purge work independently of the +/// backend's optional reflection policy. pub const GOVERNANCE_MIN_INTERVAL: Duration = Duration::from_secs(15 * 60); const MAX_COMPLETED_SESSIONS: usize = 16 * 1024; diff --git a/docs/design/memory.md b/docs/design/memory.md index 95296b5ea6..15b00f8870 100644 --- a/docs/design/memory.md +++ b/docs/design/memory.md @@ -96,7 +96,23 @@ write checks remain required. Session-end governance must distinguish a successful empty snapshot read from a failed or denied read. Any snapshot read error stops destructive governance -and retains working memories for a later retry. Session-start recall applies +and retains working memories for a later retry. Required writes must return an +exact, nonempty memory ID: an empty successful HTTP response or an unsupported +transport operation is not proof of storage. Unconfirmed episodes retain working +memory; unconfirmed scenes are failed writes. Only required episode persistence +and working cleanup determine the consolidation success cooldown. Optional +reflection/scene failures remain observable but do not replay those completed +stages. Failed episode persistence or cleanup retains retry eligibility; this +is not an automatic durable retry queue. + +The CLI transport persists typed episodes and scenes through the authenticated +Server memory proxy, sharing the direct transport's encoding and receipt +validation. It requests reflection candidates through the same proxy; successful +backend cooldown skips are empty results. CLI exit retains its shared five-second +maintenance budget and cancels/drains unfinished work rather than granting purge +on a timeout. A required archive must be confirmed before working cleanup. + +Session-start recall applies each profile/episode cap only after typed prompt admission, so invalid or inactive rows cannot displace valid evidence already returned by the backend.