From a44673fa6ba9bf8ff1bc97e19a7e076eeefa5a15 Mon Sep 17 00:00:00 2001 From: Hare Date: Sat, 25 Jul 2026 09:17:08 +0900 Subject: [PATCH] memory: consolidate staging through queue tools --- crates/manifest/src/profile.rs | 5 + crates/memory/src/backend.rs | 279 +++++++++ crates/memory/src/consolidate/input.rs | 356 ----------- crates/memory/src/consolidate/lock.rs | 321 ---------- crates/memory/src/consolidate/mod.rs | 30 +- crates/memory/src/consolidate/tidy.rs | 335 ----------- crates/worker/src/controller.rs | 16 +- crates/worker/src/feature/builtin/memory.rs | 111 +++- crates/worker/src/worker.rs | 97 +-- crates/worker/tests/consolidation_test.rs | 551 ------------------ crates/workspace-server/src/server.rs | 203 ++++++- resources/profiles/memory-consolidation.dcdl | 9 + .../internal/memory_consolidation_system.md | 48 +- 13 files changed, 701 insertions(+), 1660 deletions(-) delete mode 100644 crates/memory/src/consolidate/input.rs delete mode 100644 crates/memory/src/consolidate/lock.rs delete mode 100644 crates/memory/src/consolidate/tidy.rs delete mode 100644 crates/worker/tests/consolidation_test.rs create mode 100644 resources/profiles/memory-consolidation.dcdl diff --git a/crates/manifest/src/profile.rs b/crates/manifest/src/profile.rs index 139f1ee3..1f259ac4 100644 --- a/crates/manifest/src/profile.rs +++ b/crates/manifest/src/profile.rs @@ -62,6 +62,11 @@ const BUILTIN_PROFILES: &[BuiltinProfile] = &[ label: "builtin:reviewer", description: "Bundled Reviewer role profile", }, + BuiltinProfile { + name: "memory-consolidation", + label: "builtin:memory-consolidation", + description: "Bundled Memory staging consolidation profile", + }, ]; #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] diff --git a/crates/memory/src/backend.rs b/crates/memory/src/backend.rs index 45c17633..5d092b80 100644 --- a/crates/memory/src/backend.rs +++ b/crates/memory/src/backend.rs @@ -7,11 +7,16 @@ use std::fs; use std::io; +use std::io::Write; use std::path::PathBuf; +use chrono::Utc; +use schemars::JsonSchema; use serde::{Deserialize, Serialize}; +use uuid::Uuid; use crate::audit::{AuditEvent, append_audit_event}; +use crate::consolidate::list_staging_entries_snapshot; use crate::extract::{ ExtractedCandidate, ExtractedPayload, StagingEvidence, write_staging, write_staging_candidate, }; @@ -31,6 +36,9 @@ pub enum MemoryBackendOperation { AppendAudit(MemoryAppendAuditOperation), StageCandidate(MemoryStageCandidateOperation), StageExtracted(MemoryStageExtractedOperation), + StagingList(MemoryStagingListOperation), + StagingRead(MemoryStagingReadOperation), + StagingClose(MemoryStagingCloseOperation), } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -127,6 +135,84 @@ pub struct MemoryStageExtractedOperation { pub payload: ExtractedPayload, } +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +pub struct MemoryStagingListOperation { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub limit: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +pub struct MemoryStagingReadOperation { + pub candidate_id: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +pub struct MemoryStagingCloseOperation { + pub candidate_id: String, + pub action: MemoryStagingCloseAction, + pub reason: String, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub affected_memory: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(rename_all = "snake_case")] +pub enum MemoryStagingCloseAction { + Applied, + Discarded, + Invalid, + Duplicate, + AlreadyCovered, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +pub struct MemoryStagingAffectedMemory { + pub kind: MemoryToolKind, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub slug: Option, + pub operation: MemoryStagingAffectedMemoryOperation, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(rename_all = "snake_case")] +pub enum MemoryStagingAffectedMemoryOperation { + Read, + Write, + Edit, + Delete, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MemoryConsolidateStagingOperation { + #[serde(default)] + pub force: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub threshold_files: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub threshold_bytes: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MemoryConsolidationOutput { + pub status: String, + pub summary: String, + pub candidate_count: usize, + pub total_bytes: u64, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct MemoryStagingCloseDispositionRecord { + schema_version: u32, + candidate_id: String, + staging_path: String, + recorded_at: String, + action: MemoryStagingCloseAction, + reason: String, + affected_memory: Vec, +} + +const STAGING_RESOLUTIONS_FILE: &str = "_resolutions.jsonl"; + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct MemoryBackendAckOutput { pub summary: String, @@ -194,6 +280,15 @@ pub fn execute_memory_backend_operation( }, )) } + MemoryBackendOperation::StagingList(operation) => { + execute_staging_list(layout, operation).map(MemoryBackendOperationResult::ToolOutput) + } + MemoryBackendOperation::StagingRead(operation) => { + execute_staging_read(layout, operation).map(MemoryBackendOperationResult::ToolOutput) + } + MemoryBackendOperation::StagingClose(operation) => { + execute_staging_close(layout, operation).map(MemoryBackendOperationResult::ToolOutput) + } } } @@ -391,6 +486,109 @@ fn execute_delete( }) } +fn execute_staging_list( + layout: &WorkspaceLayout, + operation: MemoryStagingListOperation, +) -> io::Result { + let limit = operation.limit.unwrap_or(20).min(100); + let snapshot = list_staging_entries_snapshot(layout); + let total = snapshot.entries.len(); + let invalid_count = snapshot.invalid_count; + let records = snapshot + .entries + .into_iter() + .take(limit) + .map(|entry| { + serde_json::json!({ + "candidate_id": entry.id.to_string(), + "bytes": entry.bytes, + "path": entry.path.display().to_string(), + "source": entry.record.source, + "kind": entry.record.kind, + "claim": entry.record.claim, + "why_useful": entry.record.why_useful, + "staleness": entry.record.staleness, + "evidence_count": entry.record.evidence.len(), + "source_ref_count": entry.record.source_refs.len(), + }) + }) + .collect::>(); + Ok(MemoryToolOutput { + summary: format!( + "Listed {} of {total} staging candidate(s); invalid_count={invalid_count}", + records.len() + ), + content: Some(serde_json::to_string_pretty(&records).map_err(io::Error::other)?), + }) +} + +fn execute_staging_read( + layout: &WorkspaceLayout, + operation: MemoryStagingReadOperation, +) -> io::Result { + let entry = find_staging_entry(layout, &operation.candidate_id)?; + Ok(MemoryToolOutput { + summary: format!("Read staging candidate {}", entry.id), + content: Some(serde_json::to_string_pretty(&entry.record).map_err(io::Error::other)?), + }) +} + +fn execute_staging_close( + layout: &WorkspaceLayout, + operation: MemoryStagingCloseOperation, +) -> io::Result { + if operation.reason.trim().is_empty() { + return Err(invalid_input("reason is required")); + } + validate_affected_memory(&operation.affected_memory)?; + let entry = find_staging_entry(layout, &operation.candidate_id)?; + let disposition = MemoryStagingCloseDispositionRecord { + schema_version: 1, + candidate_id: entry.id.to_string(), + staging_path: entry.path.display().to_string(), + recorded_at: Utc::now().to_rfc3339(), + action: operation.action, + reason: operation.reason, + affected_memory: operation.affected_memory, + }; + let resolutions_path = layout.memory_dir().join(STAGING_RESOLUTIONS_FILE); + if let Some(parent) = resolutions_path.parent() { + fs::create_dir_all(parent)?; + } + let mut line = serde_json::to_string(&disposition).map_err(io::Error::other)?; + line.push('\n'); + fs::OpenOptions::new() + .create(true) + .append(true) + .open(&resolutions_path)? + .write_all(line.as_bytes())?; + fs::remove_file(&entry.path)?; + Ok(MemoryToolOutput { + summary: format!("Closed staging candidate {}", entry.id), + content: Some(serde_json::to_string_pretty(&disposition).map_err(io::Error::other)?), + }) +} + +fn find_staging_entry( + layout: &WorkspaceLayout, + candidate_id: &str, +) -> io::Result { + let candidate_id = Uuid::parse_str(candidate_id) + .map_err(|err| invalid_input(format!("invalid candidate_id: {err}")))?; + list_staging_entries_snapshot(layout) + .entries + .into_iter() + .find(|entry| entry.id == candidate_id) + .ok_or_else(|| invalid_input(format!("staging candidate not found: {candidate_id}"))) +} + +fn validate_affected_memory(records: &[MemoryStagingAffectedMemory]) -> io::Result<()> { + for record in records { + validate_slug_rules(record.kind, record.slug.as_deref())?; + } + Ok(()) +} + fn memory_path( layout: &WorkspaceLayout, kind: MemoryToolKind, @@ -538,3 +736,84 @@ fn tool_output_from_string(value: String) -> MemoryToolOutput { fn invalid_input(message: impl Into) -> io::Error { io::Error::new(io::ErrorKind::InvalidInput, message.into()) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::extract::{CandidateKind, ExtractedCandidate}; + + #[test] + fn staging_list_read_close_records_reason_and_deletes_candidate() { + let temp = tempfile::tempdir().unwrap(); + let layout = WorkspaceLayout::resolve(&manifest::MemoryConfig::default(), temp.path()); + let source = SourceRef { + segment_id: "segment-1".into(), + range: [0, 1], + }; + let payload = ExtractedPayload { + candidates: vec![ExtractedCandidate { + kind: CandidateKind::Preference, + claim: "User prefers short reviews".into(), + why_useful: "Review style preference".into(), + staleness: None, + evidence_ids: Vec::new(), + }], + }; + let result = execute_memory_backend_operation( + &layout, + MemoryBackendOperation::StageExtracted(MemoryStageExtractedOperation { + source, + payload, + }), + ) + .unwrap(); + let candidate_id = match result { + MemoryBackendOperationResult::StagingWritten(output) => output.staging_ids[0].clone(), + _ => panic!("expected staging write output"), + }; + + let list = execute_memory_backend_operation( + &layout, + MemoryBackendOperation::StagingList(MemoryStagingListOperation { limit: Some(10) }), + ) + .unwrap(); + let MemoryBackendOperationResult::ToolOutput(list) = list else { + panic!("expected list tool output") + }; + assert!(list.content.unwrap().contains(&candidate_id)); + + let read = execute_memory_backend_operation( + &layout, + MemoryBackendOperation::StagingRead(MemoryStagingReadOperation { + candidate_id: candidate_id.clone(), + }), + ) + .unwrap(); + let MemoryBackendOperationResult::ToolOutput(read) = read else { + panic!("expected read tool output") + }; + assert!(read.content.unwrap().contains("short reviews")); + + execute_memory_backend_operation( + &layout, + MemoryBackendOperation::StagingClose(MemoryStagingCloseOperation { + candidate_id: candidate_id.clone(), + action: MemoryStagingCloseAction::Applied, + reason: "Merged into durable request memory.".into(), + affected_memory: vec![MemoryStagingAffectedMemory { + kind: MemoryToolKind::Request, + slug: Some("review-preferences".into()), + operation: MemoryStagingAffectedMemoryOperation::Edit, + }], + }), + ) + .unwrap(); + + let snapshot = list_staging_entries_snapshot(&layout); + assert!(snapshot.entries.is_empty()); + let resolutions = + fs::read_to_string(layout.memory_dir().join(STAGING_RESOLUTIONS_FILE)).unwrap(); + assert!(resolutions.contains(&candidate_id)); + assert!(resolutions.contains("Merged into durable request memory.")); + } +} diff --git a/crates/memory/src/consolidate/input.rs b/crates/memory/src/consolidate/input.rs deleted file mode 100644 index a6a2fe9c..00000000 --- a/crates/memory/src/consolidate/input.rs +++ /dev/null @@ -1,356 +0,0 @@ -//! consolidation sub-Engine への最初のユーザー入力を組み立てる。 -//! -//! extract (`extract::build_extract_input`) と同じ方針で、固定 schema の -//! markdown セクション列にしてサブEngine に渡す。`docs/plan/memory.md` -//! §Consolidation 入力 / §整理材料 の項目に従い: -//! -//! 1. consumed staging エントリ全文（`source` 込み） -//! 2. 既存 `memory/*` 全文（summary / decisions / requests） -//! 3. Usage evidence report（明示使用回数 + resident exposure cost） -//! 4. 整理材料（Linter Warn ベース、hard protection 判定はしない） -//! - -use std::fmt::Write; - -use crate::consolidate::staging::StagingEntry; -use crate::consolidate::tidy::TidyHints; -use crate::usage::UsageReport; -use crate::workspace::{RecordKind, WorkspaceLayout}; - -/// consolidation sub-Engine の最初の user 入力。 -pub fn build_consolidate_input( - layout: &WorkspaceLayout, - staging: &[StagingEntry], - tidy: &TidyHints, - usage_report: &UsageReport, -) -> String { - let mut out = String::new(); - out.push_str( - "consolidation input. Run the integration step first \ - (fold the staging activity logs into memory), then the \ - tidy step (clean up existing records). Use the memory tools for \ - every write — direct file writes are denied by the worker scope.\n\n", - ); - - out.push_str("## Staging entries (consumed by this run)\n\n"); - out.push_str(&render_staging_records(staging)); - out.push('\n'); - - out.push_str("## Existing memory records (full content)\n\n"); - out.push_str(&render_existing_memory_records(layout)); - out.push('\n'); - - out.push_str("## Usage evidence report\n\n"); - out.push_str(&render_usage_report(usage_report)); - out.push('\n'); - - out.push_str("## Tidy hints\n\n"); - out.push_str(&render_tidy_hints(tidy)); - out.push('\n'); - - out.push_str( - "When done, end the turn with a short final assistant message describing \ - what changed.", - ); - out -} - -/// Staging エントリ群を「`### ` ヘッダ + 整形 JSON ブロック」で並べる。 -/// 空配列なら「(none)」と書く。 -pub fn render_staging_records(entries: &[StagingEntry]) -> String { - if entries.is_empty() { - return "(none)\n".to_string(); - } - let mut out = String::new(); - for entry in entries { - let _ = writeln!(&mut out, "### {}", entry.id); - let json = serde_json::to_string_pretty(&entry.record).unwrap_or_else(|_| "{}".into()); - out.push_str("```json\n"); - out.push_str(&json); - out.push_str("\n```\n\n"); - } - out -} - -/// `/.yoi/memory/{summary.md,decisions/*,requests/*}` を -/// 「`### :` ヘッダ + raw markdown ブロック」で全文渡す。 -pub fn render_existing_memory_records(layout: &WorkspaceLayout) -> String { - let mut out = String::new(); - - let summary = layout.summary_path(); - if let Ok(content) = std::fs::read_to_string(&summary) { - out.push_str("### summary\n"); - out.push_str("```markdown\n"); - out.push_str(content.trim_end_matches('\n')); - out.push_str("\n```\n\n"); - } - - push_kind_records(&mut out, layout, RecordKind::Decision); - push_kind_records(&mut out, layout, RecordKind::Request); - - if out.is_empty() { - return "(none)\n".to_string(); - } - out -} - -fn push_kind_records(out: &mut String, layout: &WorkspaceLayout, kind: RecordKind) { - let dir = match kind { - RecordKind::Decision => layout.decisions_dir(), - RecordKind::Request => layout.requests_dir(), - RecordKind::Summary => return, - }; - let entries = match std::fs::read_dir(&dir) { - Ok(it) => it, - Err(_) => return, - }; - let mut paths: Vec<(String, std::path::PathBuf)> = Vec::new(); - for entry in entries.flatten() { - let path = entry.path(); - if !path.is_file() { - continue; - } - let stem = match path.file_stem().and_then(|s| s.to_str()) { - Some(s) => s, - None => continue, - }; - if path.extension().and_then(|s| s.to_str()) != Some("md") { - continue; - } - paths.push((stem.to_string(), path)); - } - paths.sort(); - for (slug, path) in paths { - let Ok(content) = std::fs::read_to_string(&path) else { - continue; - }; - let _ = writeln!(out, "### {}:{}", kind.as_str(), slug); - out.push_str("```markdown\n"); - out.push_str(content.trim_end_matches('\n')); - out.push_str("\n```\n\n"); - } -} - -fn render_usage_report(report: &UsageReport) -> String { - if report.is_empty() { - return "(empty — no explicit memory usage events recorded yet. \ - Treat this as lack of evidence, not proof that records are unused.)\n" - .to_string(); - } - let json = serde_json::to_string_pretty(report).unwrap_or_else(|_| "{}".to_string()); - format!( - "This report is evidence only. Do not make hard tidy-protection decisions from it alone.\n\n```json\n{json}\n```\n" - ) -} - -/// Tidy hints の Markdown 描画。空ヒントなら "(none)" 1 行。 -pub fn render_tidy_hints(tidy: &TidyHints) -> String { - if tidy.is_empty() { - return "(none)\n".to_string(); - } - let mut out = String::new(); - - if !tidy.replaced_decisions.is_empty() { - out.push_str("**Replaced decisions still on disk** — collapse if the chain has settled:\n"); - for (slug, replaced_by) in &tidy.replaced_decisions { - match replaced_by { - Some(target) => { - let _ = writeln!(&mut out, "- `{slug}` → `{target}`"); - } - None => { - let _ = writeln!(&mut out, "- `{slug}` (no `replaced_by` set)"); - } - } - } - out.push('\n'); - } - - if !tidy.sources_overflow.is_empty() { - out.push_str( - "**Sources overflow** — consider trimming to the most recent entries (git log keeps the rest):\n", - ); - for s in &tidy.sources_overflow { - let _ = writeln!( - &mut out, - "- {} `{}` ({} sources)", - s.kind.as_str(), - s.slug, - s.count - ); - } - out.push('\n'); - } - - if !tidy.similar_slug_clusters.is_empty() { - out.push_str("**Similar slug clusters** — evaluate for merge / rename:\n"); - for c in &tidy.similar_slug_clusters { - let joined = c - .slugs - .iter() - .map(|s| format!("`{s}`")) - .collect::>() - .join(", "); - let _ = writeln!(&mut out, "- {}: {}", c.kind.as_str(), joined); - } - out.push('\n'); - } - - out.push_str( - "Use the Usage evidence report as soft context only; \ - require an explicit reason before deleting or heavily compressing records with recent use.\n", - ); - out -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::consolidate::tidy::{SimilarSlugCluster, SourcesOverflow}; - use crate::extract::{ - CandidateKind, ExtractedCandidate, ExtractedPayload, STAGING_SCHEMA_VERSION, - StagingEvidence, StagingRecord, write_staging, - }; - use crate::schema::{EvidenceKind, SourceEvidenceRef, SourceRef}; - use chrono::Utc; - use std::path::Path; - - fn now() -> String { - Utc::now().to_rfc3339() - } - - fn write(p: &Path, content: &str) { - if let Some(parent) = p.parent() { - std::fs::create_dir_all(parent).unwrap(); - } - std::fs::write(p, content).unwrap(); - } - - fn candidate_payload(kind: CandidateKind, claim: &str) -> ExtractedPayload { - ExtractedPayload { - candidates: vec![ExtractedCandidate { - kind, - claim: claim.into(), - why_useful: "useful for consolidation".into(), - staleness: None, - evidence_ids: Vec::new(), - }], - } - } - - #[test] - fn build_includes_all_sections_when_populated() { - let dir = tempfile::TempDir::new().unwrap(); - let layout = WorkspaceLayout::new(dir.path().to_path_buf()); - - write( - &dir.path().join(".yoi/memory/summary.md"), - &format!("---\nupdated_at: {n}\n---\nstate of the world\n", n = now()), - ); - write( - &dir.path().join(".yoi/memory/decisions/dec.md"), - &format!( - "---\ncreated_at: {n}\nupdated_at: {n}\nsources: []\nstatus: open\n---\nbody\n", - n = now() - ), - ); - let _written = write_staging( - &layout, - SourceRef { - segment_id: "s".into(), - range: [0, 1], - }, - candidate_payload(CandidateKind::Preference, "Prefer concise tickets"), - ) - .unwrap(); - let staging = crate::consolidate::staging::list_staging_entries(&layout); - let tidy = TidyHints { - replaced_decisions: [("old".to_string(), Some("new".to_string()))] - .into_iter() - .collect(), - sources_overflow: vec![SourcesOverflow { - kind: RecordKind::Decision, - slug: "dec".into(), - count: 12, - }], - similar_slug_clusters: vec![SimilarSlugCluster { - kind: RecordKind::Decision, - slugs: vec!["a".into(), "ab".into()], - }], - }; - let report = UsageReport::empty(); - - let out = build_consolidate_input(&layout, &staging, &tidy, &report); - assert!(out.contains("Staging entries")); - assert!(out.contains("Existing memory records")); - assert!(out.contains("Usage evidence report")); - assert!(out.contains("Tidy hints")); - assert!(out.contains("state of the world")); - assert!(out.contains("decision:dec")); - assert!(out.contains("Replaced decisions")); - assert!(out.contains("Sources overflow")); - assert!(out.contains("Similar slug clusters")); - assert!(out.contains("no explicit memory usage events")); - } - - #[test] - fn staging_render_preserves_entry_source_refs() { - let dir = tempfile::TempDir::new().unwrap(); - let layout = WorkspaceLayout::new(dir.path().to_path_buf()); - std::fs::create_dir_all(layout.staging_dir()).unwrap(); - let id = uuid::Uuid::now_v7(); - let record = StagingRecord { - schema_version: STAGING_SCHEMA_VERSION, - id: id.to_string(), - extract_run_id: "run-1".into(), - source: SourceRef { - segment_id: "segment-record".into(), - range: [0, 10], - }, - kind: CandidateKind::Decision, - claim: "Keep flat staging records".into(), - why_useful: "consolidation input is lossless JSON".into(), - staleness: None, - evidence: vec![StagingEvidence { - id: "ev-1".into(), - kind: EvidenceKind::new(EvidenceKind::MESSAGE), - entry_range: Some([3, 4]), - excerpt: Some("bounded excerpt".into()), - summary: Some("bounded summary".into()), - }], - source_refs: vec![SourceEvidenceRef { - session_id: Some("session-1".into()), - segment_id: Some("segment-1".into()), - entry_range: Some([3, 4]), - evidence_id: Some("ev-1".into()), - evidence_kind: Some(EvidenceKind::new(EvidenceKind::MESSAGE)), - label: Some("user message".into()), - summary: Some("bounded summary".into()), - }], - }; - std::fs::write( - layout.staging_dir().join(format!("{id}.json")), - serde_json::to_string_pretty(&record).unwrap(), - ) - .unwrap(); - let staging = crate::consolidate::staging::list_staging_entries(&layout); - - let out = render_staging_records(&staging); - - assert!(out.contains("source_refs")); - assert!(out.contains("session-1")); - assert!(out.contains("entry_range")); - assert!(out.contains("ev-1")); - assert!(out.contains("message")); - } - - #[test] - fn empty_inputs_render_placeholders() { - let dir = tempfile::TempDir::new().unwrap(); - let layout = WorkspaceLayout::new(dir.path().to_path_buf()); - let out = - build_consolidate_input(&layout, &[], &TidyHints::default(), &UsageReport::empty()); - // Both staging and tidy show "(none)"; existing memory records too. - assert!(out.contains("Staging entries")); - assert!(out.contains("(none)")); - } -} diff --git a/crates/memory/src/consolidate/lock.rs b/crates/memory/src/consolidate/lock.rs deleted file mode 100644 index 945a01df..00000000 --- a/crates/memory/src/consolidate/lock.rs +++ /dev/null @@ -1,321 +0,0 @@ -//! `_staging/.consolidation.lock` による consolidation 占有ファイル。 -//! -//! `docs/plan/memory.md` §並走防止 に従い: -//! -//! - ファイルが存在し、記録された Worker が動作している間、その Worker が排他占有 -//! - クラッシュで残った stale lock は、所有者 PID が死んでいれば次回 spawn -//! 時に上書き取得できる -//! - cleanup は consumed ID の staging エントリのみ削除し、実行中に extract -//! が追加した分は残す -//! -//! 占有判定は Linux/macOS の `kill(pid, 0)` 経由で行う（`ESRCH` で死亡判定）。 -//! Windows は対象外: Yoi は POSIX 環境を前提にしている。 - -use std::fs; -use std::path::{Path, PathBuf}; - -use chrono::{DateTime, Utc}; -use serde::{Deserialize, Serialize}; -use uuid::Uuid; - -use crate::workspace::WorkspaceLayout; - -const LOCK_FILE: &str = ".consolidation.lock"; - -/// 占有ファイルの中身。`pid` で stale 判定し、`worker_name` / `started_at` / -/// `consumed_ids` は診断とクラッシュ復旧時の参照に使う。 -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct LockRecord { - pub pid: u32, - pub worker_name: String, - pub started_at: DateTime, - /// この consolidation run が起動時スナップショットで確定した consumed staging - /// entry の UUIDv7 列。完了時はこの列のみ削除し、追加分は残す。 - pub consumed_ids: Vec, -} - -/// 占有取得 / 解放のエラー。 -#[derive(Debug, thiserror::Error)] -pub enum LockError { - /// 占有ファイルが既にあり、所有者 PID が生きているのでスキップ。 - #[error("consolidation lock held by live pid {pid} (worker {worker_name:?})")] - InUse { pid: u32, worker_name: String }, - #[error("io error at {}: {source}", .path.display())] - Io { - path: PathBuf, - #[source] - source: std::io::Error, - }, - #[error("failed to (de)serialize lock record: {0}")] - Serde(#[from] serde_json::Error), -} - -impl LockError { - fn io(path: impl Into, source: std::io::Error) -> Self { - Self::Io { - path: path.into(), - source, - } - } -} - -/// consolidation が走っている間 RAII で持つ占有ハンドル。`Drop` では何もしない — -/// 完了時の cleanup は consumed ID 列削除と一緒に行う必要があるため、明示 -/// 解放 [`StagingLock::release_with_cleanup`] を使う。明示解放しないまま -/// drop された場合は占有ファイルがそのまま残り、次回 spawn 時に PID が -/// 死んでいれば stale 上書きされる。 -#[derive(Debug)] -pub struct StagingLock { - path: PathBuf, - record: LockRecord, -} - -impl StagingLock { - pub fn record(&self) -> &LockRecord { - &self.record - } - - pub fn path(&self) -> &Path { - &self.path - } - - /// 占有取得を試みる。既に live な lock があれば - /// [`LockError::InUse`]、stale 判定なら上書き取得する。 - /// staging dir が無ければ作成する。 - pub fn acquire( - layout: &WorkspaceLayout, - pid: u32, - worker_name: impl Into, - consumed_ids: Vec, - ) -> Result { - let staging_dir = layout.staging_dir(); - fs::create_dir_all(&staging_dir).map_err(|e| LockError::io(&staging_dir, e))?; - let path = staging_dir.join(LOCK_FILE); - - if path.exists() { - let raw = fs::read_to_string(&path).map_err(|e| LockError::io(&path, e))?; - // 壊れた lock は stale とみなして上書き許可。 - if let Ok(existing) = serde_json::from_str::(&raw) { - if pid_is_alive(existing.pid) { - return Err(LockError::InUse { - pid: existing.pid, - worker_name: existing.worker_name, - }); - } - tracing::warn!( - stale_pid = existing.pid, - stale_pod = %existing.worker_name, - "consolidation stale lock detected, taking over" - ); - } else { - tracing::warn!(path = %path.display(), "consolidation lock unparseable, treating as stale"); - } - } - - let record = LockRecord { - pid, - worker_name: worker_name.into(), - started_at: Utc::now(), - consumed_ids, - }; - let json = serde_json::to_string_pretty(&record)?; - fs::write(&path, json).map_err(|e| LockError::io(&path, e))?; - Ok(Self { path, record }) - } - - /// 占有を解放しつつ consumed ID 列の staging エントリを削除する。 - /// 削除対象が見当たらない場合は黙ってスキップ（既に外部で消えていた等）。 - /// 占有ファイル自体の削除も best-effort: 失敗時は warn を出すだけで - /// エラーは伝播しない（次回 spawn 時に stale 判定で上書きされる）。 - pub fn release_with_cleanup(self, layout: &WorkspaceLayout) { - let staging_dir = layout.staging_dir(); - for id in &self.record.consumed_ids { - let target = staging_dir.join(format!("{id}.json")); - match fs::remove_file(&target) { - Ok(_) => {} - Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} - Err(e) => { - tracing::warn!( - path = %target.display(), - error = %e, - "failed to clean up consumed staging entry" - ); - } - } - } - self.unlink_lock_only(); - } - - /// 占有ファイルだけ削除し、staging エントリには触らない。consolidation - /// sub-Engine が途中で失敗した場合に使う: 入力 staging を残したまま - /// 次回再評価で再処理させる（`docs/plan/memory.md` §並走防止 の - /// 「重複作成は同一 slug update に自然収束」運用）。 - pub fn release_only(self) { - self.unlink_lock_only(); - } - - fn unlink_lock_only(&self) { - if let Err(e) = fs::remove_file(&self.path) { - if e.kind() != std::io::ErrorKind::NotFound { - tracing::warn!( - path = %self.path.display(), - error = %e, - "failed to remove consolidation lock" - ); - } - } - } -} - -#[cfg(unix)] -fn pid_is_alive(pid: u32) -> bool { - // `kill(0, 0)` and `kill(-1, 0)` are POSIX-special (process group / all - // signalable processes) and would yield false positives. Reject pids - // that don't fit a positive `pid_t` so a corrupted lock file with a - // u32::MAX-ish value is treated as stale instead of magically alive. - if pid == 0 || pid > i32::MAX as u32 { - return false; - } - // SAFETY: `kill` with sig 0 only probes whether the target pid exists - // and the caller has permission to signal it. No signal is delivered. - let rc = unsafe { libc::kill(pid as i32, 0) }; - if rc == 0 { - return true; - } - // EPERM means the process exists but we can't signal it — still alive - // for our purposes. ESRCH means it's gone. - let errno = std::io::Error::last_os_error() - .raw_os_error() - .unwrap_or(libc::EINVAL); - errno != libc::ESRCH -} - -#[cfg(not(unix))] -fn pid_is_alive(_pid: u32) -> bool { - // Unsupported platforms: assume the lock is live so we never overwrite - // someone else's claim. consolidation will skip and try again next post-run. - true -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::extract::{CandidateKind, ExtractedCandidate, ExtractedPayload, write_staging}; - use crate::schema::SourceRef; - - fn make_layout() -> (tempfile::TempDir, WorkspaceLayout) { - let dir = tempfile::TempDir::new().unwrap(); - let layout = WorkspaceLayout::new(dir.path().to_path_buf()); - std::fs::create_dir_all(layout.staging_dir()).unwrap(); - (dir, layout) - } - - fn candidate_payload(claim: &str) -> ExtractedPayload { - ExtractedPayload { - candidates: vec![ExtractedCandidate { - kind: CandidateKind::Lesson, - claim: claim.into(), - why_useful: "useful for test".into(), - staleness: None, - evidence_ids: Vec::new(), - }], - } - } - - #[test] - fn acquire_writes_lock_file() { - let (_dir, layout) = make_layout(); - let lock = StagingLock::acquire(&layout, std::process::id(), "worker", Vec::new()).unwrap(); - let path = layout.staging_dir().join(LOCK_FILE); - assert!(path.exists()); - assert_eq!(lock.record().pid, std::process::id()); - assert_eq!(lock.record().worker_name, "worker"); - } - - #[test] - fn acquire_rejects_when_live_pid_holds_lock() { - let (_dir, layout) = make_layout(); - // Use this test process's pid — it's definitely alive. - let _first = - StagingLock::acquire(&layout, std::process::id(), "worker-a", Vec::new()).unwrap(); - let err = StagingLock::acquire(&layout, std::process::id(), "worker-b", Vec::new()) - .expect_err("expected InUse"); - assert!(matches!(err, LockError::InUse { .. })); - } - - #[test] - fn acquire_overwrites_stale_lock() { - let (_dir, layout) = make_layout(); - // pid 1 is init on linux but for arbitrarily-large pids we'd need - // `kill(pid, 0)` to return ESRCH. Use u32::MAX which is guaranteed - // dead on every platform we target. - let stale = LockRecord { - pid: u32::MAX, - worker_name: "ghost".into(), - started_at: Utc::now(), - consumed_ids: Vec::new(), - }; - std::fs::write( - layout.staging_dir().join(LOCK_FILE), - serde_json::to_string_pretty(&stale).unwrap(), - ) - .unwrap(); - - let lock = StagingLock::acquire(&layout, std::process::id(), "worker", Vec::new()) - .expect("stale lock must be overwritable"); - assert_eq!(lock.record().pid, std::process::id()); - } - - #[test] - fn release_drops_consumed_entries_and_unlinks_lock() { - let (_dir, layout) = make_layout(); - let id_a = write_staging( - &layout, - SourceRef { - segment_id: "s".into(), - range: [0, 0], - }, - candidate_payload("a"), - ) - .unwrap() - .remove(0) - .id; - let id_b = write_staging( - &layout, - SourceRef { - segment_id: "s".into(), - range: [1, 1], - }, - candidate_payload("b"), - ) - .unwrap() - .remove(0) - .id; - - let lock = StagingLock::acquire(&layout, std::process::id(), "worker", vec![id_a]).unwrap(); - let lock_path = lock.path().to_path_buf(); - lock.release_with_cleanup(&layout); - - assert!(!lock_path.exists(), "lock file must be removed"); - assert!( - !layout.staging_dir().join(format!("{id_a}.json")).exists(), - "consumed entry must be deleted" - ); - assert!( - layout.staging_dir().join(format!("{id_b}.json")).exists(), - "non-consumed entry must remain" - ); - } - - #[test] - fn release_is_resilient_to_missing_consumed_entries() { - let (_dir, layout) = make_layout(); - let phantom = uuid::Uuid::now_v7(); - let lock = - StagingLock::acquire(&layout, std::process::id(), "worker", vec![phantom]).unwrap(); - let lock_path = lock.path().to_path_buf(); - // No file at /.json — release must not panic. - lock.release_with_cleanup(&layout); - assert!(!lock_path.exists()); - } -} diff --git a/crates/memory/src/consolidate/mod.rs b/crates/memory/src/consolidate/mod.rs index 3610c83e..bd658033 100644 --- a/crates/memory/src/consolidate/mod.rs +++ b/crates/memory/src/consolidate/mod.rs @@ -1,32 +1,12 @@ -//! consolidation: 統合 + 整理。 +//! Memory staging queue helpers. //! -//! extract が staging に残した活動ログを `memory/*` に -//! 統合し、続けて既存 record を `outdated | superseded | unused | noisy` -//! の観点で整理する disposable Engine を、Worker 側が組み立てるための -//! ヘルパー群を提供する。Worker は次の手順で sub-Engine を構築する: -//! -//! - [`build_consolidate_input`] を sub-Engine の最初の user 入力に -//! - memory 専用 Tool (read / write / edit) と memory 検索ツールを登録 -//! - [`StagingLock::acquire`] で並走防止 + consumed ID 確定 -//! - sub-Engine run 完了後、[`StagingLock::release_with_cleanup`] で -//! consumed ID 分の staging のみ削除し、占有ファイルを解放 -//! -//! system prompt は Worker の `PromptCatalog` -//! (`WorkerPrompt::MemoryConsolidationSystem`) で管理される。Usage report は -//! 判断材料として渡すだけで、ここでは protection の hard decision はしない -//! （`docs/plan/memory.md` §Consolidation / 整理材料）。 +//! Staging candidates are consumed by the backend-managed memory-consolidation +//! Worker through MemoryStaging tools. This module only exposes bounded staging +//! listing/read support; consolidation decisions are recorded by the backend close +//! operation before a staging file is deleted. -mod input; -mod lock; mod staging; -mod tidy; -pub use input::{ - build_consolidate_input, render_existing_memory_records, render_staging_records, - render_tidy_hints, -}; -pub use lock::{LockError, LockRecord, StagingLock}; pub use staging::{ StagingEntriesSnapshot, StagingEntry, list_staging_entries, list_staging_entries_snapshot, }; -pub use tidy::{TidyHints, collect_tidy_hints}; diff --git a/crates/memory/src/consolidate/tidy.rs b/crates/memory/src/consolidate/tidy.rs deleted file mode 100644 index dab471c0..00000000 --- a/crates/memory/src/consolidate/tidy.rs +++ /dev/null @@ -1,335 +0,0 @@ -//! 整理 step が prompt 入力に乗せる「整理材料」スキャナ。 -//! -//! `docs/plan/memory.md` §整理（GC 相当）の扱い と -//! `tickets/memory-consolidation.md` の整理材料リストに従い、 -//! メトリクス未完の現状で機械的に拾えるヒントだけを集める: -//! -//! - `replaced` chain: `status: replaced` の Decision とその `replaced_by` -//! - sources 過多: `sources` / `last_sources` 配列が閾値超過の record -//! - 類似 slug 乱立: 同 kind の slug が Levenshtein 2 以内のクラスター -//! -//! 使用頻度メトリクスベースの保護閾値情報は `tickets/memory-usage-metrics.md` -//! の成果物が出るまで空で渡る。 - -use std::collections::{BTreeMap, BTreeSet}; - -use crate::Slug; -use crate::schema::{DecisionFrontmatter, RequestFrontmatter, split_frontmatter}; -use crate::workspace::{RecordKind, WorkspaceLayout}; - -/// `sources` overflow を flag する閾値。`linter::warnings::SOURCES_OVERFLOW_THRESHOLD` -/// と同値（10）を踏襲する。Linter Warn で sources 過多が検出されるラインと -/// 整理 step で勧告するラインを揃える狙い。 -pub const SOURCES_OVERFLOW_THRESHOLD: usize = 10; -/// 類似 slug クラスタリングの距離。`linter::warnings::SIMILAR_SLUG_DISTANCE` -/// と同値。 -pub const SIMILAR_SLUG_DISTANCE: usize = 2; - -/// 整理 step 用の機械集計ヒント。空フィールドは「対象なし」を意味する。 -#[derive(Debug, Default, Clone)] -pub struct TidyHints { - /// `status: replaced` で残っている Decision の slug → `replaced_by` map。 - /// `replaced_by` が None でも置き換え滞留として列挙する。 - pub replaced_decisions: BTreeMap>, - /// kind / slug / sources count の三つ組で sources 累積ラインを表す。 - pub sources_overflow: Vec, - /// 同 kind 内で Levenshtein 距離 `<= SIMILAR_SLUG_DISTANCE` のクラスター。 - /// クラスター内の slug は sorted。 - pub similar_slug_clusters: Vec, -} - -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct SourcesOverflow { - pub kind: RecordKind, - pub slug: String, - pub count: usize, -} - -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct SimilarSlugCluster { - pub kind: RecordKind, - pub slugs: Vec, -} - -impl TidyHints { - pub fn is_empty(&self) -> bool { - self.replaced_decisions.is_empty() - && self.sources_overflow.is_empty() - && self.similar_slug_clusters.is_empty() - } -} - -/// workspace を一通りスキャンして [`TidyHints`] を組み立てる。読めない / -/// parse できない record は黙ってスキップ（Linter は write 経路で守って -/// いるので、ここで顕在化してもどうしようもない）。 -pub fn collect_tidy_hints(layout: &WorkspaceLayout) -> TidyHints { - let mut hints = TidyHints::default(); - - let decisions = read_kind_records(layout, RecordKind::Decision); - let requests = read_kind_records(layout, RecordKind::Request); - - for (slug, content) in &decisions { - let fm = parse_yaml::(content); - if let Some(fm) = fm.as_ref() { - if matches!(fm.status, crate::schema::DecisionStatus::Replaced) { - hints - .replaced_decisions - .insert(slug.clone(), fm.replaced_by.as_ref().map(|s| s.to_string())); - } - if fm.sources.len() > SOURCES_OVERFLOW_THRESHOLD { - hints.sources_overflow.push(SourcesOverflow { - kind: RecordKind::Decision, - slug: slug.clone(), - count: fm.sources.len(), - }); - } - } - } - for (slug, content) in &requests { - if let Some(fm) = parse_yaml::(content) { - if fm.sources.len() > SOURCES_OVERFLOW_THRESHOLD { - hints.sources_overflow.push(SourcesOverflow { - kind: RecordKind::Request, - slug: slug.clone(), - count: fm.sources.len(), - }); - } - } - } - hints.sources_overflow.sort_by(|a, b| { - (a.kind.as_str(), a.slug.as_str()).cmp(&(b.kind.as_str(), b.slug.as_str())) - }); - - let decision_slugs: Vec<&str> = decisions.keys().map(|s| s.as_str()).collect(); - let request_slugs: Vec<&str> = requests.keys().map(|s| s.as_str()).collect(); - if let Some(c) = cluster_similar(&decision_slugs, RecordKind::Decision) { - hints.similar_slug_clusters.extend(c); - } - if let Some(c) = cluster_similar(&request_slugs, RecordKind::Request) { - hints.similar_slug_clusters.extend(c); - } - hints - .similar_slug_clusters - .sort_by(|a, b| (a.kind.as_str(), &a.slugs).cmp(&(b.kind.as_str(), &b.slugs))); - - hints -} - -/// `/.yoi/memory//*.md` を slug ごとに `(slug, full content)` -/// 化して返す。 -fn read_kind_records(layout: &WorkspaceLayout, kind: RecordKind) -> BTreeMap { - let dir = match kind { - RecordKind::Decision => layout.decisions_dir(), - RecordKind::Request => layout.requests_dir(), - RecordKind::Summary => return BTreeMap::new(), - }; - let mut out: BTreeMap = BTreeMap::new(); - let entries = match std::fs::read_dir(&dir) { - Ok(it) => it, - Err(_) => return out, - }; - for entry in entries.flatten() { - let path = entry.path(); - if !path.is_file() { - continue; - } - let stem = match path.file_stem().and_then(|s| s.to_str()) { - Some(s) => s, - None => continue, - }; - if path.extension().and_then(|s| s.to_str()) != Some("md") { - continue; - } - if Slug::parse(stem).is_err() { - continue; - } - let content = match std::fs::read_to_string(&path) { - Ok(s) => s, - Err(_) => continue, - }; - out.insert(stem.to_string(), content); - } - out -} - -fn parse_yaml(content: &str) -> Option { - let (yaml, _body) = split_frontmatter(content).ok()?; - serde_yaml::from_str::(yaml).ok() -} - -/// Connected-component clustering over the `levenshtein <= SIMILAR_SLUG_DISTANCE` -/// graph among same-kind slugs. Returns each cluster of size >= 2 (singleton -/// clusters are not interesting for the integration step). Returns `None` -/// when there are no clusters at all. -fn cluster_similar(slugs: &[&str], kind: RecordKind) -> Option> { - if slugs.len() < 2 { - return None; - } - let n = slugs.len(); - let mut parent: Vec = (0..n).collect(); - fn find(parent: &mut [usize], i: usize) -> usize { - if parent[i] == i { - i - } else { - let root = find(parent, parent[i]); - parent[i] = root; - root - } - } - fn union(parent: &mut [usize], a: usize, b: usize) { - let ra = find(parent, a); - let rb = find(parent, b); - if ra != rb { - parent[ra] = rb; - } - } - for i in 0..n { - for j in (i + 1)..n { - if levenshtein(slugs[i], slugs[j]) <= SIMILAR_SLUG_DISTANCE { - union(&mut parent, i, j); - } - } - } - let mut groups: BTreeMap> = BTreeMap::new(); - for i in 0..n { - let root = find(&mut parent, i); - groups.entry(root).or_default().push(slugs[i].to_string()); - } - let mut out: Vec = Vec::new(); - let mut seen_canonical: BTreeSet> = BTreeSet::new(); - for (_, mut group) in groups { - if group.len() < 2 { - continue; - } - group.sort(); - if seen_canonical.insert(group.clone()) { - out.push(SimilarSlugCluster { kind, slugs: group }); - } - } - if out.is_empty() { None } else { Some(out) } -} - -/// Iterative two-row Levenshtein distance over chars (matches the Linter's -/// implementation; kept private to avoid widening that crate-internal API). -fn levenshtein(a: &str, b: &str) -> usize { - let a: Vec = a.chars().collect(); - let b: Vec = b.chars().collect(); - if a.is_empty() { - return b.len(); - } - if b.is_empty() { - return a.len(); - } - let mut prev: Vec = (0..=b.len()).collect(); - let mut curr: Vec = vec![0; b.len() + 1]; - for (i, ca) in a.iter().enumerate() { - curr[0] = i + 1; - for (j, cb) in b.iter().enumerate() { - let cost = if ca == cb { 0 } else { 1 }; - curr[j + 1] = (curr[j] + 1).min(prev[j + 1] + 1).min(prev[j] + cost); - } - std::mem::swap(&mut prev, &mut curr); - } - prev[b.len()] -} - -#[cfg(test)] -mod tests { - use super::*; - use chrono::Utc; - use std::path::Path; - - fn now() -> String { - Utc::now().to_rfc3339() - } - - fn write(p: &Path, content: &str) { - if let Some(parent) = p.parent() { - std::fs::create_dir_all(parent).unwrap(); - } - std::fs::write(p, content).unwrap(); - } - - fn workspace() -> (tempfile::TempDir, WorkspaceLayout) { - let dir = tempfile::TempDir::new().unwrap(); - let layout = WorkspaceLayout::new(dir.path().to_path_buf()); - (dir, layout) - } - - #[test] - fn collects_replaced_chain() { - let (dir, layout) = workspace(); - write( - &dir.path().join(".yoi/memory/decisions/replaced.md"), - &format!( - "---\ncreated_at: {n}\nupdated_at: {n}\nsources: []\nstatus: replaced\nreplaced_by: winner\n---\n", - n = now() - ), - ); - write( - &dir.path().join(".yoi/memory/decisions/winner.md"), - &format!( - "---\ncreated_at: {n}\nupdated_at: {n}\nsources: []\nstatus: open\n---\n", - n = now() - ), - ); - let hints = collect_tidy_hints(&layout); - assert_eq!( - hints.replaced_decisions.get("replaced").cloned(), - Some(Some("winner".into())) - ); - assert!(!hints.replaced_decisions.contains_key("winner")); - } - - #[test] - fn flags_sources_overflow() { - let (dir, layout) = workspace(); - let many_sources: String = (0..15) - .map(|i| format!(" - segment_id: s{i}\n range: [{i}, {i}]\n")) - .collect(); - write( - &dir.path().join(".yoi/memory/decisions/big.md"), - &format!( - "---\ncreated_at: {n}\nupdated_at: {n}\nstatus: open\nsources:\n{m}---\n", - n = now(), - m = many_sources - ), - ); - let hints = collect_tidy_hints(&layout); - assert_eq!(hints.sources_overflow.len(), 1); - assert_eq!(hints.sources_overflow[0].slug, "big"); - assert_eq!(hints.sources_overflow[0].kind, RecordKind::Decision); - assert_eq!(hints.sources_overflow[0].count, 15); - } - - #[test] - fn clusters_similar_slugs() { - let (dir, layout) = workspace(); - for slug in ["db-pool", "db-pol", "db-pools", "alpha"] { - write( - &dir.path().join(format!(".yoi/memory/decisions/{slug}.md")), - &format!( - "---\ncreated_at: {n}\nupdated_at: {n}\nsources: []\nstatus: open\n---\n", - n = now() - ), - ); - } - let hints = collect_tidy_hints(&layout); - assert_eq!(hints.similar_slug_clusters.len(), 1); - assert_eq!( - hints.similar_slug_clusters[0].slugs, - vec![ - "db-pol".to_string(), - "db-pool".to_string(), - "db-pools".to_string(), - ] - ); - } - - #[test] - fn empty_workspace_yields_empty_hints() { - let (_dir, layout) = workspace(); - let hints = collect_tidy_hints(&layout); - assert!(hints.is_empty()); - } -} diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 7566cbb0..5f79c8b9 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -646,10 +646,18 @@ where base_url, } = workspace_client { - for definition in crate::feature::builtin::memory::workspace_http_memory_tools( - workspace_id, - base_url, - ) { + let definitions = if spawner_name == "memory-consolidation" { + crate::feature::builtin::memory::workspace_http_memory_consolidation_tools( + workspace_id, + base_url, + ) + } else { + crate::feature::builtin::memory::workspace_http_memory_tools( + workspace_id, + base_url, + ) + }; + for definition in definitions { worker.register_tool(definition); } } else { diff --git a/crates/worker/src/feature/builtin/memory.rs b/crates/worker/src/feature/builtin/memory.rs index 9014e3bc..d32051ae 100644 --- a/crates/worker/src/feature/builtin/memory.rs +++ b/crates/worker/src/feature/builtin/memory.rs @@ -12,9 +12,11 @@ use llm_engine::tool::{ }; use memory::backend::{ MemoryBackendHttpResponse, MemoryBackendOperation, MemoryBackendOperationResult, - MemoryDeleteOperation, MemoryEditOperation, MemoryQueryOperation, MemoryReadOperation, - MemoryToolOutput, MemoryWriteOperation, + MemoryConsolidateStagingOperation, MemoryConsolidationOutput, MemoryDeleteOperation, + MemoryEditOperation, MemoryQueryOperation, MemoryReadOperation, MemoryStagingCloseOperation, + MemoryStagingListOperation, MemoryStagingReadOperation, MemoryToolOutput, MemoryWriteOperation, }; +use schemars::JsonSchema; use serde::de::DeserializeOwned; use serde_json::json; @@ -91,6 +93,28 @@ impl WorkspaceClient { } } } + + pub async fn request_memory_staging_consolidation( + &self, + operation: MemoryConsolidateStagingOperation, + ) -> Result { + match self { + WorkspaceClient::Http { + workspace_id, + base_url, + } => execute_http_memory_consolidation(workspace_id, base_url, operation).await, + WorkspaceClient::Available { kind } => Err(WorkspaceMemoryBackendError::Unavailable { + reason: format!( + "workspace client kind `{kind}` does not expose the Backend Workspace API" + ), + }), + WorkspaceClient::Unavailable { reason } => { + Err(WorkspaceMemoryBackendError::Unavailable { + reason: reason.clone(), + }) + } + } + } } async fn execute_http_memory_backend( @@ -121,6 +145,29 @@ async fn execute_http_memory_backend( } } +async fn execute_http_memory_consolidation( + workspace_id: &str, + base_url: &str, + operation: MemoryConsolidateStagingOperation, +) -> Result { + let url = format!( + "{}/api/w/{}/memory/consolidation", + base_url.trim_end_matches('/'), + workspace_id + ); + let response = reqwest::Client::new() + .post(url) + .json(&operation) + .send() + .await?; + let status = response.status(); + let body = response.text().await?; + if !status.is_success() { + return Err(WorkspaceMemoryBackendError::Http { status, body }); + } + serde_json::from_str::(&body).map_err(Into::into) +} + pub fn workspace_http_memory_tools( workspace_id: impl Into, base_url: impl Into, @@ -185,6 +232,58 @@ pub fn workspace_http_memory_tools( ] } +pub fn workspace_http_memory_consolidation_tools( + workspace_id: impl Into, + base_url: impl Into, +) -> Vec { + let workspace_id = workspace_id.into(); + let base_url = base_url.into(); + let mut tools = workspace_http_memory_tools(workspace_id.clone(), base_url.clone()); + let backend = WorkspaceHttpMemoryBackend::new(workspace_id, base_url); + tools.extend([ + memory_tool( + "MemoryStagingList", + STAGING_LIST_DESCRIPTION, + schema_for::(), + backend.clone(), + |input| { + Ok(MemoryBackendOperation::StagingList(parse_input::< + MemoryStagingListOperation, + >( + input + )?)) + }, + ), + memory_tool( + "MemoryStagingRead", + STAGING_READ_DESCRIPTION, + schema_for::(), + backend.clone(), + |input| { + Ok(MemoryBackendOperation::StagingRead(parse_input::< + MemoryStagingReadOperation, + >( + input + )?)) + }, + ), + memory_tool( + "MemoryStagingClose", + STAGING_CLOSE_DESCRIPTION, + schema_for::(), + backend, + |input| { + Ok(MemoryBackendOperation::StagingClose(parse_input::< + MemoryStagingCloseOperation, + >( + input + )?)) + }, + ), + ]); + tools +} + type OperationBuilder = fn(&str) -> Result; fn memory_tool( @@ -229,6 +328,10 @@ fn parse_input(input: &str) -> Result { serde_json::from_str(input).map_err(|error| ToolError::InvalidArgument(error.to_string())) } +fn schema_for() -> serde_json::Value { + serde_json::to_value(schemars::schema_for!(T)).expect("memory tool schema should serialize") +} + fn tool_output(output: MemoryToolOutput) -> ToolOutput { ToolOutput { summary: output.summary, @@ -243,6 +346,10 @@ const EDIT_DESCRIPTION: &str = "Replace text in a durable memory record through Workspace authority."; const DELETE_DESCRIPTION: &str = "Delete a durable memory record through Workspace authority."; const QUERY_DESCRIPTION: &str = "Query durable memory records through Workspace authority."; +const STAGING_LIST_DESCRIPTION: &str = + "List pending Memory staging candidates without loading full record payloads."; +const STAGING_READ_DESCRIPTION: &str = "Read one pending Memory staging candidate by candidate_id."; +const STAGING_CLOSE_DESCRIPTION: &str = "Close one staging candidate with a required reason; records disposition and deletes the staging record."; fn kind_schema() -> serde_json::Value { json!({"type":"string","enum":["summary","decision","request"]}) diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 3f61cfdb..3e208464 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -520,10 +520,7 @@ pub struct Worker { /// the flag survives across `try_post_run_extract` calls without a /// `&mut self` race. extract_in_flight: Arc, - /// consolidation (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 Worker from racing on the staging snapshot. + /// consolidation (memory.consolidation) in-process reentry guard. consolidation_in_flight: Arc, /// Last completed extract boundary. `None` means no extract has /// run yet on this session — next extract starts from entry 0. @@ -3274,11 +3271,10 @@ impl Worker { Ok(ExtractDecision::Completed) } - /// consolidation (memory.consolidation) trigger. + /// Request Backend-managed Memory staging consolidation after a Worker turn. /// - /// Worker no longer has direct Workspace filesystem authority. Until consolidation is - /// exposed as a Backend Workspace Authority operation, the Worker must not inspect - /// staging, acquire staging locks, or register local memory tools directly. + /// Worker has no local Workspace memory authority. It only asks the Backend + /// Workspace to notify or spawn the dedicated consolidater Worker. pub async fn try_post_run_consolidate(&mut self) -> Result<(), WorkerError> { let Some(memory_cfg) = self.manifest.memory.clone() else { return Ok(()); @@ -3289,30 +3285,64 @@ impl Worker { .unwrap_or(&self.manifest.model); let files_threshold = memory_cfg.consolidation_threshold_files.filter(|n| *n > 0); let bytes_threshold = memory_cfg.consolidation_threshold_bytes.filter(|n| *n > 0); - let reason = if files_threshold.is_none() && bytes_threshold.is_none() { - "consolidation_threshold_disabled" - } else { - "consolidation_backend_operation_unavailable" - }; - WorkerAuditBase::new( - memory::audit::AuditWorker::MemoryConsolidation, - memory::audit::AuditTrigger::StagingBacklog, - Some(model_audit_from_manifest(model)), - ) - .emit( - self.workspace_client(), - self.event_tx.as_ref(), - memory::audit::WorkerLifecycleStatus::Skipped, - reason, - None, - None, - None, - ) - .await; - if reason == "consolidation_backend_operation_unavailable" { - tracing::debug!( - "workspace memory consolidation skipped: backend operation is unavailable" - ); + if files_threshold.is_none() && bytes_threshold.is_none() { + WorkerAuditBase::new( + memory::audit::AuditWorker::MemoryConsolidation, + memory::audit::AuditTrigger::StagingBacklog, + Some(model_audit_from_manifest(model)), + ) + .emit( + self.workspace_client(), + self.event_tx.as_ref(), + memory::audit::WorkerLifecycleStatus::Skipped, + "consolidation_threshold_disabled", + None, + None, + None, + ) + .await; + return Ok(()); + } + + match self + .workspace_client() + .request_memory_staging_consolidation( + memory::backend::MemoryConsolidateStagingOperation { + force: false, + threshold_files: files_threshold, + threshold_bytes: bytes_threshold, + }, + ) + .await + { + Ok(output) => { + tracing::debug!( + status = output.status.as_str(), + summary = output.summary.as_str(), + "requested backend memory staging consolidation" + ); + } + Err(error) => { + tracing::warn!( + error = %error, + "failed to request backend memory staging consolidation" + ); + WorkerAuditBase::new( + memory::audit::AuditWorker::MemoryConsolidation, + memory::audit::AuditTrigger::StagingBacklog, + Some(model_audit_from_manifest(model)), + ) + .emit( + self.workspace_client(), + self.event_tx.as_ref(), + memory::audit::WorkerLifecycleStatus::Skipped, + "consolidation_backend_operation_failed", + None, + None, + None, + ) + .await; + } } Ok(()) } @@ -4587,9 +4617,6 @@ pub enum WorkerError { #[error("feature install failed: {0}")] FeatureInstall(String), - #[error("memory consolidation lock acquisition failed: {0}")] - ConsolidationLock(#[source] memory::consolidate::LockError), - #[error("session {segment_id} has no entries to restore")] SegmentEmpty { segment_id: SegmentId }, diff --git a/crates/worker/tests/consolidation_test.rs b/crates/worker/tests/consolidation_test.rs deleted file mode 100644 index b2722557..00000000 --- a/crates/worker/tests/consolidation_test.rs +++ /dev/null @@ -1,551 +0,0 @@ -//! consolidation (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_engine::Engine; -use llm_engine::llm_client::event::{Event as LlmEvent, ResponseStatus, StatusEvent}; -use llm_engine::llm_client::{ClientError, LlmClient, Request}; -use memory::WorkspaceLayout; -use memory::extract::{CandidateKind, ExtractedCandidate, ExtractedPayload, write_staging}; -use memory::schema::SourceRef; -use session_store::FsStore; -use session_store::{CombinedStore, FsWorkerStore}; - -type TestStore = CombinedStore; -use tokio::sync::broadcast; - -use worker::{Event, Worker}; - -#[derive(Clone)] -struct MockClient { - responses: Arc>>, - call_count: Arc, -} - -impl MockClient { - fn new(responses: Vec>) -> Self { - Self { - responses: Arc::new(responses), - call_count: Arc::new(AtomicUsize::new(0)), - } - } -} - -#[async_trait] -impl LlmClient for MockClient { - fn clone_boxed(&self) -> Box { - Box::new(self.clone()) - } - - async fn stream( - &self, - _request: Request, - ) -> Result> + 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 { - 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#" -[worker] -name = "test-worker" - -[model] -scheme = "anthropic" -model_id = "test-model" - -[engine] -max_tokens = 100 - -[[scope.allow]] -target = "./" -permission = "write" -"#; - -const MEMORY_NO_THRESHOLDS_TOML: &str = r#" -[worker] -name = "test-worker" - -[model] -scheme = "anthropic" -model_id = "test-model" - -[engine] -max_tokens = 100 - -[memory] - -[[scope.allow]] -target = "./" -permission = "write" -"#; - -const FILES_THRESHOLD_TOML: &str = r#" -[worker] -name = "test-worker" - -[model] -scheme = "anthropic" -model_id = "test-model" - -[engine] -max_tokens = 100 - -[memory] -consolidation_threshold_files = 2 - -[[scope.allow]] -target = "./" -permission = "write" -"#; - -const ZERO_THRESHOLDS_TOML: &str = r#" -[worker] -name = "test-worker" - -[model] -scheme = "anthropic" -model_id = "test-model" - -[engine] -max_tokens = 100 - -[memory] -consolidation_threshold_files = 0 -consolidation_threshold_bytes = 0 - -[[scope.allow]] -target = "./" -permission = "write" -"#; - -async fn make_worker_with( - manifest_toml: &str, - pwd: std::path::PathBuf, - client: MockClient, -) -> Worker { - let manifest = worker::WorkerManifest::from_toml(manifest_toml).unwrap(); - - let store_tmp = tempfile::tempdir().unwrap(); - let store = CombinedStore::new( - FsStore::new(store_tmp.path()).unwrap(), - FsWorkerStore::new(store_tmp.path().join("pods")).unwrap(), - ); - std::mem::forget(store_tmp); - - let scope = worker::Scope::writable(&pwd).unwrap(); - let worker = Engine::new(client); - Worker::new( - manifest, - worker, - store, - worker::WorkerWorkspaceContext::local_filesystem(None), - worker::WorkerFilesystemAuthority::local(pwd.clone(), pwd.clone()), - scope, - ) - .await - .unwrap() -} - -fn staging_payload(claim: String) -> ExtractedPayload { - ExtractedPayload { - candidates: vec![ExtractedCandidate { - kind: CandidateKind::Lesson, - claim, - why_useful: "useful for consolidation trigger tests".into(), - staleness: None, - evidence_ids: Vec::new(), - }], - } -} - -fn write_n_staging(layout: &WorkspaceLayout, n: usize) -> Vec { - let mut ids = Vec::new(); - for i in 0..n { - let id = write_staging( - layout, - SourceRef { - segment_id: format!("s-{i}"), - range: [i as u64, i as u64], - }, - staging_payload(format!("candidate-{i}")), - ) - .unwrap() - .remove(0) - .id; - ids.push(id); - } - ids -} - -fn attach_event_receiver(worker: &mut Worker) -> broadcast::Receiver { - let (tx, rx) = broadcast::channel(16); - worker.attach_event_tx(tx); - rx -} - -fn collect_memory_worker_reasons(rx: &mut broadcast::Receiver) -> Vec { - let mut reasons = Vec::new(); - loop { - match rx.try_recv() { - Ok(Event::MemoryWorker(event)) => reasons.push(event.reason), - Ok(_) => {} - Err(broadcast::error::TryRecvError::Empty) => break, - Err(err) => panic!("unexpected broadcast receive error: {err}"), - } - } - reasons -} - -fn read_audit_jsonl(layout: &WorkspaceLayout) -> Vec { - let text = std::fs::read_to_string(layout.audit_current_log_path()).unwrap(); - text.lines() - .map(|line| serde_json::from_str::(line).unwrap()) - .collect() -} - -#[tokio::test] -async fn no_memory_section_is_a_noop() { - let pwd = tempfile::tempdir().unwrap(); - let client = MockClient::new(vec![]); - let mut worker = make_worker_with(NO_MEMORY_TOML, pwd.path().to_path_buf(), client).await; - worker - .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 worker = - make_worker_with(MEMORY_NO_THRESHOLDS_TOML, pwd.path().to_path_buf(), client).await; - worker - .try_post_run_consolidate() - .await - .expect("consolidation disabled when both thresholds are None"); - - // No staging entries removed. - assert_eq!(memory::consolidate::list_staging_entries(&layout).len(), 5); -} - -#[tokio::test] -async fn zero_thresholds_treated_as_disabled() { - // Without the `Some(0) → None` collapse, `total_files >= 0` and - // `total_bytes >= 0` would always evaluate true and consolidation would - // fire on every post-run with any staging activity. - 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 worker = make_worker_with(ZERO_THRESHOLDS_TOML, pwd.path().to_path_buf(), client).await; - worker - .try_post_run_consolidate() - .await - .expect("zero thresholds must collapse to disabled, not fire on every staging entry"); - - assert_eq!( - memory::consolidate::list_staging_entries(&layout).len(), - 5, - "staging must be untouched when both thresholds are zero" - ); - let lock_path = layout.staging_dir().join(".consolidation.lock"); - assert!(!lock_path.exists(), "no lock should be acquired"); -} - -#[tokio::test] -async fn empty_staging_skips() { - let pwd = tempfile::tempdir().unwrap(); - let client = MockClient::new(vec![]); - let mut worker = make_worker_with(FILES_THRESHOLD_TOML, pwd.path().to_path_buf(), client).await; - worker.try_post_run_consolidate().await.unwrap(); - // No mock calls expected. -} - -#[tokio::test] -async fn empty_staging_skip_is_audit_only() { - let pwd = tempfile::tempdir().unwrap(); - let client = MockClient::new(vec![]); - let mut worker = make_worker_with(FILES_THRESHOLD_TOML, pwd.path().to_path_buf(), client).await; - let mut rx = attach_event_receiver(&mut worker); - - worker.try_post_run_consolidate().await.unwrap(); - - assert!(collect_memory_worker_reasons(&mut rx).is_empty()); -} - -#[tokio::test] -async fn invalid_only_staging_is_distinct_from_no_staging() { - let pwd = tempfile::tempdir().unwrap(); - let layout = WorkspaceLayout::new(pwd.path().to_path_buf()); - std::fs::create_dir_all(layout.staging_dir()).unwrap(); - let invalid_id = uuid::Uuid::now_v7(); - let invalid_path = layout.staging_dir().join(format!("{invalid_id}.json")); - std::fs::write( - &invalid_path, - serde_json::json!({ - "source": { - "session_id": "legacy-session", - "range": [0, 1] - }, - "requests": [] - }) - .to_string(), - ) - .unwrap(); - - let client = MockClient::new(vec![]); - let mut worker = make_worker_with(FILES_THRESHOLD_TOML, pwd.path().to_path_buf(), client).await; - let mut rx = attach_event_receiver(&mut worker); - - worker.try_post_run_consolidate().await.unwrap(); - - assert!(invalid_path.exists(), "invalid staging is not auto-deleted"); - let reasons = collect_memory_worker_reasons(&mut rx); - assert_eq!(reasons, vec!["no_valid_staging_entries invalid=1"]); - - let audit = read_audit_jsonl(&layout); - let last = audit.last().unwrap(); - assert_eq!(last["reason"], "no_valid_staging_entries invalid=1"); - assert_eq!(last["consolidation"]["staging_count"], 0); - assert_eq!(last["consolidation"]["invalid_staging_count"], 1); -} - -#[tokio::test] -async fn below_threshold_skip_is_audit_only() { - 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 worker = make_worker_with(FILES_THRESHOLD_TOML, pwd.path().to_path_buf(), client).await; - let mut rx = attach_event_receiver(&mut worker); - - worker.try_post_run_consolidate().await.unwrap(); - - assert!(collect_memory_worker_reasons(&mut rx).is_empty()); - let audit = read_audit_jsonl(&layout); - let reason = audit.last().unwrap()["reason"] - .as_str() - .expect("audit reason must be a string"); - assert!(reason.starts_with("threshold_not_reached ")); -} - -#[tokio::test] -async fn completed_event_survives_terminal_empty_drain_skip() { - let pwd = tempfile::tempdir().unwrap(); - let layout = WorkspaceLayout::new(pwd.path().to_path_buf()); - write_n_staging(&layout, 2); // threshold is 2 — fires. - - let client = MockClient::new(vec![done("ok")]); - let mut worker = make_worker_with(FILES_THRESHOLD_TOML, pwd.path().to_path_buf(), client).await; - let mut rx = attach_event_receiver(&mut worker); - - worker.try_post_run_consolidate().await.unwrap(); - - let reasons = collect_memory_worker_reasons(&mut rx); - assert_eq!(reasons.len(), 2); - assert!(reasons[0].starts_with("staging_threshold_reached files=2 bytes=")); - assert_eq!(reasons[1], "completed_no_record_changes"); - let audit = read_audit_jsonl(&layout); - assert_eq!(audit.last().unwrap()["reason"], "no_staging_entries"); -} - -#[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 worker = make_worker_with(FILES_THRESHOLD_TOML, pwd.path().to_path_buf(), client).await; - worker.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 consolidation prompt - // tells it to call memory tools; the mock skips those, but `Engine::run` - // returns Ok regardless once the LLM closes with a final text. - let client = MockClient::new(vec![done("ok")]); - let mut worker = make_worker_with(FILES_THRESHOLD_TOML, pwd.path().to_path_buf(), client).await; - worker.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 in_flight_guard_skips_reentry_without_clearing() { - use std::sync::atomic::Ordering; - - let pwd = tempfile::tempdir().unwrap(); - let layout = WorkspaceLayout::new(pwd.path().to_path_buf()); - write_n_staging(&layout, 2); - - let client = MockClient::new(vec![]); - let mut worker = make_worker_with( - FILES_THRESHOLD_TOML, - pwd.path().to_path_buf(), - client.clone(), - ) - .await; - - // Pre-set the in-flight flag as if another concurrent caller had - // entered run_consolidate_once. The CAS at the top of - // try_post_run_consolidate must take the early return without - // touching staging or the LLM, and must leave the flag intact for - // the holder to clear. - let in_flight = worker.consolidation_in_flight_handle(); - in_flight.store(true, Ordering::Release); - - worker.try_post_run_consolidate().await.unwrap(); - - assert!( - in_flight.load(Ordering::Acquire), - "reentry skip must not clear the in-flight flag — that's the holder's job" - ); - assert_eq!( - memory::consolidate::list_staging_entries(&layout).len(), - 2, - "staging must remain untouched on reentry skip" - ); - assert_eq!( - client.call_count.load(Ordering::SeqCst), - 0, - "no LLM calls should fire on reentry skip" - ); - - // Sanity: when the flag is cleared, the same worker fires normally and - // resets the flag itself (i.e. it isn't accidentally sticky). - in_flight.store(false, Ordering::Release); - let client2 = MockClient::new(vec![done("ok")]); - let mut worker2 = - make_worker_with(FILES_THRESHOLD_TOML, pwd.path().to_path_buf(), client2).await; - worker2.try_post_run_consolidate().await.unwrap(); - assert!( - !worker2 - .consolidation_in_flight_handle() - .load(Ordering::Acquire), - "in-flight flag must be cleared after a normal run" - ); -} - -#[tokio::test] -async fn coalesce_loop_terminates_with_one_iteration_when_snapshot_drains_staging() { - use std::sync::atomic::Ordering; - - // Coalesce semantics from `docs/plan/memory.md` §並走防止: a single - // run consumes the snapshot taken at acquire time; the loop - // re-evaluates against any post-snapshot extract additions. With no - // concurrent additions, the second iteration sees an empty staging - // and bails out — exercised here by counting LLM calls. - let pwd = tempfile::tempdir().unwrap(); - let layout = WorkspaceLayout::new(pwd.path().to_path_buf()); - write_n_staging(&layout, 4); - - // Provide just one mock response. If the loop wrongly re-enters - // run_consolidate_once after Completed, the second sub-worker run - // would exhaust the mock and surface as an error. - let client = MockClient::new(vec![done("ok")]); - let mut worker = make_worker_with( - FILES_THRESHOLD_TOML, - pwd.path().to_path_buf(), - client.clone(), - ) - .await; - worker.try_post_run_consolidate().await.unwrap(); - - assert_eq!( - client.call_count.load(Ordering::SeqCst), - 1, - "Coalesce must terminate once the staging snapshot is drained — got an extra LLM call" - ); - assert!( - memory::consolidate::list_staging_entries(&layout).is_empty(), - "staging must be empty after the single iteration" - ); -} - -#[tokio::test] -async fn live_lock_held_by_other_worker_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 consolidation path must skip without error. - let _live_lock = memory::consolidate::StagingLock::acquire( - &layout, - std::process::id(), - "other-worker", - Vec::new(), - ) - .unwrap(); - - let client = MockClient::new(vec![]); - let mut worker = make_worker_with(FILES_THRESHOLD_TOML, pwd.path().to_path_buf(), client).await; - worker - .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); -} diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 535e6200..b4eee0ba 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -1,7 +1,7 @@ use std::collections::{HashMap, HashSet}; use std::path::{Component, Path, PathBuf}; -use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; use axum::extract::ws::{Message as WsMessage, WebSocket, WebSocketUpgrade}; use axum::extract::{Path as AxumPath, Query, State}; @@ -13,7 +13,8 @@ use axum::{Json, Router}; use chrono::{Duration, SecondsFormat, Utc}; use futures::{SinkExt, StreamExt}; use memory::backend::{ - MemoryBackendHttpResponse, MemoryBackendOperation, execute_memory_backend_operation, + MemoryBackendHttpResponse, MemoryBackendOperation, MemoryConsolidateStagingOperation, + MemoryConsolidationOutput, execute_memory_backend_operation, }; use protocol::stream::{decode_method, encode_event}; use serde::{Deserialize, Serialize}; @@ -230,6 +231,7 @@ pub struct WorkspaceApi { store: Arc, records: LocalProjectRecordReader, runtime: Arc, + memory_consolidater_worker: Arc>>, companion: Arc, observation_proxy: BackendObservationProxy, resource_broker: BackendResourceBroker, @@ -334,6 +336,7 @@ impl WorkspaceApi { config, store, runtime, + memory_consolidater_worker: Arc::new(Mutex::new(None)), companion, observation_proxy, resource_broker, @@ -488,6 +491,10 @@ pub fn build_router(api: WorkspaceApi) -> Router { "/api/w/{workspace_id}/memory/backend", post(scoped_memory_backend_operation), ) + .route( + "/api/w/{workspace_id}/memory/consolidation", + post(scoped_memory_consolidation), + ) .route("/api/tickets/{id}", get(get_ticket)) .route("/api/w/{workspace_id}/skills", get(scoped_list_skills)) .route("/api/w/{workspace_id}/skills/lint", get(scoped_lint_skills)) @@ -1470,6 +1477,198 @@ async fn scoped_memory_backend_operation( Ok(Json(response)) } +async fn scoped_memory_consolidation( + State(api): State, + AxumPath(path): AxumPath, + Json(operation): Json, +) -> ApiResult> { + validate_workspace_scope(&api, &path.workspace_id)?; + let memory_config = manifest::MemoryConfig::default(); + let layout = memory::WorkspaceLayout::resolve(&memory_config, &api.config.workspace_root); + Ok(Json(start_memory_staging_consolidation( + api, &layout, operation, + )?)) +} + +fn start_memory_staging_consolidation( + api: WorkspaceApi, + layout: &memory::WorkspaceLayout, + operation: MemoryConsolidateStagingOperation, +) -> ApiResult { + let snapshot = memory::consolidate::list_staging_entries_snapshot(layout); + let candidate_count = snapshot.entries.len(); + let total_bytes = snapshot + .entries + .iter() + .map(|entry| entry.bytes) + .sum::(); + if candidate_count == 0 { + return Ok(MemoryConsolidationOutput { + status: "skipped_empty".to_string(), + summary: "No Memory staging candidates are pending.".to_string(), + candidate_count, + total_bytes, + }); + } + let reached_files = operation + .threshold_files + .is_some_and(|threshold| candidate_count >= threshold); + let reached_bytes = operation + .threshold_bytes + .is_some_and(|threshold| total_bytes >= threshold); + if !operation.force && !reached_files && !reached_bytes { + return Ok(MemoryConsolidationOutput { + status: "skipped_below_threshold".to_string(), + summary: format!( + "Memory staging backlog has {candidate_count} candidate(s), {total_bytes} byte(s), below configured threshold." + ), + candidate_count, + total_bytes, + }); + } + + let runtime_id = select_memory_consolidation_runtime(&api)?; + if let Some((existing_runtime_id, existing_worker_id)) = api + .memory_consolidater_worker + .lock() + .expect("memory consolidater worker lock poisoned") + .clone() + { + match api + .runtime + .worker(&existing_runtime_id, &existing_worker_id) + { + Ok(worker) if worker.state == "idle" => { + let delete = api + .runtime + .delete_worker(&existing_runtime_id, &existing_worker_id) + .map_err(|err| err.into_error())?; + if delete.state != WorkerOperationState::Accepted || !delete.deleted { + return Ok(MemoryConsolidationOutput { + status: "skipped_existing_not_idle".to_string(), + summary: format!( + "Existing Memory consolidater '{}' was idle but could not be deleted.", + existing_worker_id + ), + candidate_count, + total_bytes, + }); + } + *api.memory_consolidater_worker + .lock() + .expect("memory consolidater worker lock poisoned") = None; + } + Ok(worker) => { + return Ok(MemoryConsolidationOutput { + status: "skipped_existing_not_idle".to_string(), + summary: format!( + "Existing Memory consolidater '{}' is '{}', not confirmed idle.", + existing_worker_id, worker.state + ), + candidate_count, + total_bytes, + }); + } + Err(_) => { + *api.memory_consolidater_worker + .lock() + .expect("memory consolidater worker lock poisoned") = None; + } + } + } + + let profile_selector = ProfileSelector::Builtin("memory-consolidation".to_string()); + let resolved_config_bundle = crate::profile_settings::build_workspace_profile_config_bundle( + &api.config.workspace_root, + &api.config.workspace_id, + &api.config.workspace_created_at, + "memory-consolidation", + )?; + let input = EmbeddedWorkerInput { + kind: EmbeddedWorkerInputKind::User, + content: format!( + "Process pending Memory staging candidates through MemoryStagingList, MemoryStagingRead, Memory tools, and MemoryStagingClose. Current backlog: {candidate_count} candidate(s), {total_bytes} byte(s)." + ), + segments: None, + }; + let result = api + .runtime + .spawn_worker( + &runtime_id, + WorkerSpawnRequest { + requested_worker_name: Some("memory-consolidation".to_string()), + intent: WorkerSpawnIntent::WorkspaceOrchestrator, + acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { + expected_segments: 1, + }, + profile: Some(profile_selector), + initial_input: Some(input), + working_directory_request: None, + resolved_working_directory_request: None, + resolved_working_directory: None, + resolved_config_bundle, + }, + ) + .map_err(|err| err.into_error())?; + if result.state != WorkerOperationState::Accepted { + return Ok(MemoryConsolidationOutput { + status: "skipped_spawn_rejected".to_string(), + summary: "Runtime rejected Memory consolidater spawn.".to_string(), + candidate_count, + total_bytes, + }); + } + let Some(worker) = result.worker else { + return Ok(MemoryConsolidationOutput { + status: "skipped_spawn_missing_worker".to_string(), + summary: + "Runtime accepted Memory consolidater spawn without returning a Worker summary." + .to_string(), + candidate_count, + total_bytes, + }); + }; + *api.memory_consolidater_worker + .lock() + .expect("memory consolidater worker lock poisoned") = + Some((worker.runtime_id.clone(), worker.worker_id.clone())); + Ok(MemoryConsolidationOutput { + status: "started".to_string(), + summary: format!( + "Started Memory consolidater '{}' for {candidate_count} staging candidate(s).", + worker.worker_id + ), + candidate_count, + total_bytes, + }) +} + +fn select_memory_consolidation_runtime(api: &WorkspaceApi) -> ApiResult { + let runtimes = api.runtime.list_runtimes(100); + if let Some(runtime) = runtimes + .items + .iter() + .find(|runtime| { + runtime.runtime_id == EMBEDDED_WORKER_RUNTIME_ID + && runtime.capabilities.can_spawn_worker + }) + .or_else(|| { + runtimes + .items + .iter() + .find(|runtime| runtime.capabilities.can_spawn_worker) + }) + { + return Ok(runtime.runtime_id.clone()); + } + Err(Error::RuntimeOperationFailed { + runtime_id: "memory-consolidation".to_string(), + code: "memory_consolidation_runtime_unavailable".to_string(), + message: "No runtime capable of spawning the Memory consolidater is available".to_string(), + } + .into()) +} + async fn scoped_list_skills( State(api): State, AxumPath(path): AxumPath, diff --git a/resources/profiles/memory-consolidation.dcdl b/resources/profiles/memory-consolidation.dcdl new file mode 100644 index 00000000..a0a33f19 --- /dev/null +++ b/resources/profiles/memory-consolidation.dcdl @@ -0,0 +1,9 @@ +worker = { name = "memory-consolidation"; } +feature = { + task = { enabled = false; }; + memory = { enabled = true; }; + web = { enabled = false; }; + workers = { enabled = false; }; + ticket = { enabled = false; thread = false; }; +}; +system_prompt = { file = "internal/memory_consolidation_system.md"; }; diff --git a/resources/prompts/internal/memory_consolidation_system.md b/resources/prompts/internal/memory_consolidation_system.md index c6293049..88635624 100644 --- a/resources/prompts/internal/memory_consolidation_system.md +++ b/resources/prompts/internal/memory_consolidation_system.md @@ -1,16 +1,26 @@ -# Memory consolidation worker +# Memory staging consolidater -Your job is to take extract activity-log staging entries together with the workspace's current `memory/*` records, then run two steps back-to-back in this single session: +You are the internal Memory staging consolidater. -1. **Integration step** — fold staging into memory. -2. **Tidy step** — produce a compact tidy report for stale / redundant / protected memory records. +Your job is to consume Memory staging candidates through tools. Do not assume the prompt contains all staging data. -You may use: +Use this loop: -- `MemoryQuery` to find existing memory records. -- `MemoryRead`, `MemoryWrite`, `MemoryEdit`, `MemoryDelete` for memory records. +1. Call `MemoryStagingList` to inspect pending candidates. +2. Pick one candidate and call `MemoryStagingRead` for the full record. +3. Use `MemoryQuery` / `MemoryRead` to compare against durable Memory. +4. If the candidate should change durable Memory, use `MemoryWrite`, `MemoryEdit`, or `MemoryDelete` first. +5. Call `MemoryStagingClose` exactly once for every candidate you finish. Always include a concrete reason. -Your initial user message contains the staging entries, the full memory records, and the tidy hints. +`MemoryStagingClose` removes the staging candidate after recording your disposition. Use it for both applied candidates and candidates that should not become Memory. + +Valid close actions: + +- `applied`: you changed durable Memory for this candidate. +- `discarded`: the candidate is understandable but should not become durable Memory. +- `invalid`: the staging record is malformed or cannot be interpreted. +- `duplicate`: the candidate duplicates another candidate or record. +- `already_covered`: durable Memory already covers the candidate sufficiently. ## Language @@ -19,24 +29,4 @@ Your initial user message contains the staging entries, the full memory records, - Preserve literal identifiers, paths, commands, branch names, issue IDs, tool names, model names, and quoted user/system text as-is. - If the configured language is unclear, use English. -## Integration step - -The generated memory principle is: do not mirror tickets, task boards, reports, changelogs, git history, or generated artifacts verbatim. Keep durable abstractions, policy, rationale, recurring constraints, and user preferences. - -- Summary (`summary.md`) is resident body context. Keep it concise and strategic. -- Decisions (`memory/decisions/*.md`) capture durable accepted direction/rationale. -- Requests (`memory/requests/*.md`) capture durable user preferences or standing instructions. -- Preserve and update sources for decisions/requests when staging entries support them. -- **Do not invent provenance.** Decisions / Requests `sources` arrays MUST be copied from the staging `source` field for the originating activity log entries. Do not synthesise `session_id` or entry ranges. -- Keep records useful as durable context. Delete or merge stale duplicates only when safe and supported by the supplied evidence. - -## Tidy step - -Once the integration step is done, evaluate every existing memory record against four categories: - -- `obsolete`: contradicted or no longer useful. -- `superseded`: replaced by a newer/more accurate record; include the replacement slug if obvious. -- `duplicate`: overlaps enough to merge/delete; include the canonical slug if obvious. -- `protected`: keep despite low recent use because it is foundational, policy-like, safety-critical, or currently relevant. - -Emit tidy recommendations in your final response. Do not delete protected records. +Do not create Tickets, documents, child Workers, or filesystem files. Use only Memory and MemoryStaging tools.