feat: observe retained worker sessions without restore

crates/session-store/src/lib.rs

@@ -57,6 +57,10 @@ pub use history::{
 };
 pub use logged_item::{LoggedContentPart, LoggedItem, LoggedRole, from_logged, to_logged};
 pub use paste_artifact::PasteArtifactLimits;
+pub use public_snapshot::{
+    DEFAULT_RETAINED_SNAPSHOT_MAX_BYTES, RetainedSessionIdentity, RetainedSessionSnapshot,
+    RetainedSnapshotReadError, project_session_snapshot, read_retained_session_snapshot,
+};
 pub use segment::{
     SegmentStartState, append_entry, append_system_item, classify_logged_history_entry,
     create_compacted_segment, create_segment, create_segment_with_ids, ensure_head_or_fork, fork,

crates/session-store/src/public_snapshot.rs

@@ -7,11 +7,119 @@ use protocol::{
     SessionSnapshotEntry, SessionSnapshotEntryData, SessionToolAttachment,
 };
 
+use std::path::Path;
+
+use thiserror::Error;
+
 use crate::{
     LogEntry, LoggedContentPart, LoggedHistoryEntry, LoggedItem, LoggedRole,
-    LoggedSessionHistoryOrigin, SessionId, SystemItem,
+    LoggedSessionHistoryOrigin, SessionId, Store, StoreError, SystemItem, WorkerAggregateStore,
+    WorkerMetadataStore, WorkerSessionStore, WorkerStoreError,
 };
 
+/// Projection limit leaves one MiB for the typed API envelope so a public
+/// retained-session response remains below the 16 MiB response ceiling.
+pub const DEFAULT_RETAINED_SNAPSHOT_MAX_BYTES: u64 = 15 * 1024 * 1024;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RetainedSessionIdentity {
+    pub session_id: String,
+    pub segment_id: String,
+    pub entry_count: u64,
+}
+
+#[derive(Debug, Clone, PartialEq)]
+pub struct RetainedSessionSnapshot {
+    pub identity: RetainedSessionIdentity,
+    pub snapshot: SessionSnapshot,
+}
+
+#[derive(Debug, Error)]
+pub enum RetainedSnapshotReadError {
+    #[error("retained worker state is unavailable")]
+    RetentionMissing,
+    #[error("retained worker state has no active session pointer")]
+    ActivePointerMissing,
+    #[error("retained session requires migration")]
+    MigrationRequired,
+    #[error("retained session log is corrupt")]
+    CorruptLog,
+    #[error("retained session snapshot exceeds the observation limit")]
+    SnapshotTooLarge,
+}
+
+/// Read the active retained Session without creating, migrating, restoring, or
+/// rewriting any persisted state.
+pub fn read_retained_session_snapshot(
+    aggregate_root: &Path,
+    worker_name: &str,
+    max_bytes: u64,
+) -> Result<RetainedSessionSnapshot, RetainedSnapshotReadError> {
+    let aggregate = WorkerAggregateStore::open_read_only(aggregate_root, worker_name)
+        .map_err(map_worker_store_error)?;
+    let metadata = aggregate
+        .read_by_name(worker_name)
+        .map_err(map_worker_store_error)?
+        .ok_or(RetainedSnapshotReadError::RetentionMissing)?;
+    let active = metadata
+        .active
+        .ok_or(RetainedSnapshotReadError::ActivePointerMissing)?;
+    let segment_id = active
+        .segment_id
+        .ok_or(RetainedSnapshotReadError::ActivePointerMissing)?;
+    let store = WorkerSessionStore::open_read_only(aggregate_root.join("session"))
+        .map_err(map_store_error)?;
+    if store.segment_log_len(segment_id).map_err(map_store_error)? > max_bytes {
+        return Err(RetainedSnapshotReadError::SnapshotTooLarge);
+    }
+    let entries = store
+        .read_all(active.session_id, segment_id)
+        .map_err(map_store_error)?;
+    let snapshot = project_session_snapshot(active.session_id, &entries);
+    if serde_json::to_vec(&snapshot)
+        .map_err(|_| RetainedSnapshotReadError::CorruptLog)?
+        .len() as u64
+        > max_bytes
+    {
+        return Err(RetainedSnapshotReadError::SnapshotTooLarge);
+    }
+    Ok(RetainedSessionSnapshot {
+        identity: RetainedSessionIdentity {
+            session_id: active.session_id.to_string(),
+            segment_id: segment_id.to_string(),
+            entry_count: entries.len() as u64,
+        },
+        snapshot,
+    })
+}
+
+fn map_worker_store_error(error: WorkerStoreError) -> RetainedSnapshotReadError {
+    match error {
+        WorkerStoreError::Io(error) if error.kind() == std::io::ErrorKind::NotFound => {
+            RetainedSnapshotReadError::RetentionMissing
+        }
+        WorkerStoreError::Serde(_) | WorkerStoreError::InvalidWorkerName(_) => {
+            RetainedSnapshotReadError::CorruptLog
+        }
+        WorkerStoreError::Io(_) => RetainedSnapshotReadError::CorruptLog,
+    }
+}
+
+fn map_store_error(error: StoreError) -> RetainedSnapshotReadError {
+    match error {
+        StoreError::Io(error) if error.kind() == std::io::ErrorKind::NotFound => {
+            RetainedSnapshotReadError::RetentionMissing
+        }
+        StoreError::Corrupt { message, .. } if message.contains("requires migration") => {
+            RetainedSnapshotReadError::MigrationRequired
+        }
+        StoreError::Io(_) | StoreError::Serde(_) | StoreError::Corrupt { .. } => {
+            RetainedSnapshotReadError::CorruptLog
+        }
+        _ => RetainedSnapshotReadError::CorruptLog,
+    }
+}
+
 /// Project a complete current-segment log. A valid segment starts with one
 /// canonical annotated SegmentStart record; malformed partial input uses the
 /// nil session only to keep the public failure projection deterministic.
@@ -559,4 +667,65 @@ mod tests {
             SessionEntryProvenance::ModelOutput
         );
     }
