memory: route staging lifecycle through sqlite authority

This commit is contained in:
Keisuke Hirata 2026-07-26 15:44:51 +09:00
parent faaf258a96
commit e751a11a92
No known key found for this signature in database
5 changed files with 191 additions and 71 deletions

View File

@ -80,8 +80,8 @@ pub enum WorkerPrompt {
/// after the scope summary when an AGENTS.md is present. /// after the scope summary when an AGENTS.md is present.
AgentsMdSection, AgentsMdSection,
/// Trailing `## Resident memory summary` section, appended after the /// Trailing `## Resident memory summary` section, appended after the
/// AGENTS.md section when memory is enabled, summary injection is enabled, /// AGENTS.md section when memory is enabled, resident injection is enabled,
/// and `memory/summary.md` has a valid non-empty body. /// and the workspace Memory document has a valid non-empty body.
ResidentMemorySummarySection, ResidentMemorySummarySection,
/// Trailing Worker orchestration guidance, appended when registered tools /// Trailing Worker orchestration guidance, appended when registered tools
/// include Worker-management capabilities. /// include Worker-management capabilities.

View File

@ -153,9 +153,9 @@ pub struct SystemPromptContext<'a> {
/// Not visible from the template; consumed by the trailing-section /// Not visible from the template; consumed by the trailing-section
/// formatter in [`SystemPromptTemplate::render`]. /// formatter in [`SystemPromptTemplate::render`].
pub agents_md: Option<String>, pub agents_md: Option<String>,
/// The body of `<workspace>/.yoi/memory/summary.md`, with /// The body of the workspace Memory document. `None` disables the
/// frontmatter stripped. `None` disables the resident summary section; /// resident summary section; empty strings are ignored by the trailing-section
/// empty strings are ignored by the trailing-section formatter. /// formatter.
pub resident_summary: Option<&'a str>, pub resident_summary: Option<&'a str>,
/// Catalog used to render the fixed trailing section headers. /// Catalog used to render the fixed trailing section headers.
/// Passed by reference so callers do not give up ownership across /// Passed by reference so callers do not give up ownership across

View File

@ -507,8 +507,8 @@ pub struct Worker<C: LlmClient, St: Store> {
/// [`Self::from_manifest`], or defaults to the builtin pack when a /// [`Self::from_manifest`], or defaults to the builtin pack when a
/// Worker is constructed through lower-level paths that have no loader. /// Worker is constructed through lower-level paths that have no loader.
prompts: Arc<PromptCatalog>, prompts: Arc<PromptCatalog>,
/// When true (default), the system-prompt assembler may append the /// When true (default), the system-prompt assembler may append resident
/// workspace memory summary (`memory/summary.md`). Internal disposable /// context from the workspace Memory document. Internal disposable
/// workers disable this so resident memory exposure is opt-in per Worker. /// workers disable this so resident memory exposure is opt-in per Worker.
inject_resident_summary: bool, inject_resident_summary: bool,
/// When true (default), the system-prompt assembler may append resident /// When true (default), the system-prompt assembler may append resident
@ -833,7 +833,7 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
self.inject_resident_summary = enabled; self.inject_resident_summary = enabled;
} }
/// Toggle `memory/summary.md` resident injection in the system prompt. /// Toggle workspace Memory document resident injection in the system prompt.
pub fn set_resident_summary_injection(&mut self, enabled: bool) { pub fn set_resident_summary_injection(&mut self, enabled: bool) {
self.inject_resident_summary = enabled; self.inject_resident_summary = enabled;
} }

View File

