feat: wire pod metadata lifecycle writes

crates/pod/tests/compact_events_test.rs

@@ -17,7 +17,7 @@ use llm_worker::llm_client::event::{Event as LlmEvent, ResponseStatus, StatusEve
 use llm_worker::llm_client::types::Item;
 use llm_worker::llm_client::{ClientError, LlmClient, Request};
 use protocol::Event;
-use session_store::{FsStore, LogEntry, Store};
+use session_store::{FsStore, LogEntry, PodMetadataStore, Store};
 use tokio::sync::broadcast;
 
 use pod::Pod;
@@ -158,7 +158,9 @@ async fn make_pod_with_manifest(
     std::mem::forget(pwd_tmp);
 
     let worker = Worker::new(client);
-    Pod::new(manifest, worker, store, pwd, scope).await.unwrap()
+    let mut pod = Pod::new(manifest, worker, store, pwd, scope).await.unwrap();
+    pod.enable_pod_metadata_write_through().unwrap();
+    pod
 }
 
 async fn make_pod(client: MockClient) -> Pod<MockClient, FsStore> {
@@ -213,6 +215,36 @@ fn system_texts_in_sink_session_start(
     Vec::new()
 }
 
+/// Pod metadata starts with a reserved Session and no Segment, then becomes
+/// active once the first SegmentStart is materialized by `run`.
+#[tokio::test]
+async fn pod_metadata_moves_from_pending_to_active_on_first_run() {
+    let client = MockClient::new(vec![single_text_events("hi")]);
+    let mut pod = make_pod(client).await;
+    let store = pod.store().clone();
+    let session_id = pod.session_id();
+    let initial_segment_id = pod.segment_id();
+
+    let pending = store
+        .read_by_name("test-pod")
+        .unwrap()
+        .expect("metadata should be initialized at Pod construction");
+    assert_eq!(pending.pod_name, "test-pod");
+    let pending_active = pending.active.expect("active session pointer missing");
+    assert_eq!(pending_active.session_id, session_id);
+    assert_eq!(pending_active.segment_id, None);
+
+    pod.run_text("first").await.unwrap();
+
+    let resolved = store
+        .read_by_name("test-pod")
+        .unwrap()
+        .expect("metadata should still exist after first run");
+    let active = resolved.active.expect("active session pointer missing");
+    assert_eq!(active.session_id, session_id);
+    assert_eq!(active.segment_id, Some(initial_segment_id));
+}
+
 /// Live auto-fork: when another writer extends the segment behind the
 /// Pod's back, the next run's `ensure_segment_head` detects the
 /// entry-count drift and branches into a fresh segment **within the same
@@ -274,6 +306,13 @@ permission = "write"
     let new_segment_id = pod.segment_id();
     assert_ne!(new_segment_id, source_segment_id);
     assert_eq!(pod.session_id(), session_id, "auto-fork stays in-Session");
+    let metadata = store
+        .read_by_name("test-pod")
+        .unwrap()
+        .expect("metadata should exist after auto-fork");
+    let active = metadata.active.expect("active session pointer missing");
+    assert_eq!(active.session_id, session_id);
+    assert_eq!(active.segment_id, Some(new_segment_id));
 
     // New segment records forked_from pointing at the source.
     let new_entries = store.read_all(session_id, new_segment_id).unwrap();
@@ -312,7 +351,17 @@ async fn compact_emits_session_start_carrying_summary_and_task_snapshot() {
     pod.attach_event_tx(tx);
 
     pod.run_text("first").await.unwrap();
+    let session_id = pod.session_id();
     pod.compact(10_000).await.unwrap();
+    let compacted_segment_id = pod.segment_id();
+    let metadata = pod
+        .store()
+        .read_by_name("test-pod")
+        .unwrap()
+        .expect("metadata should exist after compaction");
+    let active = metadata.active.expect("active session pointer missing");
+    assert_eq!(active.session_id, session_id);
+    assert_eq!(active.segment_id, Some(compacted_segment_id));
 
     let system_texts = system_texts_in_sink_session_start(&pod);
     // The post-compaction `SegmentStart.history` carries the new system

crates/pod/tests/controller_test.rs

@@ -417,21 +417,23 @@ async fn events_are_broadcast() {
 
 #[tokio::test]
 async fn double_run_returns_error() {
-    // Create a client that streams slowly
+    // Keep the first turn in-flight until the test drops the handle. A
+    // finite stream can finish before the second Method reaches the
+    // controller in the full test suite, making this assertion racy.
     let events = vec![
         LlmEvent::text_block_start(0),
         LlmEvent::text_delta(0, "slow..."),
-        // No stop/completed — the stream will end but without proper completion
     ];
-    let client = MockClient::new(events);
+    let client = MockClient::sequential(vec![MockResponse::Hang(events)]);
     let pod = make_pod(client).await;
     let handle = spawn_controller(pod).await;
     let mut rx = handle.subscribe();
 
-    // Send first run
+    // Send first run and wait until the controller has entered Running.
     handle.send(Method::run_text("first")).await.unwrap();
+    wait_for_status(&handle, PodStatus::Running).await;
 
-    // Immediately send second run (should get error)
+    // Now the second run must be rejected by drive_turn's live Method arm.
     handle.send(Method::run_text("second")).await.unwrap();
 
     // Look for the error event

crates/pod/tests/restore_test.rs

@@ -8,7 +8,7 @@
 use std::sync::{LazyLock, Mutex};
 
 use pod::{Pod, PodError};
-use session_store::{FsStore, StoreError};
+use session_store::{FsStore, PodActiveSegmentRef, PodMetadata, PodMetadataStore, StoreError};
 
 const MINIMAL_MANIFEST_TOML: &str = r#"
 [pod]
@@ -31,6 +31,96 @@ permission = "write"
 /// pattern used by other integration tests in this crate.
 static ENV_LOCK: LazyLock<Mutex<()>> = LazyLock::new(|| Mutex::new(()));
 
+#[tokio::test]
+async fn restore_from_pod_metadata_rejects_missing_metadata() {
+    let _lock = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+
+    let store_tmp = tempfile::tempdir().unwrap();
+    let store = FsStore::new(store_tmp.path()).unwrap();
+    let manifest = pod::PodManifest::from_toml(MINIMAL_MANIFEST_TOML).unwrap();
+
+    let result = Pod::restore_from_pod_metadata(
+        "restore-test",
+        manifest,
+        store,
+        pod::PromptLoader::builtins_only(),
+    )
+    .await;
+
+    match result {
+        Err(PodError::PodMetadataMissing { pod_name }) => assert_eq!(pod_name, "restore-test"),
+        Err(other) => panic!("expected PodMetadataMissing, got {other:?}"),
+        Ok(_) => panic!("expected missing pod metadata to fail"),
+    }
+}
+
+#[tokio::test]
+async fn restore_from_pod_metadata_rejects_pending_segment() {
+    let _lock = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+
+    let store_tmp = tempfile::tempdir().unwrap();
+    let store = FsStore::new(store_tmp.path()).unwrap();
+    let manifest = pod::PodManifest::from_toml(MINIMAL_MANIFEST_TOML).unwrap();
+    let session_id = session_store::new_session_id();
+    store
+        .write(&PodMetadata::new(
+            "restore-test",
+            Some(PodActiveSegmentRef::pending_segment(session_id)),
+        ))
+        .unwrap();
+
+    let result = Pod::restore_from_pod_metadata(
+        "restore-test",
+        manifest,
+        store,
+        pod::PromptLoader::builtins_only(),
+    )
+    .await;
+
+    match result {
+        Err(PodError::PodMetadataPending {
+            pod_name,
+            session_id: actual,
+        }) => {
+            assert_eq!(pod_name, "restore-test");
+            assert_eq!(actual, session_id);
+        }
+        Err(other) => panic!("expected PodMetadataPending, got {other:?}"),
+        Ok(_) => panic!("expected pending pod metadata to fail"),
+    }
+}
+
+#[tokio::test]
+async fn restore_from_pod_metadata_resolves_active_pointer_through_session_log() {
+    let _lock = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+
+    let store_tmp = tempfile::tempdir().unwrap();
+    let store = FsStore::new(store_tmp.path()).unwrap();
+    let manifest = pod::PodManifest::from_toml(MINIMAL_MANIFEST_TOML).unwrap();
+    let session_id = session_store::new_session_id();
+    let segment_id = session_store::new_segment_id();
+    store
+        .write(&PodMetadata::new(
+            "restore-test",
+            Some(PodActiveSegmentRef::active_segment(session_id, segment_id)),
+        ))
+        .unwrap();
+
+    let result = Pod::restore_from_pod_metadata(
+        "restore-test",
+        manifest,
+        store,
+        pod::PromptLoader::builtins_only(),
+    )
+    .await;
+
+    match result {
+        Err(PodError::Store(StoreError::NotFound(id))) => assert_eq!(id, segment_id),
+        Err(other) => panic!("expected Store(NotFound) from resolved segment, got {other:?}"),
+        Ok(_) => panic!("expected unknown resolved segment to fail"),
+    }
+}
+
 #[tokio::test]
 async fn restore_from_manifest_rejects_unknown_segment() {
     let _lock = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());