+
+    #[test]
+    fn retained_snapshot_read_projects_active_segment_without_rewriting_files() {
+        let root = tempfile::tempdir().unwrap();
+        let aggregate_root = root.path().join("worker-a");
+        let aggregate = WorkerAggregateStore::new(&aggregate_root, "worker-a").unwrap();
+        let session_id = crate::new_session_id();
+        let segment_id = crate::new_segment_id();
+        aggregate
+            .write(&crate::WorkerMetadata::new(
+                "worker-a",
+                Some(crate::WorkerActiveSegmentRef::active_segment(
+                    session_id, segment_id,
+                )),
+            ))
+            .unwrap();
+        let session = WorkerSessionStore::new(aggregate_root.join("session")).unwrap();
+        session.create_segment(session_id, segment_id, &[]).unwrap();
+        let metadata_before = std::fs::read(aggregate_root.join("metadata.json")).unwrap();
+        let manifest_before = std::fs::read(aggregate_root.join("session/session.json")).unwrap();
+
+        let retained = read_retained_session_snapshot(
+            &aggregate_root,
+            "worker-a",
+            DEFAULT_RETAINED_SNAPSHOT_MAX_BYTES,
+        )
+        .unwrap();
+
+        assert_eq!(retained.identity.session_id, session_id.to_string());
+        assert_eq!(retained.identity.segment_id, segment_id.to_string());
+        assert_eq!(retained.identity.entry_count, 0);
+        assert!(retained.snapshot.entries.is_empty());
+        assert_eq!(
+            std::fs::read(aggregate_root.join("metadata.json")).unwrap(),
+            metadata_before
+        );
+        assert_eq!(
+            std::fs::read(aggregate_root.join("session/session.json")).unwrap(),
+            manifest_before
+        );
+        assert!(matches!(
+            read_retained_session_snapshot(&aggregate_root, "worker-a", 0),
+            Err(RetainedSnapshotReadError::SnapshotTooLarge)
+        ));
+    }
+
+    #[test]
+    fn retained_snapshot_read_does_not_create_missing_aggregate() {
+        let root = tempfile::tempdir().unwrap();
+        let aggregate = root.path().join("missing-worker");
+
+        let error = read_retained_session_snapshot(
+            &aggregate,
+            "missing-worker",
+            DEFAULT_RETAINED_SNAPSHOT_MAX_BYTES,
+        )
+        .unwrap_err();
+
+        assert!(matches!(error, RetainedSnapshotReadError::RetentionMissing));
+        assert!(!aggregate.exists());
+    }
 }

crates/session-store/src/worker_metadata.rs

@@ -383,6 +383,27 @@ impl WorkerAggregateStore {
         })
     }
 
+    /// Open an existing aggregate without creating or rewriting retained files.
+    pub fn open_read_only(
+        root: impl Into<PathBuf>,
+        worker_name: impl Into<String>,
+    ) -> Result<Self, WorkerStoreError> {
+        let root = root.into();
+        let worker_name = worker_name.into();
+        validate_worker_name(&worker_name)?;
+        if !root.is_dir() {
+            return Err(WorkerStoreError::Io(std::io::Error::new(
+                std::io::ErrorKind::NotFound,
+                "worker aggregate is not retained",
+            )));
+        }
+        Ok(Self {
+            update_lock: metadata_lock(&root),
+            root,
+            worker_name,
+        })
+    }
+
     fn validate_name(&self, worker_name: &str) -> Result<(), WorkerStoreError> {
         validate_worker_name(worker_name)?;
         if worker_name == self.worker_name {

crates/session-store/src/worker_session_store.rs

@@ -84,6 +84,32 @@ impl WorkerSessionStore {
         })
     }
 
+    /// Open an already-retained Session without creating directories or migrating
+    /// persisted data. Observation paths must never mutate retained state.
+    pub fn open_read_only(root: impl Into<PathBuf>) -> Result<Self, StoreError> {
+        let root = root.into();
+        let bytes = fs::read(root.join(SESSION_FILE))?;
+        let manifest: SessionManifest = serde_json::from_slice(&bytes)?;
+        if manifest.schema_version != SESSION_SCHEMA_VERSION {
+            return Err(StoreError::Corrupt {
+                line: 0,
+                message: format!(
+                    "Worker Session schema version {} requires migration; expected {}",
+                    manifest.schema_version, SESSION_SCHEMA_VERSION
+                ),
+            });
+        }
+        Ok(Self {
+            root,
+            session_id: Arc::new(Mutex::new(Some(manifest.session_id))),
+            append_lock: Arc::new(Mutex::new(())),
+        })
+    }
+
+    pub fn segment_log_len(&self, segment_id: SegmentId) -> Result<u64, StoreError> {
+        Ok(fs::metadata(self.log_path(segment_id))?.len())
+    }
+
     pub fn root_dir(&self) -> &Path {
         &self.root
     }