@ -1,5 +1,3 @@
use memory::WorkspaceLayout;
use memory::consolidate::{StagingEntry, list_staging_entries_snapshot};
use memory::extract::StagingRecord; use memory::extract::StagingRecord;
use memory::schema::{SourceEvidenceRef, SourceRef}; use memory::schema::{SourceEvidenceRef, SourceRef};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
@ -63,32 +61,11 @@ pub struct MemorySourceEvidenceRefSummary {
pub summary: Option<String>, pub summary: Option<String>,
} }
pub fn list_memory_staging( #[derive(Debug, Clone, Copy, PartialEq, Eq)]
layout: &WorkspaceLayout, pub struct MemoryStagingBacklogSummary {
requested_limit: Option<usize>, pub candidate_count: usize,
) -> MemoryStagingListResponse { pub total_bytes: u64,
let limit = requested_limit pub invalid_count: usize,
.unwrap_or(DEFAULT_MEMORY_STAGING_LIMIT)
.min(MAX_MEMORY_STAGING_LIMIT);
let snapshot = list_staging_entries_snapshot(layout);
let total_valid_count = snapshot.entries.len();
let items = snapshot
.entries
.into_iter()
.take(limit)
.map(memory_staging_entry_summary)
.collect::<Vec<_>>();
let returned_count = items.len();
MemoryStagingListResponse {
limit,
returned_count,
total_valid_count,
invalid_count: snapshot.invalid_count,
truncated: total_valid_count > returned_count,
order: "uuidv7_ascending".to_string(),
record_authority: "workspace_memory_staging".to_string(),
items,
}
} }
pub fn list_memory_staging_from_authority<A: MemoryAuthority>( pub fn list_memory_staging_from_authority<A: MemoryAuthority>(
@ -127,18 +104,33 @@ pub fn list_memory_staging_from_authority<A: MemoryAuthority>(
total_valid_count, total_valid_count,
invalid_count, invalid_count,
truncated: total_valid_count > returned_count || fetched_count > MAX_MEMORY_STAGING_LIMIT, truncated: total_valid_count > returned_count || fetched_count > MAX_MEMORY_STAGING_LIMIT,
order: "uuidv7_ascending".to_string(), order: "imported_at_desc_candidate_id_asc".to_string(),
record_authority: "sqlite_workspace_authority.memory_staging".to_string(), record_authority: "sqlite_workspace_authority.memory_staging".to_string(),
items: valid_items, items: valid_items,
}) })
} }
fn memory_staging_entry_summary(entry: StagingEntry) -> MemoryStagingEntrySummary { pub fn memory_staging_backlog_from_authority<A: MemoryAuthority>(
MemoryStagingEntrySummary { authority: &A,
id: entry.id.to_string(), ) -> Result<MemoryStagingBacklogSummary> {
byte_len: entry.bytes, let entries = authority.list_memory_staging_records(i64::MAX as usize)?;
record: memory_staging_record_summary(entry.record), let mut candidate_count = 0usize;
let mut total_bytes = 0u64;
let mut invalid_count = 0usize;
for entry in entries {
match serde_json::from_str::<StagingRecord>(&entry.raw_json) {
Ok(_) => {
candidate_count += 1;
total_bytes += entry.raw_json.len() as u64;
} }
Err(_) => invalid_count += 1,
}
}
Ok(MemoryStagingBacklogSummary {
candidate_count,
total_bytes,
invalid_count,
})
} }
fn memory_staging_record_summary(record: StagingRecord) -> MemoryStagingRecordSummary { fn memory_staging_record_summary(record: StagingRecord) -> MemoryStagingRecordSummary {
@ -189,7 +181,9 @@ fn memory_source_evidence_ref_summary(
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
use memory::extract::{CandidateKind, ExtractedCandidate, ExtractedPayload, write_staging}; use crate::authority::{MemoryAuthority, SqliteWorkspaceAuthority};
use crate::store::{ControlPlaneStore, SqliteWorkspaceStore, WorkspaceRecord};
use memory::extract::{CandidateKind, ExtractedCandidate, StagingRecord};
use tempfile::TempDir; use tempfile::TempDir;
fn source() -> SourceRef { fn source() -> SourceRef {
@ -199,33 +193,93 @@ mod tests {
} }
} }
fn payload(claim: &str) -> ExtractedPayload { fn record_json(id: &str, claim: &str) -> String {
ExtractedPayload { let record = StagingRecord::from_candidate(
candidates: vec![ExtractedCandidate { id,
"extract-run-1",
source(),
ExtractedCandidate {
kind: CandidateKind::Decision, kind: CandidateKind::Decision,
claim: claim.to_string(), claim: claim.to_string(),
why_useful: "useful for future work".to_string(), why_useful: "useful for future work".to_string(),
staleness: None, staleness: None,
evidence_ids: Vec::new(), evidence_ids: Vec::new(),
}], },
} Vec::new(),
Vec::new(),
);
serde_json::to_string(&record).unwrap()
} }
#[test] async fn authority() -> (TempDir, SqliteWorkspaceAuthority) {
fn lists_memory_staging_records_with_cap() {
let temp = TempDir::new().unwrap(); let temp = TempDir::new().unwrap();
let layout = WorkspaceLayout::new(temp.path().to_path_buf()); let db_path = temp.path().join("workspace.db");
write_staging(&layout, source(), payload("first claim")).unwrap(); let store = SqliteWorkspaceStore::open(&db_path).unwrap();
write_staging(&layout, source(), payload("second claim")).unwrap(); store
.upsert_workspace(&WorkspaceRecord {
workspace_id: "workspace-test".to_string(),
owner_account_id: None,
display_name: "Workspace Test".to_string(),
state: "active".to_string(),
created_at: "2026-01-01T00:00:00Z".to_string(),
updated_at: "2026-01-01T00:00:00Z".to_string(),
})
.await
.unwrap();
let authority = SqliteWorkspaceAuthority::new(db_path, "workspace-test").unwrap();
(temp, authority)
}
let response = list_memory_staging(&layout, Some(1)); #[tokio::test]
async fn lists_memory_staging_records_with_cap_from_authority() {
let (_temp, authority) = authority().await;
authority
.upsert_memory_staging_record(
"00000000000000000000000001",
&record_json("00000000000000000000000001", "first claim"),
None,
)
.unwrap();
authority
.upsert_memory_staging_record(
"00000000000000000000000002",
&record_json("00000000000000000000000002", "second claim"),
None,
)
.unwrap();
let response = list_memory_staging_from_authority(&authority, Some(1)).unwrap();
assert_eq!(response.limit, 1); assert_eq!(response.limit, 1);
assert_eq!(response.returned_count, 1); assert_eq!(response.returned_count, 1);
assert_eq!(response.total_valid_count, 2); assert_eq!(response.total_valid_count, 2);
assert!(response.truncated); assert!(response.truncated);
assert_eq!(response.record_authority, "workspace_memory_staging"); assert_eq!(
response.record_authority,
"sqlite_workspace_authority.memory_staging"
);
assert_eq!(response.items[0].record.kind, "decision"); assert_eq!(response.items[0].record.kind, "decision");
assert_eq!(response.items[0].record.claim, "first claim"); }
#[tokio::test]
async fn summarizes_memory_staging_backlog_from_authority_records() {
let (_temp, authority) = authority().await;
let first = record_json("00000000000000000000000001", "first claim");
let second = record_json("00000000000000000000000002", "second claim");
authority
.upsert_memory_staging_record("00000000000000000000000001", &first, None)
.unwrap();
authority
.upsert_memory_staging_record("00000000000000000000000002", &second, None)
.unwrap();
authority
.upsert_memory_staging_record("invalid-json-shape", r#"{"legacy":"shape"}"#, None)
.unwrap();
let backlog = memory_staging_backlog_from_authority(&authority).unwrap();
assert_eq!(backlog.candidate_count, 2);
assert_eq!(backlog.total_bytes, (first.len() + second.len()) as u64);
assert_eq!(backlog.invalid_count, 1);
} }
} }

