fix(memory): confirm CLI archive writes and preserve consolidation cooldown

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<String, String> {
+        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<String, String> {
+        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::ReflectSummary, String> {
+        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<u64, String> {
         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() {

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<String, String> {
+    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<String, String> {
-        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<String, String> {
-        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<ReflectCandidate> {
         .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<Vec<MemoriaMemory>, MemoriaOperationError> {
+            Ok(Vec::new())
+        }
+        async fn store(
+            &self,
+            _: &str,
+            _: &str,
+            _: Option<&str>,
+            _: Option<&str>,
+        ) -> Result<String, String> {
+            unreachable!("unsupported persistence must not fabricate a write")
+        }
+        async fn purge_working(&self, _: &str) -> Result<u64, String> {
+            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();