Make managed-agent recovery contextual and preserve conversations

crates/freshell-agent-runtime/src/host_actor.rs

@@ -237,6 +237,15 @@ pub trait FreshAgentTransport: Send + Sync {
     async fn capture(&self, _max_bytes: usize) -> Result<FreshAgentCapture, String> {
         Err("provider does not expose a hosted snapshot".into())
     }
+    async fn snapshot(&self) -> Result<Value, String> {
+        Err("provider does not expose a hosted snapshot".into())
+    }
+
+    /// A registered zero-turn identity is transport-owned, not a durable native identity.
+    /// Only transports that can prove their current local registration opt in.
+    async fn registered_snapshot_identity(&self) -> Option<String> {
+        None
+    }
     /// Whether this actor still owns a usable provider enclosure. Provider
     /// adapters may self-heal a child internally; they should report false
     /// only when no live owned session remains.
@@ -871,6 +880,64 @@ impl FreshAgentHostActor {
             .map_err(ActorError::Transport)
     }
 
+    pub async fn snapshot(&self) -> Result<Value, ActorError> {
+        if !self.transport.is_live().await {
+            return Err(ActorError::Transport("provider is not live".into()));
+        }
+        let snapshot = self
+            .transport
+            .snapshot()
+            .await
+            .map_err(ActorError::Transport)?;
+        if !self.transport.is_live().await {
+            return Err(ActorError::Transport(
+                "provider exited during snapshot read".into(),
+            ));
+        }
+        let profile = self.profile().await;
+        let provider = if profile.provider == freshell_runtime_protocol::FreshProvider::Kilroy {
+            "claude"
+        } else {
+            profile.provider.as_str()
+        };
+        let session_type = match profile.provider {
+            FreshProvider::Claude => "freshclaude",
+            FreshProvider::Codex => "freshcodex",
+            FreshProvider::Opencode => "freshopencode",
+            FreshProvider::Kilroy => "kilroy",
+        };
+        let registered =
+            if profile.native_session_id.is_none() && profile.provider == FreshProvider::Opencode {
+                self.transport.registered_snapshot_identity().await
+            } else {
+                None
+            };
+        // Recheck after the registration read: materialization must win over a placeholder.
+        let profile = self.profile().await;
+        let expected = profile
+            .native_session_id
+            .as_deref()
+            .or(registered.as_deref());
+        if expected.is_none()
+            || snapshot["threadId"].as_str() != expected
+            || snapshot["provider"].as_str() != Some(provider)
+            || snapshot["sessionType"].as_str() != Some(session_type)
+        {
+            return Err(ActorError::NativeIdentityMismatch);
+        }
+        // Snapshots use the existing native-history reply allowance, including envelope overhead.
+        if serde_json::to_vec(&snapshot)
+            .map_err(|error| ActorError::Transport(error.to_string()))?
+            .len()
+            > freshell_runtime_protocol::MAX_NATIVE_HISTORY_FRAME_BYTES - 4096
+        {
+            return Err(ActorError::Transport(
+                "provider snapshot exceeds history reply frame limit".into(),
+            ));
+        }
+        Ok(snapshot)
+    }
+
     pub async fn record_event(&self, event: AgentEvent) -> Result<u64, ActorError> {
         let mut state = self.state.lock().await;
         let sequence =

crates/freshell-agent-runtime/src/host_actor_tests.rs

@@ -1,6 +1,192 @@
 use super::*;
 use std::sync::atomic::{AtomicUsize, Ordering};
 
+struct SnapshotTransport {
+    value: Value,
+    live: std::sync::atomic::AtomicBool,
+    exit_during_read: bool,
+    initial_native: Option<String>,
+    registered: Option<String>,
+    pause: Option<(Arc<tokio::sync::Notify>, Arc<tokio::sync::Notify>)>,
+}
+
+#[async_trait]
+impl FreshAgentTransport for SnapshotTransport {
+    async fn start(&self, _: &FreshAgentProfile) -> Result<TransportStart, String> {
+        Ok(TransportStart {
+            native_session_id: self.initial_native.clone(),
+        })
+    }
+    async fn dispatch(
+        &self,
+        _: &RequestId,
+        _: &str,
+        _: &FreshAgentProfile,
+    ) -> Result<DispatchAck, DispatchFailure> {
+        panic!("snapshot must not dispatch")
+    }
+    async fn resolve_permission(&self, _: &str, _: Value) -> Result<(), DispatchFailure> {
+        panic!("snapshot must not resolve")
+    }
+    async fn interrupt(&self) -> Result<(), String> {
+        panic!("snapshot must not interrupt")
+    }
+    async fn stop(self: Arc<Self>) -> Result<(), String> {
+        Ok(())
+    }
+    fn take_event_stream(&self) -> Option<tokio::sync::mpsc::Receiver<AgentEvent>> {
+        None
+    }
+    async fn is_live(&self) -> bool {
+        self.live.load(Ordering::SeqCst)
+    }
+    async fn registered_snapshot_identity(&self) -> Option<String> {
+        self.registered.clone()
+    }
+    async fn snapshot(&self) -> Result<Value, String> {
+        if let Some((entered, release)) = &self.pause {
+            entered.notify_one();
+            release.notified().await;
+        }
+        if self.exit_during_read {
+            self.live.store(false, Ordering::SeqCst);
+        }
+        Ok(self.value.clone())
+    }
+}
+
+#[tokio::test]
+async fn snapshot_read_preserves_actor_state_and_rejects_wrong_identity_size_or_liveness() {
+    for scenario in [
+        "live",
+        "large",
+        "wrong-thread",
+        "wrong-provider",
+        "wrong-type",
+        "oversized",
+        "not-live",
+        "exit-during-read",
+    ] {
+        let dir = tempfile::tempdir().unwrap();
+        let transport = Arc::new(SnapshotTransport {
+            value: serde_json::json!({
+                "threadId":if scenario == "wrong-thread" { "different-thread" } else { "snapshot-native" },
+                "provider":if scenario == "wrong-provider" { "codex" } else { "claude" },
+                "sessionType":if scenario == "wrong-type" { "freshopencode" } else { "freshclaude" }, "status":"idle",
+                "turns":match scenario { "oversized" => "x".repeat(freshell_runtime_protocol::MAX_NATIVE_HISTORY_FRAME_BYTES), "large" => "x".repeat(2 * freshell_runtime_protocol::MAX_CONTROL_FRAME_BYTES), _ => "retained".into() },
+            }),
+            live: std::sync::atomic::AtomicBool::new(scenario != "not-live"),
+            exit_during_read: scenario == "exit-during-read",
+            initial_native: Some("snapshot-native".into()),
+            registered: None,
+            pause: None,
+        });
+        let actor = FreshAgentHostActor::open(
+            dir.path(),
+            profile(FreshProvider::Claude, "snapshot-store", None),
+            transport,
+        )
+        .await
+        .unwrap();
+        let before = fs::read(dir.path().join("fresh-agent-state.json")).unwrap();
+        let result = actor.snapshot().await;
+        assert_eq!(
+            result.is_ok(),
+            matches!(scenario, "live" | "large"),
+            "{scenario}"
+        );
+        if let Ok(snapshot) = result {
+            assert_eq!(snapshot["threadId"], "snapshot-native");
+            if scenario == "large" {
+                assert_eq!(
+                    snapshot["turns"].as_str().unwrap().len(),
+                    2 * freshell_runtime_protocol::MAX_CONTROL_FRAME_BYTES
+                );
+            }
+        }
+        assert_eq!(
+            fs::read(dir.path().join("fresh-agent-state.json")).unwrap(),
+            before,
+            "{scenario}"
+        );
+    }
+}
+
+#[tokio::test]
+async fn registered_pre_native_snapshot_requires_exact_opencode_proof_and_current_materialization()
+{
+    for scenario in [
+        "exact",
+        "default-deny",
+        "foreign-placeholder",
+        "other-provider",
+        "materialized-during-read",
+    ] {
+        let dir = tempfile::tempdir().unwrap();
+        let entered = Arc::new(tokio::sync::Notify::new());
+        let release = Arc::new(tokio::sync::Notify::new());
+        let transport = Arc::new(SnapshotTransport {
+            value: serde_json::json!({"threadId":"freshopencode-owned", "provider":"opencode", "sessionType":"freshopencode", "turns":[]}),
+            live: std::sync::atomic::AtomicBool::new(true),
+            exit_during_read: false,
+            initial_native: None,
+            registered: if scenario == "default-deny" {
+                None
+            } else {
+                Some(
+                    if scenario == "foreign-placeholder" {
+                        "freshopencode-foreign"
+                    } else {
+                        "freshopencode-owned"
+                    }
+                    .into(),
+                )
+            },
+            pause: (scenario == "materialized-during-read")
+                .then(|| (entered.clone(), release.clone())),
+        });
+        let actor = FreshAgentHostActor::open(
+            dir.path(),
+            profile(
+                if scenario == "other-provider" {
+                    FreshProvider::Claude
+                } else {
+                    FreshProvider::Opencode
+                },
+                "owned",
+                None,
+            ),
+            transport,
+        )
+        .await
+        .unwrap();
+        let before = fs::read(dir.path().join("fresh-agent-state.json")).unwrap();
+        let read = tokio::spawn({
+            let actor = actor.clone();
+            async move { actor.snapshot().await }
+        });
+        if scenario == "materialized-during-read" {
+            entered.notified().await;
+            actor
+                .observe_native_identity("ses_materialized".into())
+                .await
+                .unwrap();
+            release.notify_one();
+        }
+        assert_eq!(
+            read.await.unwrap().is_ok(),
+            scenario == "exact",
+            "{scenario}"
+        );
+        if scenario != "materialized-during-read" {
+            assert_eq!(
+                fs::read(dir.path().join("fresh-agent-state.json")).unwrap(),
+                before
+            );
+        }
+    }
+}
+
 struct OperationTransport {
     operations: std::sync::Mutex<Vec<(String, FreshAgentOperation)>>,
     supported: bool,

crates/freshell-agent-runtime/src/snapshot_projection.rs

@@ -23,14 +23,35 @@ const BOOLEAN_CAPABILITIES: &[&str] = &[
 /// Rewrites every recognized snapshot in `payload`, including snapshots nested
 /// in a `freshAgent.event` envelope. Non-snapshot provider events are untouched.
 pub fn project_hosted_snapshot(payload: &mut Value, provider: &str, session_type: &str) {
-    let known_pair = matches!(
+    visit(
+        payload,
+        provider,
+        session_type,
+        known_pair(provider, session_type),
+    );
+}
+
+fn known_pair(provider: &str, session_type: &str) -> bool {
+    matches!(
         (provider, session_type),
         ("claude", "freshclaude")
             | ("claude", "kilroy")
             | ("codex", "freshcodex")
             | ("opencode", "freshopencode")
-    );
-    visit(payload, provider, session_type, known_pair);
+    )
+}
+
+/// The existing REST snapshot has no event `type` field.
+pub fn project_hosted_rest_snapshot(snapshot: &mut Value, provider: &str, session_type: &str) {
+    if let Some(object) = snapshot.as_object_mut() {
+        let identity_matches = object.get("provider").and_then(Value::as_str) == Some(provider)
+            && object.get("sessionType").and_then(Value::as_str) == Some(session_type);
+        project_capabilities(
+            object,
+            provider,
+            known_pair(provider, session_type) && identity_matches,
+        );
+    }
 }
 
 fn visit(value: &mut Value, provider: &str, session_type: &str, known_pair: bool) {

crates/freshell-freshagent/Cargo.toml

@@ -72,6 +72,7 @@ async-trait = "0.1"
 serde = { workspace = true }
 serde_json = { workspace = true }
 sha2 = "0.10"
+rusqlite = { version = "0.31", features = ["bundled", "blob"] }
 # `<think>`/`<thinking>` balanced-tag segmentation for opencode assistant text
 # (`itemsFromAssistantTextPart`/`normalizeBalancedThinkTags`, normalize.ts:100-189) needs a
 # backreference (`<(thinking|think)...>...</\1>`) to match only same-name open/close pairs --