View File

@ -59,7 +59,10 @@ use crate::hosts::{
}; };
use crate::identity::WorkspaceIdentity; use crate::identity::WorkspaceIdentity;
use crate::memory_backend::execute_memory_backend_operation_with_authority; use crate::memory_backend::execute_memory_backend_operation_with_authority;
use crate::memory_staging::{MemoryStagingListResponse, list_memory_staging_from_authority}; use crate::memory_staging::{
MemoryStagingListResponse, list_memory_staging_from_authority,
memory_staging_backlog_from_authority,
};
use crate::observation::{ use crate::observation::{
BackendObservationProxy, ObservationProxyError, RuntimeObservationClient, BackendObservationProxy, ObservationProxyError, RuntimeObservationClient,
RuntimeObservationSource, RuntimeObservationSourceConfig, RuntimeObservationSource, RuntimeObservationSourceConfig,
@ -1487,25 +1490,16 @@ async fn scoped_memory_consolidation(
Json(operation): Json<MemoryConsolidateStagingOperation>, Json(operation): Json<MemoryConsolidateStagingOperation>,
) -> ApiResult<Json<MemoryConsolidationOutput>> { ) -> ApiResult<Json<MemoryConsolidationOutput>> {
validate_workspace_scope(&api, &path.workspace_id)?; validate_workspace_scope(&api, &path.workspace_id)?;
let memory_config = manifest::MemoryConfig::default(); Ok(Json(start_memory_staging_consolidation(api, operation)?))
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( fn start_memory_staging_consolidation(
api: WorkspaceApi, api: WorkspaceApi,
layout: &memory::WorkspaceLayout,
operation: MemoryConsolidateStagingOperation, operation: MemoryConsolidateStagingOperation,
) -> ApiResult<MemoryConsolidationOutput> { ) -> ApiResult<MemoryConsolidationOutput> {
let snapshot = memory::consolidate::list_staging_entries_snapshot(layout); let backlog = memory_staging_backlog_from_authority(&api.authority)?;
let candidate_count = snapshot.entries.len(); let candidate_count = backlog.candidate_count;
let total_bytes = snapshot let total_bytes = backlog.total_bytes;
.entries
.iter()
.map(|entry| entry.bytes)
.sum::<u64>();
if candidate_count == 0 { if candidate_count == 0 {
return Ok(MemoryConsolidationOutput { return Ok(MemoryConsolidationOutput {
status: "skipped_empty".to_string(), status: "skipped_empty".to_string(),
@ -7219,6 +7213,62 @@ mod tests {
config config
} }
fn memory_staging_record_json(id: &str, claim: &str) -> String {
json!({
"schema_version": 1,
"id": id,
"extract_run_id": "extract-run-1",
"source": {
"segment_id": "segment-1",
"range": [0, 10],
},
"kind": "working_assumption",
"claim": claim,
"why_useful": "useful for future work",
"staleness": null,
"evidence": [],
"source_refs": [],
})
.to_string()
}
#[tokio::test]
async fn memory_consolidation_backlog_ignores_legacy_filesystem_staging() {
let dir = tempfile::tempdir().unwrap();
let config = test_server_config(dir.path());
let store = SqliteWorkspaceStore::open(&config.database_path).unwrap();
let api = WorkspaceApi::new_with_execution_backend(
config,
Arc::new(store),
Arc::new(DeterministicExecutionBackend::default()),
)
.await
.unwrap();
let legacy_staging_dir = dir.path().join(".yoi/memory/_staging");
fs::create_dir_all(&legacy_staging_dir).unwrap();
fs::write(
legacy_staging_dir.join("00000000001J4.json"),
memory_staging_record_json("00000000001J4", "legacy filesystem candidate"),
)
.unwrap();
let output = match start_memory_staging_consolidation(
api,
MemoryConsolidateStagingOperation {
force: true,
threshold_files: None,
threshold_bytes: None,
},
) {
Ok(output) => output,
Err(_) => panic!("unexpected ApiError from memory consolidation trigger"),
};
assert_eq!(output.status, "skipped_empty");
assert_eq!(output.candidate_count, 0);
assert_eq!(output.total_bytes, 0);
}
fn init_clean_git_workspace(path: &std::path::Path) { fn init_clean_git_workspace(path: &std::path::Path) {
for args in [ for args in [
vec!["init"], vec!["init"],
@ -8377,7 +8427,7 @@ mod tests {
.upsert_memory_staging_record(&MemoryStagingRecord { .upsert_memory_staging_record(&MemoryStagingRecord {
workspace_id: TEST_WORKSPACE_ID.to_string(), workspace_id: TEST_WORKSPACE_ID.to_string(),
candidate_id: "00000000001J4".to_string(), candidate_id: "00000000001J4".to_string(),
raw_json: r#"{"id":"00000000001J4","kind":"working_assumption"}"#.to_string(), raw_json: memory_staging_record_json("00000000001J4", "SQLite memory candidate"),
source_path: None, source_path: None,
imported_at: "2026-01-01T00:00:00Z".to_string(), imported_at: "2026-01-01T00:00:00Z".to_string(),
}) })
@ -8509,6 +8559,22 @@ mod tests {
"memory-architecture-overview.md" "memory-architecture-overview.md"
); );
let memory_staging = get_json(
app.clone(),
&format!("/api/w/{TEST_WORKSPACE_ID}/memory/staging?limit=10"),
)
.await;
assert_eq!(
memory_staging["record_authority"],
"sqlite_workspace_authority.memory_staging"
);
assert_eq!(memory_staging["total_valid_count"], 1);
assert_eq!(memory_staging["invalid_count"], 0);
assert_eq!(
memory_staging["items"][0]["record"]["claim"],
"SQLite memory candidate"
);
let repositories = get_json(app.clone(), "/api/repositories").await; let repositories = get_json(app.clone(), "/api/repositories").await;
assert_eq!(repositories["items"][0]["id"], TEST_REPOSITORY_ID); assert_eq!(repositories["items"][0]["id"], TEST_REPOSITORY_ID);
assert_eq!(repositories["items"][0]["kind"], "git"); assert_eq!(repositories["items"][0]["kind"], "git");