メモリPhase2の実装

crates/pod/Cargo.toml

@@ -27,6 +27,7 @@ fs4 = { version = "0.13.1", features = ["sync"] }
 libc = "0.2.185"
 schemars = "1.2.1"
 memory = { version = "0.1.0", path = "../memory" }
+uuid = { version = "1.23.1", features = ["v7"] }
 
 [dev-dependencies]
 async-trait = "0.1.89"

crates/pod/src/controller.rs

@@ -402,6 +402,14 @@ impl PodController {
                                     format!("post-run memory extract error: {e}"),
                                 );
                             }
+                            if let Err(e) = pod.try_post_run_consolidate().await {
+                                tracing::warn!(error = %e, "Post-run memory consolidate error");
+                                alerter.alert(
+                                    AlertLevel::Warn,
+                                    AlertSource::Pod,
+                                    format!("post-run memory consolidate error: {e}"),
+                                );
+                            }
                             if let Err(e) = pod.try_post_run_compact().await {
                                 tracing::warn!(error = %e, "Post-run compaction error");
                                 alerter.alert(
@@ -461,6 +469,14 @@ impl PodController {
                                     format!("post-run memory extract error: {e}"),
                                 );
                             }
+                            if let Err(e) = pod.try_post_run_consolidate().await {
+                                tracing::warn!(error = %e, "Post-run memory consolidate error");
+                                alerter.alert(
+                                    AlertLevel::Warn,
+                                    AlertSource::Pod,
+                                    format!("post-run memory consolidate error: {e}"),
+                                );
+                            }
                             if let Err(e) = pod.try_post_run_compact().await {
                                 tracing::warn!(error = %e, "Post-run compaction error");
                                 alerter.alert(
@@ -517,6 +533,14 @@ impl PodController {
                                     format!("post-run memory extract error: {e}"),
                                 );
                             }
+                            if let Err(e) = pod.try_post_run_consolidate().await {
+                                tracing::warn!(error = %e, "Post-run memory consolidate error");
+                                alerter.alert(
+                                    AlertLevel::Warn,
+                                    AlertSource::Pod,
+                                    format!("post-run memory consolidate error: {e}"),
+                                );
+                            }
                             if let Err(e) = pod.try_post_run_compact().await {
                                 tracing::warn!(error = %e, "Post-run compaction error");
                                 alerter.alert(

crates/pod/src/pod.rs

@@ -136,6 +136,11 @@ pub struct Pod<C: LlmClient, St: Store> {
     /// the flag survives across `try_post_run_extract` calls without a
     /// `&mut self` race.
     extract_in_flight: Arc<AtomicBool>,
+    /// Phase 2 (memory.consolidation) in-process reentry guard. The
+    /// staging-side `StagingLock` already provides cross-process
+    /// exclusion, but this AtomicBool keeps a careless concurrent caller
+    /// inside the same Pod from racing on the staging snapshot.
+    consolidation_in_flight: Arc<AtomicBool>,
     /// Last completed Phase 1 boundary. `None` means no extract has
     /// run yet on this session — next extract starts from entry 0.
     /// Restored from `RestoredState.extensions` on `restore`, updated
@@ -197,6 +202,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
             prompts,
             inject_resident_knowledge: true,
             extract_in_flight: Arc::new(AtomicBool::new(false)),
+            consolidation_in_flight: Arc::new(AtomicBool::new(false)),
             extract_pointer: Mutex::new(None),
             user_segments: Vec::new(),
         };
@@ -1490,6 +1496,173 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
 
         Ok(ExtractDecision::Completed)
     }
+
+    /// Build the LlmClient for the Phase 2 (memory.consolidation) Worker.
+    ///
+    /// Uses `memory.consolidation_model` from manifest if set, otherwise
+    /// clones the main client. Mirrors [`build_extractor_client`].
+    fn build_consolidator_client(
+        &self,
+        memory_cfg: &manifest::MemoryConfig,
+    ) -> Result<Box<dyn LlmClient>, PodError> {
+        if let Some(ref m) = memory_cfg.consolidation_model {
+            let client = provider::build_client(m)?;
+            return Ok(client);
+        }
+        let worker = self.worker.as_ref().expect("worker taken during run");
+        Ok(worker.client().clone_boxed())
+    }
+
+    /// Phase 2 (memory.consolidation) post-run trigger.
+    ///
+    /// Called by the Controller **after** [`try_post_run_extract`] and
+    /// **before** [`try_post_run_compact`]: extract feeds staging, compact
+    /// rewrites history. Phase 2 must consume staging before compact
+    /// reshapes the session.
+    ///
+    /// Behaviour follows `docs/plan/memory.md` §Phase 2 / §並走防止:
+    /// the staging-side `StagingLock` enforces cross-process exclusion;
+    /// `consolidation_in_flight` keeps in-process callers honest. On
+    /// success, the lock is released *with* consumed-id cleanup; on
+    /// worker failure, only the lock file is unlinked so the staging
+    /// entries remain for a future retry.
+    pub async fn try_post_run_consolidate(&mut self) -> Result<(), PodError> {
+        let Some(memory_cfg) = self.manifest.memory.clone() else {
+            return Ok(());
+        };
+        let files_threshold = memory_cfg.consolidation_threshold_files;
+        let bytes_threshold = memory_cfg.consolidation_threshold_bytes;
+        if files_threshold.is_none() && bytes_threshold.is_none() {
+            return Ok(());
+        }
+
+        loop {
+            if self
+                .consolidation_in_flight
+                .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
+                .is_err()
+            {
+                return Ok(());
+            }
+            let result = self
+                .run_consolidate_once(&memory_cfg, files_threshold, bytes_threshold)
+                .await;
+            self.consolidation_in_flight.store(false, Ordering::Release);
+
+            match result {
+                Ok(ConsolidateDecision::Skipped) => return Ok(()),
+                Ok(ConsolidateDecision::Completed) => continue,
+                Err(e) => {
+                    tracing::warn!(error = %e, "Phase 2 consolidation failed");
+                    self.alert(
+                        AlertLevel::Warn,
+                        AlertSource::Pod,
+                        format!("memory Phase 2 consolidation failed: {e}"),
+                    );
+                    return Ok(());
+                }
+            }
+        }
+    }
+
+    /// Single consolidation iteration: snapshot staging, decide whether to
+    /// fire, run the worker if so, release the lock and clean up consumed
+    /// IDs.
+    async fn run_consolidate_once(
+        &mut self,
+        memory_cfg: &manifest::MemoryConfig,
+        files_threshold: Option<usize>,
+        bytes_threshold: Option<u64>,
+    ) -> Result<ConsolidateDecision, PodError> {
+        use memory::consolidate;
+
+        let layout = memory::WorkspaceLayout::resolve(memory_cfg, &self.pwd);
+
+        let entries = consolidate::list_staging_entries(&layout);
+        if entries.is_empty() {
+            return Ok(ConsolidateDecision::Skipped);
+        }
+
+        let total_files = entries.len();
+        let total_bytes: u64 = entries.iter().map(|e| e.bytes).sum();
+        let files_hit = files_threshold.is_some_and(|n| total_files >= n);
+        let bytes_hit = bytes_threshold.is_some_and(|n| total_bytes >= n);
+        if !files_hit && !bytes_hit {
+            return Ok(ConsolidateDecision::Skipped);
+        }
+
+        let consumed_ids: Vec<uuid::Uuid> = entries.iter().map(|e| e.id).collect();
+        let lock = match consolidate::StagingLock::acquire(
+            &layout,
+            std::process::id(),
+            self.manifest.pod.name.clone(),
+            consumed_ids,
+        ) {
+            Ok(l) => l,
+            Err(memory::consolidate::LockError::InUse { .. }) => {
+                return Ok(ConsolidateDecision::Skipped);
+            }
+            Err(e) => return Err(PodError::ConsolidationLock(e)),
+        };
+
+        let cap = memory_cfg
+            .consolidation_worker_max_input_tokens
+            .unwrap_or(manifest::defaults::MEMORY_CONSOLIDATION_WORKER_MAX_INPUT_TOKENS);
+        let client = match self.build_consolidator_client(memory_cfg) {
+            Ok(c) => c,
+            Err(e) => {
+                lock.release_only();
+                return Err(e);
+            }
+        };
+        let mut worker =
+            Worker::new(client).system_prompt(consolidate::CONSOLIDATION_SYSTEM_PROMPT);
+
+        let input_so_far = Arc::new(std::sync::atomic::AtomicU64::new(0));
+        {
+            let acc = input_so_far.clone();
+            worker.on_usage(move |event| {
+                if let Some(tokens) = event.input_tokens {
+                    acc.fetch_add(tokens, Ordering::Relaxed);
+                }
+            });
+        }
+        worker.set_interceptor(MemoryConsolidationWorkerInterceptor {
+            input_so_far: input_so_far.clone(),
+            max_input_tokens: cap,
+        });
+
+        // Memory tools are self-contained — they bypass ScopedFs and write
+        // directly under the workspace via WorkspaceLayout. Resident
+        // knowledge injection (`Pod::set_resident_knowledge_injection`) is
+        // a Pod-level concern; this disposable Worker is built without it
+        // by construction, in keeping with `docs/plan/memory.md` §Phase 2
+        // のKnowledgeアクセス (agent pulls knowledge through the search
+        // tool instead of via system-prompt residency).
+        let query_cfg = memory::tool::QueryConfig::from(memory_cfg);
+        worker.register_tool(memory::tool::read_tool(layout.clone()));
+        worker.register_tool(memory::tool::write_tool(layout.clone()));
+        worker.register_tool(memory::tool::edit_tool(layout.clone()));
+        worker.register_tool(memory::tool::memory_query_tool(layout.clone(), query_cfg));
+        worker.register_tool(memory::tool::knowledge_query_tool(layout.clone(), query_cfg));
+
+        let tidy = consolidate::collect_tidy_hints(&layout);
+        let candidates = consolidate::KnowledgeCandidateReport::empty();
+        let input_text =
+            consolidate::build_consolidate_input(&layout, &entries, &tidy, &candidates);
+
+        let run_result = worker.run(input_text).await;
+        match run_result {
+            Ok(_) => {
+                lock.release_with_cleanup(&layout);
+                Ok(ConsolidateDecision::Completed)
+            }
+            Err(e) => {
+                lock.release_only();
+                Err(PodError::Worker(e))
+            }
+        }
+    }
 }
 
 /// Outcome of a single Phase 1 extract iteration. Internal to
@@ -1526,6 +1699,40 @@ impl llm_worker::interceptor::Interceptor for MemoryExtractWorkerInterceptor {
     }
 }
 
+/// Outcome of a single Phase 2 consolidation iteration. Internal to
+/// `try_post_run_consolidate` / `run_consolidate_once`.
+enum ConsolidateDecision {
+    /// Either threshold not met, no staging, or another Pod holds the lock.
+    Skipped,
+    /// Consolidation ran. Caller re-evaluates threshold against any
+    /// staging entries that arrived during the run (Coalesce).
+    Completed,
+}
+
+/// Pre-request interceptor for the Phase 2 consolidation worker. Same
+/// shape as the extract interceptor; kept separate so the abort message
+/// names the right subsystem.
+struct MemoryConsolidationWorkerInterceptor {
+    input_so_far: Arc<std::sync::atomic::AtomicU64>,
+    max_input_tokens: u64,
+}
+
+#[async_trait]
+impl llm_worker::interceptor::Interceptor for MemoryConsolidationWorkerInterceptor {
+    async fn pre_llm_request(
+        &self,
+        _context: &mut Vec<Item>,
+    ) -> llm_worker::interceptor::PreRequestAction {
+        if self.input_so_far.load(Ordering::Relaxed) > self.max_input_tokens {
+            return llm_worker::interceptor::PreRequestAction::Cancel(format!(
+                "Phase 2 consolidation worker input exceeded {} tokens",
+                self.max_input_tokens
+            ));
+        }
+        llm_worker::interceptor::PreRequestAction::Continue
+    }
+}
+
 impl<St: Store> Pod<Box<dyn LlmClient>, St> {
     /// Create a Pod entirely from a validated manifest.
     ///
@@ -1596,6 +1803,7 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
             prompts: common.prompts,
             inject_resident_knowledge: true,
             extract_in_flight: Arc::new(AtomicBool::new(false)),
+            consolidation_in_flight: Arc::new(AtomicBool::new(false)),
             extract_pointer: Mutex::new(None),
             user_segments: Vec::new(),
         };
@@ -1653,6 +1861,7 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
             prompts: common.prompts,
             inject_resident_knowledge: true,
             extract_in_flight: Arc::new(AtomicBool::new(false)),
+            consolidation_in_flight: Arc::new(AtomicBool::new(false)),
             extract_pointer: Mutex::new(None),
             user_segments: Vec::new(),
         };
@@ -1762,6 +1971,7 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
             prompts: common.prompts,
             inject_resident_knowledge: true,
             extract_in_flight: Arc::new(AtomicBool::new(false)),
+            consolidation_in_flight: Arc::new(AtomicBool::new(false)),
             extract_pointer: Mutex::new(extract_pointer),
             user_segments: state.user_segments,
         };
@@ -1963,6 +2173,9 @@ pub enum PodError {
     #[error("memory Phase 1 staging write failed: {0}")]
     ExtractStaging(#[source] memory::extract::StagingError),
 
+    #[error("memory Phase 2 lock acquisition failed: {0}")]
+    ConsolidationLock(#[source] memory::consolidate::LockError),
+
     #[error("session {session_id} has no entries to restore")]
     SessionEmpty { session_id: SessionId },
 }

crates/pod/tests/consolidation_test.rs

@@ -0,0 +1,277 @@
+//! Phase 2 (memory.consolidation) post-run trigger.
+//!
+//! Covers the gating, lock and cleanup behaviour without exercising the
+//! full sub-worker tool loop:
+//!
+//! - no `[memory]` section → no-op
+//! - `[memory]` present but no thresholds → no-op
+//! - staging empty → skip
+//! - staging below thresholds → skip + lock not acquired
+//! - staging above threshold → sub-worker runs, consumed entries removed
+//! - existing live lock → skip without error
+//!
+//! The sub-worker is fed a no-op LLM response (plain text) so it returns
+//! immediately. The post-run path then exercises lock acquisition,
+//! cleanup, and the empty-payload fast path.
+
+use std::pin::Pin;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+use async_trait::async_trait;
+use futures::Stream;
+use llm_worker::Worker;
+use llm_worker::llm_client::event::{Event as LlmEvent, ResponseStatus, StatusEvent};
+use llm_worker::llm_client::{ClientError, LlmClient, Request};
+use memory::WorkspaceLayout;
+use memory::extract::{ExtractedPayload, write_staging};
+use memory::schema::SourceRef;
+use session_store::FsStore;
+
+use pod::Pod;
+
+#[derive(Clone)]
+struct MockClient {
+    responses: Arc<Vec<Vec<LlmEvent>>>,
+    call_count: Arc<AtomicUsize>,
+}
+
+impl MockClient {
+    fn new(responses: Vec<Vec<LlmEvent>>) -> Self {
+        Self {
+            responses: Arc::new(responses),
+            call_count: Arc::new(AtomicUsize::new(0)),
+        }
+    }
+}
+
+#[async_trait]
+impl LlmClient for MockClient {
+    fn clone_boxed(&self) -> Box<dyn LlmClient> {
+        Box::new(self.clone())
+    }
+
+    async fn stream(
+        &self,
+        _request: Request,
+    ) -> Result<Pin<Box<dyn Stream<Item = Result<LlmEvent, ClientError>> + Send>>, ClientError>
+    {
+        let count = self.call_count.fetch_add(1, Ordering::SeqCst);
+        if count >= self.responses.len() {
+            return Err(ClientError::Config("mock client exhausted".into()));
+        }
+        let events = self.responses[count].clone();
+        let stream = futures::stream::iter(events.into_iter().map(Ok));
+        Ok(Box::pin(stream))
+    }
+}
+
+fn done(text: &str) -> Vec<LlmEvent> {
+    vec![
+        LlmEvent::text_block_start(0),
+        LlmEvent::text_delta(0, text),
+        LlmEvent::text_block_stop(0, None),
+        LlmEvent::Status(StatusEvent {
+            status: ResponseStatus::Completed,
+        }),
+    ]
+}
+
+const NO_MEMORY_TOML: &str = r#"
+[pod]
+name = "test-pod"
+
+[model]
+scheme = "anthropic"
+model_id = "test-model"
+
+[worker]
+max_tokens = 100
+
+[[scope.allow]]
+target = "./"
+permission = "write"
+"#;
+
+const MEMORY_NO_THRESHOLDS_TOML: &str = r#"
+[pod]
+name = "test-pod"
+
+[model]
+scheme = "anthropic"
+model_id = "test-model"
+
+[worker]
+max_tokens = 100
+
+[memory]
+
+[[scope.allow]]
+target = "./"
+permission = "write"
+"#;
+
+const FILES_THRESHOLD_TOML: &str = r#"
+[pod]
+name = "test-pod"
+
+[model]
+scheme = "anthropic"
+model_id = "test-model"
+
+[worker]
+max_tokens = 100
+
+[memory]
+consolidation_threshold_files = 2
+
+[[scope.allow]]
+target = "./"
+permission = "write"
+"#;
+
+async fn make_pod_with(
+    manifest_toml: &str,
+    pwd: std::path::PathBuf,
+    client: MockClient,
+) -> Pod<MockClient, FsStore> {
+    let manifest = pod::PodManifest::from_toml(manifest_toml).unwrap();
+
+    let store_tmp = tempfile::tempdir().unwrap();
+    let store = FsStore::new(store_tmp.path()).await.unwrap();
+    std::mem::forget(store_tmp);
+
+    let scope = pod::Scope::writable(&pwd).unwrap();
+    let worker = Worker::new(client);
+    Pod::new(manifest, worker, store, pwd, scope).await.unwrap()
+}
+
+fn write_n_staging(layout: &WorkspaceLayout, n: usize) -> Vec<uuid::Uuid> {
+    let mut ids = Vec::new();
+    for i in 0..n {
+        let (id, _) = write_staging(
+            layout,
+            SourceRef {
+                session_id: format!("s-{i}"),
+                range: [i as u64, i as u64],
+            },
+            ExtractedPayload::default(),
+        )
+        .unwrap();
+        ids.push(id);
+    }
+    ids
+}
+
+#[tokio::test]
+async fn no_memory_section_is_a_noop() {
+    let pwd = tempfile::tempdir().unwrap();
+    let client = MockClient::new(vec![]);
+    let mut pod = make_pod_with(NO_MEMORY_TOML, pwd.path().to_path_buf(), client).await;
+    pod.try_post_run_consolidate()
+        .await
+        .expect("missing memory section must skip cleanly");
+}
+
+#[tokio::test]
+async fn no_thresholds_is_a_noop() {
+    let pwd = tempfile::tempdir().unwrap();
+    let layout = WorkspaceLayout::new(pwd.path().to_path_buf());
+    write_n_staging(&layout, 5);
+
+    let client = MockClient::new(vec![]);
+    let mut pod = make_pod_with(MEMORY_NO_THRESHOLDS_TOML, pwd.path().to_path_buf(), client).await;
+    pod.try_post_run_consolidate()
+        .await
+        .expect("phase 2 disabled when both thresholds are None");
+
+    // No staging entries removed.
+    assert_eq!(
+        memory::consolidate::list_staging_entries(&layout).len(),
+        5
+    );
+}
+
+#[tokio::test]
+async fn empty_staging_skips() {
+    let pwd = tempfile::tempdir().unwrap();
+    let client = MockClient::new(vec![]);
+    let mut pod = make_pod_with(FILES_THRESHOLD_TOML, pwd.path().to_path_buf(), client).await;
+    pod.try_post_run_consolidate().await.unwrap();
+    // No mock calls expected.
+}
+
+#[tokio::test]
+async fn below_threshold_skips_and_does_not_take_lock() {
+    let pwd = tempfile::tempdir().unwrap();
+    let layout = WorkspaceLayout::new(pwd.path().to_path_buf());
+    write_n_staging(&layout, 1); // threshold is 2
+
+    let client = MockClient::new(vec![]);
+    let mut pod = make_pod_with(FILES_THRESHOLD_TOML, pwd.path().to_path_buf(), client).await;
+    pod.try_post_run_consolidate().await.unwrap();
+
+    // Staging untouched.
+    assert_eq!(
+        memory::consolidate::list_staging_entries(&layout).len(),
+        1
+    );
+    // Lock file must not exist.
+    let lock_path = layout.staging_dir().join(".consolidation.lock");
+    assert!(!lock_path.exists(), "lock file should not be created");
+}
+
+#[tokio::test]
+async fn fires_on_threshold_and_cleans_up_consumed_entries() {
+    let pwd = tempfile::tempdir().unwrap();
+    let layout = WorkspaceLayout::new(pwd.path().to_path_buf());
+    write_n_staging(&layout, 2); // threshold is 2 — fires.
+
+    // Sub-worker is given a single text-only response. The Phase 2 prompt
+    // tells it to call memory tools; the mock skips those, but `Worker::run`
+    // returns Ok regardless once the LLM closes with a final text.
+    let client = MockClient::new(vec![done("ok")]);
+    let mut pod = make_pod_with(FILES_THRESHOLD_TOML, pwd.path().to_path_buf(), client).await;
+    pod.try_post_run_consolidate().await.unwrap();
+
+    // Consumed entries removed.
+    assert!(
+        memory::consolidate::list_staging_entries(&layout).is_empty(),
+        "consumed staging entries must be cleaned up"
+    );
+    // Lock removed too.
+    let lock_path = layout.staging_dir().join(".consolidation.lock");
+    assert!(
+        !lock_path.exists(),
+        "lock file must be removed on success"
+    );
+}
+
+#[tokio::test]
+async fn live_lock_held_by_other_pod_skips() {
+    let pwd = tempfile::tempdir().unwrap();
+    let layout = WorkspaceLayout::new(pwd.path().to_path_buf());
+    write_n_staging(&layout, 3);
+
+    // Pre-acquire lock with this test's PID — definitely alive — and
+    // *don't* release it. The Phase 2 path must skip without error.
+    let _live_lock = memory::consolidate::StagingLock::acquire(
+        &layout,
+        std::process::id(),
+        "other-pod",
+        Vec::new(),
+    )
+    .unwrap();
+
+    let client = MockClient::new(vec![]);
+    let mut pod = make_pod_with(FILES_THRESHOLD_TOML, pwd.path().to_path_buf(), client).await;
+    pod.try_post_run_consolidate()
+        .await
+        .expect("InUse lock must surface as graceful skip");
+
+    // Staging untouched: lock holder owns the snapshot, not us.
+    assert_eq!(
+        memory::consolidate::list_staging_entries(&layout).len(),
+        3
+    );
+}