feat: connect memory tools to workspace authority
This commit is contained in:
@@ -28,6 +28,8 @@ use crate::workspace::WorkspaceLayout;
|
||||
#[serde(tag = "operation", rename_all = "snake_case")]
|
||||
pub enum MemoryBackendOperation {
|
||||
Query(MemoryQueryOperation),
|
||||
ReadDocument(MemoryDocumentReadOperation),
|
||||
UpdateDocument(MemoryDocumentUpdateOperation),
|
||||
Read(MemoryReadOperation),
|
||||
Write(MemoryWriteOperation),
|
||||
Edit(MemoryEditOperation),
|
||||
@@ -73,6 +75,19 @@ pub struct MemoryQueryOperation {
|
||||
pub query: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
|
||||
pub struct MemoryDocumentReadOperation {
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub offset: Option<usize>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub limit: Option<usize>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
|
||||
pub struct MemoryDocumentUpdateOperation {
|
||||
pub body_md: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct MemoryReadOperation {
|
||||
pub kind: MemoryToolKind,
|
||||
@@ -232,6 +247,14 @@ pub fn execute_memory_backend_operation(
|
||||
MemoryBackendOperation::Query(operation) => {
|
||||
execute_query(layout, operation).map(MemoryBackendOperationResult::ToolOutput)
|
||||
}
|
||||
MemoryBackendOperation::ReadDocument(_operation) => Err(io::Error::new(
|
||||
io::ErrorKind::Unsupported,
|
||||
"Memory document operations require WorkspaceAuthority-backed executor",
|
||||
)),
|
||||
MemoryBackendOperation::UpdateDocument(_operation) => Err(io::Error::new(
|
||||
io::ErrorKind::Unsupported,
|
||||
"Memory document operations require WorkspaceAuthority-backed executor",
|
||||
)),
|
||||
MemoryBackendOperation::Read(operation) => {
|
||||
execute_read(layout, operation).map(MemoryBackendOperationResult::ToolOutput)
|
||||
}
|
||||
|
||||
@@ -12,9 +12,9 @@ use llm_engine::tool::{
|
||||
};
|
||||
use memory::backend::{
|
||||
MemoryBackendHttpResponse, MemoryBackendOperation, MemoryBackendOperationResult,
|
||||
MemoryConsolidateStagingOperation, MemoryConsolidationOutput, MemoryDeleteOperation,
|
||||
MemoryEditOperation, MemoryQueryOperation, MemoryReadOperation, MemoryStagingCloseOperation,
|
||||
MemoryStagingListOperation, MemoryStagingReadOperation, MemoryToolOutput, MemoryWriteOperation,
|
||||
MemoryConsolidateStagingOperation, MemoryConsolidationOutput, MemoryDocumentReadOperation,
|
||||
MemoryDocumentUpdateOperation, MemoryQueryOperation, MemoryStagingCloseOperation,
|
||||
MemoryStagingListOperation, MemoryStagingReadOperation, MemoryToolOutput,
|
||||
};
|
||||
use schemars::JsonSchema;
|
||||
use serde::de::DeserializeOwned;
|
||||
@@ -175,47 +175,29 @@ pub fn workspace_http_memory_tools(
|
||||
let backend = WorkspaceHttpMemoryBackend::new(workspace_id, base_url);
|
||||
vec![
|
||||
memory_tool(
|
||||
"MemoryRead",
|
||||
READ_DESCRIPTION,
|
||||
read_schema(),
|
||||
"MemoryReadDocument",
|
||||
READ_DOCUMENT_DESCRIPTION,
|
||||
document_read_schema(),
|
||||
backend.clone(),
|
||||
|input| {
|
||||
Ok(MemoryBackendOperation::Read(parse_input::<
|
||||
MemoryReadOperation,
|
||||
>(input)?))
|
||||
Ok(MemoryBackendOperation::ReadDocument(parse_input::<
|
||||
MemoryDocumentReadOperation,
|
||||
>(
|
||||
input
|
||||
)?))
|
||||
},
|
||||
),
|
||||
memory_tool(
|
||||
"MemoryWrite",
|
||||
WRITE_DESCRIPTION,
|
||||
write_schema(),
|
||||
"MemoryUpdateDocument",
|
||||
UPDATE_DOCUMENT_DESCRIPTION,
|
||||
document_update_schema(),
|
||||
backend.clone(),
|
||||
|input| {
|
||||
Ok(MemoryBackendOperation::Write(parse_input::<
|
||||
MemoryWriteOperation,
|
||||
>(input)?))
|
||||
},
|
||||
),
|
||||
memory_tool(
|
||||
"MemoryEdit",
|
||||
EDIT_DESCRIPTION,
|
||||
edit_schema(),
|
||||
backend.clone(),
|
||||
|input| {
|
||||
Ok(MemoryBackendOperation::Edit(parse_input::<
|
||||
MemoryEditOperation,
|
||||
>(input)?))
|
||||
},
|
||||
),
|
||||
memory_tool(
|
||||
"MemoryDelete",
|
||||
DELETE_DESCRIPTION,
|
||||
delete_schema(),
|
||||
backend.clone(),
|
||||
|input| {
|
||||
Ok(MemoryBackendOperation::Delete(parse_input::<
|
||||
MemoryDeleteOperation,
|
||||
>(input)?))
|
||||
Ok(MemoryBackendOperation::UpdateDocument(parse_input::<
|
||||
MemoryDocumentUpdateOperation,
|
||||
>(
|
||||
input
|
||||
)?))
|
||||
},
|
||||
),
|
||||
memory_tool(
|
||||
@@ -339,72 +321,34 @@ fn tool_output(output: MemoryToolOutput) -> ToolOutput {
|
||||
}
|
||||
}
|
||||
|
||||
const READ_DESCRIPTION: &str = "Read a durable memory record through Workspace authority.";
|
||||
const WRITE_DESCRIPTION: &str =
|
||||
"Create or overwrite a durable memory record through Workspace authority.";
|
||||
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 READ_DOCUMENT_DESCRIPTION: &str =
|
||||
"Read the Workspace memory Markdown document through Workspace authority.";
|
||||
const UPDATE_DOCUMENT_DESCRIPTION: &str =
|
||||
"Replace the Workspace memory Markdown document through Workspace authority.";
|
||||
const QUERY_DESCRIPTION: &str = "Query the Workspace memory document 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"]})
|
||||
}
|
||||
|
||||
fn read_schema() -> serde_json::Value {
|
||||
fn document_read_schema() -> serde_json::Value {
|
||||
json!({
|
||||
"type":"object",
|
||||
"additionalProperties": false,
|
||||
"required":["kind"],
|
||||
"properties":{
|
||||
"kind": kind_schema(),
|
||||
"slug":{"type":["string","null"]},
|
||||
"offset":{"type":["integer","null"],"minimum":0},
|
||||
"limit":{"type":["integer","null"],"minimum":0}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
fn write_schema() -> serde_json::Value {
|
||||
fn document_update_schema() -> serde_json::Value {
|
||||
json!({
|
||||
"type":"object",
|
||||
"additionalProperties": false,
|
||||
"required":["kind","content"],
|
||||
"required":["body_md"],
|
||||
"properties":{
|
||||
"kind": kind_schema(),
|
||||
"slug":{"type":["string","null"]},
|
||||
"content":{"type":"string"}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
fn edit_schema() -> serde_json::Value {
|
||||
json!({
|
||||
"type":"object",
|
||||
"additionalProperties": false,
|
||||
"required":["kind","old_string","new_string"],
|
||||
"properties":{
|
||||
"kind": kind_schema(),
|
||||
"slug":{"type":["string","null"]},
|
||||
"old_string":{"type":"string"},
|
||||
"new_string":{"type":"string"},
|
||||
"replace_all":{"type":"boolean","default":false}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
fn delete_schema() -> serde_json::Value {
|
||||
json!({
|
||||
"type":"object",
|
||||
"additionalProperties": false,
|
||||
"required":["kind"],
|
||||
"properties":{
|
||||
"kind": kind_schema(),
|
||||
"slug":{"type":["string","null"]}
|
||||
"body_md":{"type":"string"}
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -441,6 +385,12 @@ mod tests {
|
||||
));
|
||||
|
||||
assert!(names.contains(&"MemoryQuery".to_string()));
|
||||
assert!(names.contains(&"MemoryReadDocument".to_string()));
|
||||
assert!(names.contains(&"MemoryUpdateDocument".to_string()));
|
||||
assert!(!names.contains(&"MemoryRead".to_string()));
|
||||
assert!(!names.contains(&"MemoryWrite".to_string()));
|
||||
assert!(!names.contains(&"MemoryEdit".to_string()));
|
||||
assert!(!names.contains(&"MemoryDelete".to_string()));
|
||||
assert!(!names.contains(&"MemoryStagingList".to_string()));
|
||||
assert!(!names.contains(&"MemoryStagingRead".to_string()));
|
||||
assert!(!names.contains(&"MemoryStagingClose".to_string()));
|
||||
@@ -454,6 +404,8 @@ mod tests {
|
||||
));
|
||||
|
||||
assert!(names.contains(&"MemoryQuery".to_string()));
|
||||
assert!(names.contains(&"MemoryReadDocument".to_string()));
|
||||
assert!(names.contains(&"MemoryUpdateDocument".to_string()));
|
||||
assert!(names.contains(&"MemoryStagingList".to_string()));
|
||||
assert!(names.contains(&"MemoryStagingRead".to_string()));
|
||||
assert!(names.contains(&"MemoryStagingClose".to_string()));
|
||||
|
||||
@@ -201,10 +201,8 @@ impl<'a> SystemPromptContext<'a> {
|
||||
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
|
||||
struct ToolCapabilities {
|
||||
memory_query: bool,
|
||||
memory_read: bool,
|
||||
memory_write: bool,
|
||||
memory_edit: bool,
|
||||
memory_delete: bool,
|
||||
memory_read_document: bool,
|
||||
memory_update_document: bool,
|
||||
worker_spawn: bool,
|
||||
worker_send: bool,
|
||||
worker_read_output: bool,
|
||||
@@ -219,10 +217,8 @@ impl ToolCapabilities {
|
||||
for name in names {
|
||||
match name.as_str() {
|
||||
"MemoryQuery" => capabilities.memory_query = true,
|
||||
"MemoryRead" => capabilities.memory_read = true,
|
||||
"MemoryWrite" => capabilities.memory_write = true,
|
||||
"MemoryEdit" => capabilities.memory_edit = true,
|
||||
"MemoryDelete" => capabilities.memory_delete = true,
|
||||
"MemoryReadDocument" => capabilities.memory_read_document = true,
|
||||
"MemoryUpdateDocument" => capabilities.memory_update_document = true,
|
||||
"SpawnWorker" => capabilities.worker_spawn = true,
|
||||
"SendToWorker" => capabilities.worker_send = true,
|
||||
"ReadWorkerOutput" => capabilities.worker_read_output = true,
|
||||
@@ -236,11 +232,7 @@ impl ToolCapabilities {
|
||||
}
|
||||
|
||||
fn memory_records(self) -> bool {
|
||||
self.memory_query
|
||||
|| self.memory_read
|
||||
|| self.memory_write
|
||||
|| self.memory_edit
|
||||
|| self.memory_delete
|
||||
self.memory_query || self.memory_read_document || self.memory_update_document
|
||||
}
|
||||
|
||||
fn memory_any(self) -> bool {
|
||||
@@ -248,7 +240,7 @@ impl ToolCapabilities {
|
||||
}
|
||||
|
||||
fn memory_mutation(self) -> bool {
|
||||
self.memory_write || self.memory_edit || self.memory_delete
|
||||
self.memory_update_document
|
||||
}
|
||||
|
||||
fn worker_management(self) -> bool {
|
||||
@@ -265,10 +257,14 @@ impl ToolCapabilities {
|
||||
map.insert("memory_any", Value::from(self.memory_any()));
|
||||
map.insert("memory_records", Value::from(self.memory_records()));
|
||||
map.insert("memory_query", Value::from(self.memory_query));
|
||||
map.insert("memory_read", Value::from(self.memory_read));
|
||||
map.insert("memory_write", Value::from(self.memory_write));
|
||||
map.insert("memory_edit", Value::from(self.memory_edit));
|
||||
map.insert("memory_delete", Value::from(self.memory_delete));
|
||||
map.insert(
|
||||
"memory_read_document",
|
||||
Value::from(self.memory_read_document),
|
||||
);
|
||||
map.insert(
|
||||
"memory_update_document",
|
||||
Value::from(self.memory_update_document),
|
||||
);
|
||||
map.insert("memory_mutation", Value::from(self.memory_mutation()));
|
||||
map.insert("worker_management", Value::from(self.worker_management()));
|
||||
Value::from(map)
|
||||
@@ -407,16 +403,10 @@ mod tests {
|
||||
}
|
||||
|
||||
fn memory_tool_names() -> Vec<String> {
|
||||
[
|
||||
"MemoryQuery",
|
||||
"MemoryRead",
|
||||
"MemoryWrite",
|
||||
"MemoryEdit",
|
||||
"MemoryDelete",
|
||||
]
|
||||
.into_iter()
|
||||
.map(String::from)
|
||||
.collect()
|
||||
["MemoryQuery", "MemoryReadDocument", "MemoryUpdateDocument"]
|
||||
.into_iter()
|
||||
.map(String::from)
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn worker_management_tool_names() -> Vec<String> {
|
||||
@@ -463,9 +453,9 @@ mod tests {
|
||||
assert!(rendered.contains("### Memory"));
|
||||
assert!(rendered.contains("small targeted `MemoryQuery`"));
|
||||
assert!(rendered.contains("Strong lookup triggers include"));
|
||||
assert!(rendered.contains("MemoryRead(kind=summary)"));
|
||||
assert!(rendered.contains("MemoryReadDocument"));
|
||||
assert!(rendered.contains("Do not query memory every turn"));
|
||||
assert!(rendered.contains("MemoryWrite"));
|
||||
assert!(rendered.contains("MemoryUpdateDocument"));
|
||||
assert!(rendered.contains("## Language"));
|
||||
assert!(rendered.contains("`language`: `match the user's language"));
|
||||
// Trailing section must be present.
|
||||
@@ -508,15 +498,15 @@ mod tests {
|
||||
.render(&ctx(
|
||||
dir.path(),
|
||||
&scope,
|
||||
vec!["MemoryQuery".into(), "MemoryRead".into()],
|
||||
vec!["MemoryQuery".into(), "MemoryReadDocument".into()],
|
||||
None,
|
||||
))
|
||||
.unwrap();
|
||||
|
||||
assert!(rendered.contains("### Memory"));
|
||||
assert!(rendered.contains("small targeted `MemoryQuery`"));
|
||||
assert!(rendered.contains("MemoryRead(kind=summary)"));
|
||||
assert!(!rendered.contains("MemoryWrite"));
|
||||
assert!(rendered.contains("MemoryReadDocument"));
|
||||
assert!(!rendered.contains("MemoryUpdateDocument"));
|
||||
assert!(!rendered.contains("MemoryEdit"));
|
||||
assert!(!rendered.contains("MemoryDelete"));
|
||||
}
|
||||
@@ -555,7 +545,7 @@ mod tests {
|
||||
.render(&ctx(
|
||||
dir.path(),
|
||||
&scope,
|
||||
vec!["Read".into(), "Edit".into(), "MemoryRead".into()],
|
||||
vec!["Read".into(), "Edit".into(), "MemoryReadDocument".into()],
|
||||
None,
|
||||
))
|
||||
.unwrap();
|
||||
|
||||
@@ -10,6 +10,7 @@ pub mod companion;
|
||||
pub mod config;
|
||||
pub mod hosts;
|
||||
pub mod identity;
|
||||
pub mod memory_backend;
|
||||
pub mod memory_staging;
|
||||
pub mod observation;
|
||||
pub mod profile_settings;
|
||||
|
||||
@@ -0,0 +1,485 @@
|
||||
use memory::backend::{
|
||||
MemoryBackendAckOutput, MemoryBackendOperation, MemoryBackendOperationResult,
|
||||
MemoryDocumentReadOperation, MemoryDocumentUpdateOperation, MemoryQueryOperation,
|
||||
MemoryReadOperation, MemoryStagingCloseAction, MemoryStagingCloseOperation,
|
||||
MemoryStagingListOperation, MemoryStagingReadOperation, MemoryStagingWriteOutput,
|
||||
MemoryToolOutput,
|
||||
};
|
||||
use memory::extract::{ExtractedCandidate, StagingEvidence, StagingRecord};
|
||||
use memory::schema::{SourceEvidenceRef, SourceRef};
|
||||
use serde_json::json;
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::authority::MemoryAuthority;
|
||||
use crate::{Error, Result};
|
||||
|
||||
const DEFAULT_READ_LIMIT: usize = 2000;
|
||||
const MAX_STAGING_LIST_LIMIT: usize = 100;
|
||||
|
||||
pub fn execute_memory_backend_operation_with_authority<A: MemoryAuthority>(
|
||||
authority: &A,
|
||||
operation: MemoryBackendOperation,
|
||||
) -> Result<MemoryBackendOperationResult> {
|
||||
match operation {
|
||||
MemoryBackendOperation::Query(operation) => {
|
||||
execute_query(authority, operation).map(MemoryBackendOperationResult::ToolOutput)
|
||||
}
|
||||
MemoryBackendOperation::ReadDocument(operation) => execute_read_document(authority, operation)
|
||||
.map(MemoryBackendOperationResult::ToolOutput),
|
||||
MemoryBackendOperation::UpdateDocument(operation) => {
|
||||
execute_update_document(authority, operation).map(MemoryBackendOperationResult::ToolOutput)
|
||||
}
|
||||
MemoryBackendOperation::ResidentSummary(_operation) => Ok(
|
||||
MemoryBackendOperationResult::ToolOutput(execute_resident_summary(authority)?),
|
||||
),
|
||||
MemoryBackendOperation::AppendAudit(_operation) => Ok(
|
||||
MemoryBackendOperationResult::Acknowledged(MemoryBackendAckOutput {
|
||||
summary: "memory audit event accepted by Workspace authority".to_string(),
|
||||
}),
|
||||
),
|
||||
MemoryBackendOperation::StageCandidate(operation) => {
|
||||
let id = stage_candidate(
|
||||
authority,
|
||||
operation.source,
|
||||
operation.extract_run_id,
|
||||
operation.candidate,
|
||||
operation.evidence,
|
||||
operation.source_refs,
|
||||
)?;
|
||||
Ok(MemoryBackendOperationResult::StagingWritten(
|
||||
MemoryStagingWriteOutput {
|
||||
staging_count: 1,
|
||||
staging_ids: vec![id],
|
||||
},
|
||||
))
|
||||
}
|
||||
MemoryBackendOperation::StageExtracted(operation) => {
|
||||
let extract_run_id = Uuid::now_v7().to_string();
|
||||
let mut ids = Vec::with_capacity(operation.payload.candidates.len());
|
||||
for candidate in operation.payload.candidates {
|
||||
ids.push(stage_candidate(
|
||||
authority,
|
||||
operation.source.clone(),
|
||||
extract_run_id.clone(),
|
||||
candidate,
|
||||
Vec::new(),
|
||||
Vec::new(),
|
||||
)?);
|
||||
}
|
||||
Ok(MemoryBackendOperationResult::StagingWritten(
|
||||
MemoryStagingWriteOutput {
|
||||
staging_count: ids.len(),
|
||||
staging_ids: ids,
|
||||
},
|
||||
))
|
||||
}
|
||||
MemoryBackendOperation::StagingList(operation) => execute_staging_list(authority, operation)
|
||||
.map(MemoryBackendOperationResult::ToolOutput),
|
||||
MemoryBackendOperation::StagingRead(operation) => execute_staging_read(authority, operation)
|
||||
.map(MemoryBackendOperationResult::ToolOutput),
|
||||
MemoryBackendOperation::StagingClose(operation) => execute_staging_close(authority, operation)
|
||||
.map(MemoryBackendOperationResult::ToolOutput),
|
||||
MemoryBackendOperation::Read(operation) => execute_legacy_read(authority, operation)
|
||||
.map(MemoryBackendOperationResult::ToolOutput),
|
||||
MemoryBackendOperation::Write(_)
|
||||
| MemoryBackendOperation::Edit(_)
|
||||
| MemoryBackendOperation::Delete(_) => Err(Error::Store(
|
||||
"legacy summary/decision/request Memory mutation operations are not supported by Workspace authority; use MemoryUpdateDocument".to_string(),
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
fn execute_resident_summary<A: MemoryAuthority>(authority: &A) -> Result<MemoryToolOutput> {
|
||||
let document = authority.ensure_memory_document()?;
|
||||
if document.body_md.trim().is_empty() {
|
||||
Ok(MemoryToolOutput {
|
||||
summary: "resident memory summary unavailable".to_string(),
|
||||
content: None,
|
||||
})
|
||||
} else {
|
||||
Ok(MemoryToolOutput {
|
||||
summary: "resident memory document collected".to_string(),
|
||||
content: Some(document.body_md),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
fn execute_query<A: MemoryAuthority>(
|
||||
authority: &A,
|
||||
operation: MemoryQueryOperation,
|
||||
) -> Result<MemoryToolOutput> {
|
||||
let document = authority.ensure_memory_document()?;
|
||||
let query = operation
|
||||
.query
|
||||
.as_deref()
|
||||
.map(str::trim)
|
||||
.filter(|query| !query.is_empty());
|
||||
match query {
|
||||
Some(query) => {
|
||||
if let Some(body) = excerpt_matching(&document.body_md, query) {
|
||||
Ok(MemoryToolOutput {
|
||||
summary: format!("Memory document matched {query:?}"),
|
||||
content: Some(body),
|
||||
})
|
||||
} else {
|
||||
Ok(MemoryToolOutput {
|
||||
summary: format!("No memory document content matched {query:?}"),
|
||||
content: None,
|
||||
})
|
||||
}
|
||||
}
|
||||
None => Ok(MemoryToolOutput {
|
||||
summary: "Memory document overview".to_string(),
|
||||
content: Some(excerpt_head(&document.body_md, 120)),
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
fn execute_legacy_read<A: MemoryAuthority>(
|
||||
authority: &A,
|
||||
operation: MemoryReadOperation,
|
||||
) -> Result<MemoryToolOutput> {
|
||||
if operation.kind.as_str() != "summary" || operation.slug.is_some() {
|
||||
return Err(Error::Store(
|
||||
"legacy kind/slug Memory reads are not supported by Workspace authority; use MemoryReadDocument".to_string(),
|
||||
));
|
||||
}
|
||||
execute_read_document(
|
||||
authority,
|
||||
MemoryDocumentReadOperation {
|
||||
offset: operation.offset,
|
||||
limit: operation.limit,
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
fn execute_read_document<A: MemoryAuthority>(
|
||||
authority: &A,
|
||||
operation: MemoryDocumentReadOperation,
|
||||
) -> Result<MemoryToolOutput> {
|
||||
let document = authority.ensure_memory_document()?;
|
||||
let offset = operation.offset.unwrap_or(0);
|
||||
let limit = operation.limit.unwrap_or(DEFAULT_READ_LIMIT);
|
||||
let lines: Vec<&str> = document.body_md.lines().collect();
|
||||
let selected = lines
|
||||
.iter()
|
||||
.enumerate()
|
||||
.skip(offset)
|
||||
.take(limit)
|
||||
.map(|(index, line)| format!("{:>6}\t{}", index + 1, line))
|
||||
.collect::<Vec<_>>()
|
||||
.join("\n");
|
||||
let total = lines.len();
|
||||
Ok(MemoryToolOutput {
|
||||
summary: format!(
|
||||
"Read {} lines from Workspace memory document",
|
||||
selected.lines().count()
|
||||
),
|
||||
content: Some(format!(
|
||||
"total_lines={total} offset={offset} limit={limit}\n{selected}"
|
||||
)),
|
||||
})
|
||||
}
|
||||
|
||||
fn execute_update_document<A: MemoryAuthority>(
|
||||
authority: &A,
|
||||
operation: MemoryDocumentUpdateOperation,
|
||||
) -> Result<MemoryToolOutput> {
|
||||
if operation.body_md.trim().is_empty() {
|
||||
return Err(Error::Store(
|
||||
"memory document body_md must not be empty".to_string(),
|
||||
));
|
||||
}
|
||||
let document = authority.update_memory_document(&operation.body_md)?;
|
||||
Ok(MemoryToolOutput {
|
||||
summary: format!(
|
||||
"Updated Workspace memory document ({} bytes)",
|
||||
document.body_md.len()
|
||||
),
|
||||
content: Some(format!("updated_at={}", document.updated_at)),
|
||||
})
|
||||
}
|
||||
|
||||
fn stage_candidate<A: MemoryAuthority>(
|
||||
authority: &A,
|
||||
source: SourceRef,
|
||||
extract_run_id: String,
|
||||
candidate: ExtractedCandidate,
|
||||
evidence: Vec<StagingEvidence>,
|
||||
source_refs: Vec<SourceEvidenceRef>,
|
||||
) -> Result<String> {
|
||||
let candidate_id = Uuid::now_v7().to_string();
|
||||
let record = StagingRecord::from_candidate(
|
||||
candidate_id.clone(),
|
||||
extract_run_id,
|
||||
source,
|
||||
candidate,
|
||||
evidence,
|
||||
source_refs,
|
||||
);
|
||||
let raw_json = serde_json::to_string_pretty(&record)
|
||||
.map_err(|err| Error::Store(format!("serialize memory staging record: {err}")))?;
|
||||
authority.upsert_memory_staging_record(&candidate_id, &raw_json, None)?;
|
||||
Ok(candidate_id)
|
||||
}
|
||||
|
||||
fn execute_staging_list<A: MemoryAuthority>(
|
||||
authority: &A,
|
||||
operation: MemoryStagingListOperation,
|
||||
) -> Result<MemoryToolOutput> {
|
||||
let limit = operation.limit.unwrap_or(20).min(MAX_STAGING_LIST_LIMIT);
|
||||
let entries = authority.list_memory_staging_records(limit)?;
|
||||
let records = entries
|
||||
.iter()
|
||||
.map(|entry| {
|
||||
staging_summary_json(
|
||||
&entry.candidate_id,
|
||||
&entry.raw_json,
|
||||
entry.source_path.as_deref(),
|
||||
)
|
||||
})
|
||||
.collect::<Result<Vec<_>>>()?;
|
||||
Ok(MemoryToolOutput {
|
||||
summary: format!(
|
||||
"Listed {} memory staging candidate(s) from Workspace authority",
|
||||
records.len()
|
||||
),
|
||||
content: Some(
|
||||
serde_json::to_string_pretty(&records)
|
||||
.map_err(|err| Error::Store(format!("serialize memory staging list: {err}")))?,
|
||||
),
|
||||
})
|
||||
}
|
||||
|
||||
fn execute_staging_read<A: MemoryAuthority>(
|
||||
authority: &A,
|
||||
operation: MemoryStagingReadOperation,
|
||||
) -> Result<MemoryToolOutput> {
|
||||
let entry = authority.memory_staging_record(&operation.candidate_id)?;
|
||||
let raw: serde_json::Value = serde_json::from_str(&entry.raw_json)
|
||||
.map_err(|err| Error::Store(format!("decode memory staging record: {err}")))?;
|
||||
Ok(MemoryToolOutput {
|
||||
summary: format!("Read memory staging candidate {}", entry.candidate_id),
|
||||
content: Some(
|
||||
serde_json::to_string_pretty(&raw)
|
||||
.map_err(|err| Error::Store(format!("serialize memory staging record: {err}")))?,
|
||||
),
|
||||
})
|
||||
}
|
||||
|
||||
fn execute_staging_close<A: MemoryAuthority>(
|
||||
authority: &A,
|
||||
operation: MemoryStagingCloseOperation,
|
||||
) -> Result<MemoryToolOutput> {
|
||||
if operation.reason.trim().is_empty() {
|
||||
return Err(Error::Store("reason is required".to_string()));
|
||||
}
|
||||
let affected_refs_json = serde_json::to_string(&operation.affected_memory)
|
||||
.map_err(|err| Error::Store(format!("serialize affected memory refs: {err}")))?;
|
||||
let action = staging_close_action_name(&operation.action);
|
||||
let resolution = authority.close_memory_staging_record(
|
||||
&operation.candidate_id,
|
||||
action,
|
||||
&operation.reason,
|
||||
&affected_refs_json,
|
||||
)?;
|
||||
Ok(MemoryToolOutput {
|
||||
summary: format!("Closed memory staging candidate {}", resolution.candidate_id),
|
||||
content: Some(
|
||||
serde_json::to_string_pretty(&json!({
|
||||
"candidate_id": resolution.candidate_id,
|
||||
"action": resolution.action,
|
||||
"reason": resolution.reason,
|
||||
"affected_refs": serde_json::from_str::<serde_json::Value>(&resolution.affected_refs_json).unwrap_or_else(|_| json!([])),
|
||||
"resolved_at": resolution.resolved_at,
|
||||
"record_authority": resolution.record_source,
|
||||
}))
|
||||
.map_err(|err| Error::Store(format!("serialize memory staging resolution: {err}")))?,
|
||||
),
|
||||
})
|
||||
}
|
||||
|
||||
fn staging_close_action_name(action: &MemoryStagingCloseAction) -> &'static str {
|
||||
match action {
|
||||
MemoryStagingCloseAction::Applied => "applied",
|
||||
MemoryStagingCloseAction::Discarded => "discarded",
|
||||
MemoryStagingCloseAction::Invalid => "invalid",
|
||||
MemoryStagingCloseAction::Duplicate => "duplicate",
|
||||
MemoryStagingCloseAction::AlreadyCovered => "already_covered",
|
||||
}
|
||||
}
|
||||
|
||||
fn staging_summary_json(
|
||||
candidate_id: &str,
|
||||
raw_json: &str,
|
||||
source_path: Option<&str>,
|
||||
) -> Result<serde_json::Value> {
|
||||
let raw: serde_json::Value = serde_json::from_str(raw_json).map_err(|err| {
|
||||
Error::Store(format!(
|
||||
"decode memory staging record {candidate_id}: {err}"
|
||||
))
|
||||
})?;
|
||||
Ok(json!({
|
||||
"candidate_id": candidate_id,
|
||||
"bytes": raw_json.len(),
|
||||
"source_path": source_path,
|
||||
"source": raw.get("source").cloned().unwrap_or(serde_json::Value::Null),
|
||||
"kind": raw.get("kind").cloned().unwrap_or(serde_json::Value::Null),
|
||||
"claim": raw.get("claim").cloned().unwrap_or(serde_json::Value::Null),
|
||||
"why_useful": raw.get("why_useful").cloned().unwrap_or(serde_json::Value::Null),
|
||||
"staleness": raw.get("staleness").cloned().unwrap_or(serde_json::Value::Null),
|
||||
"evidence_count": raw.get("evidence").and_then(|value| value.as_array()).map_or(0, Vec::len),
|
||||
"source_ref_count": raw.get("source_refs").and_then(|value| value.as_array()).map_or(0, Vec::len),
|
||||
}))
|
||||
}
|
||||
|
||||
fn excerpt_head(body: &str, max_lines: usize) -> String {
|
||||
let mut lines = body.lines().take(max_lines).collect::<Vec<_>>().join("\n");
|
||||
if body.lines().count() > max_lines {
|
||||
lines.push_str("\n... [truncated]");
|
||||
}
|
||||
lines
|
||||
}
|
||||
|
||||
fn excerpt_matching(body: &str, query: &str) -> Option<String> {
|
||||
let query_lower = query.to_lowercase();
|
||||
let lines = body.lines().collect::<Vec<_>>();
|
||||
let index = lines
|
||||
.iter()
|
||||
.position(|line| line.to_lowercase().contains(&query_lower))?;
|
||||
let start = index.saturating_sub(3);
|
||||
let end = (index + 4).min(lines.len());
|
||||
Some(
|
||||
lines[start..end]
|
||||
.iter()
|
||||
.enumerate()
|
||||
.map(|(offset, line)| format!("{:>6}\t{}", start + offset + 1, line))
|
||||
.collect::<Vec<_>>()
|
||||
.join("\n"),
|
||||
)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::authority::SqliteWorkspaceAuthority;
|
||||
use memory::backend::{
|
||||
MemoryDocumentReadOperation, MemoryDocumentUpdateOperation, MemoryStagingCloseAction,
|
||||
MemoryStagingCloseOperation, MemoryStagingListOperation,
|
||||
};
|
||||
use memory::extract::{CandidateKind, ExtractedCandidate};
|
||||
use memory::schema::SourceRef;
|
||||
use tempfile::TempDir;
|
||||
|
||||
fn authority(temp: &TempDir) -> SqliteWorkspaceAuthority {
|
||||
let db_path = temp.path().join("workspace.sqlite3");
|
||||
let authority = SqliteWorkspaceAuthority::new(&db_path, "workspace").unwrap();
|
||||
let conn = rusqlite::Connection::open(&db_path).unwrap();
|
||||
conn.execute(
|
||||
"INSERT INTO workspaces (
|
||||
workspace_id, owner_account_id, display_name, state, created_at, updated_at
|
||||
) VALUES (?1, NULL, ?2, ?3, ?4, ?4)",
|
||||
rusqlite::params!["workspace", "Workspace", "active", "2026-01-01T00:00:00Z"],
|
||||
)
|
||||
.unwrap();
|
||||
authority
|
||||
}
|
||||
|
||||
fn source() -> SourceRef {
|
||||
SourceRef {
|
||||
segment_id: "segment-1".into(),
|
||||
range: [1, 2],
|
||||
}
|
||||
}
|
||||
|
||||
fn candidate() -> ExtractedCandidate {
|
||||
ExtractedCandidate {
|
||||
kind: CandidateKind::Decision,
|
||||
claim: "Use SQLite memory authority".into(),
|
||||
why_useful: "Keeps Worker tools away from .yoi/memory filesystem authority".into(),
|
||||
staleness: None,
|
||||
evidence_ids: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn document_operations_use_workspace_authority() {
|
||||
let temp = TempDir::new().unwrap();
|
||||
let authority = authority(&temp);
|
||||
execute_memory_backend_operation_with_authority(
|
||||
&authority,
|
||||
MemoryBackendOperation::UpdateDocument(MemoryDocumentUpdateOperation {
|
||||
body_md: "# Memory\n\nSQLite backed body".into(),
|
||||
}),
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
let result = execute_memory_backend_operation_with_authority(
|
||||
&authority,
|
||||
MemoryBackendOperation::ReadDocument(MemoryDocumentReadOperation {
|
||||
offset: None,
|
||||
limit: None,
|
||||
}),
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
let MemoryBackendOperationResult::ToolOutput(output) = result else {
|
||||
panic!("expected tool output")
|
||||
};
|
||||
assert!(output.content.unwrap().contains("SQLite backed body"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn staging_close_moves_record_to_workspace_resolution() {
|
||||
let temp = TempDir::new().unwrap();
|
||||
let authority = authority(&temp);
|
||||
let result = execute_memory_backend_operation_with_authority(
|
||||
&authority,
|
||||
MemoryBackendOperation::StageCandidate(
|
||||
memory::backend::MemoryStageCandidateOperation {
|
||||
source: source(),
|
||||
extract_run_id: "run-1".into(),
|
||||
candidate: candidate(),
|
||||
evidence: Vec::new(),
|
||||
source_refs: Vec::new(),
|
||||
},
|
||||
),
|
||||
)
|
||||
.unwrap();
|
||||
let MemoryBackendOperationResult::StagingWritten(written) = result else {
|
||||
panic!("expected staging write")
|
||||
};
|
||||
assert_eq!(written.staging_count, 1);
|
||||
|
||||
let list = execute_memory_backend_operation_with_authority(
|
||||
&authority,
|
||||
MemoryBackendOperation::StagingList(MemoryStagingListOperation { limit: None }),
|
||||
)
|
||||
.unwrap();
|
||||
let MemoryBackendOperationResult::ToolOutput(list) = list else {
|
||||
panic!("expected list")
|
||||
};
|
||||
assert!(
|
||||
list.content
|
||||
.unwrap()
|
||||
.contains("Use SQLite memory authority")
|
||||
);
|
||||
|
||||
execute_memory_backend_operation_with_authority(
|
||||
&authority,
|
||||
MemoryBackendOperation::StagingClose(MemoryStagingCloseOperation {
|
||||
candidate_id: written.staging_ids[0].clone(),
|
||||
action: MemoryStagingCloseAction::Applied,
|
||||
reason: "merged into document".into(),
|
||||
affected_memory: Vec::new(),
|
||||
}),
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(authority.list_memory_staging_records(10).unwrap().len(), 0);
|
||||
assert_eq!(
|
||||
authority.list_memory_staging_resolutions(10).unwrap().len(),
|
||||
1
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -4,6 +4,9 @@ use memory::extract::StagingRecord;
|
||||
use memory::schema::{SourceEvidenceRef, SourceRef};
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use crate::Result;
|
||||
use crate::authority::MemoryAuthority;
|
||||
|
||||
const DEFAULT_MEMORY_STAGING_LIMIT: usize = 100;
|
||||
const MAX_MEMORY_STAGING_LIMIT: usize = 500;
|
||||
|
||||
@@ -88,6 +91,48 @@ pub fn list_memory_staging(
|
||||
}
|
||||
}
|
||||
|
||||
pub fn list_memory_staging_from_authority<A: MemoryAuthority>(
|
||||
authority: &A,
|
||||
requested_limit: Option<usize>,
|
||||
) -> Result<MemoryStagingListResponse> {
|
||||
let limit = requested_limit
|
||||
.unwrap_or(DEFAULT_MEMORY_STAGING_LIMIT)
|
||||
.min(MAX_MEMORY_STAGING_LIMIT);
|
||||
let entries = authority.list_memory_staging_records(MAX_MEMORY_STAGING_LIMIT + 1)?;
|
||||
let fetched_count = entries.len();
|
||||
let mut invalid_count = 0usize;
|
||||
let mut total_valid_count = 0usize;
|
||||
let mut valid_items = Vec::with_capacity(fetched_count.min(limit));
|
||||
for entry in entries {
|
||||
let record = match serde_json::from_str::<StagingRecord>(&entry.raw_json) {
|
||||
Ok(record) => record,
|
||||
Err(_) => {
|
||||
invalid_count += 1;
|
||||
continue;
|
||||
}
|
||||
};
|
||||
total_valid_count += 1;
|
||||
if valid_items.len() < limit {
|
||||
valid_items.push(MemoryStagingEntrySummary {
|
||||
id: entry.candidate_id,
|
||||
byte_len: entry.raw_json.len() as u64,
|
||||
record: memory_staging_record_summary(record),
|
||||
});
|
||||
}
|
||||
}
|
||||
let returned_count = valid_items.len();
|
||||
Ok(MemoryStagingListResponse {
|
||||
limit,
|
||||
returned_count,
|
||||
total_valid_count,
|
||||
invalid_count,
|
||||
truncated: total_valid_count > returned_count || fetched_count > MAX_MEMORY_STAGING_LIMIT,
|
||||
order: "uuidv7_ascending".to_string(),
|
||||
record_authority: "sqlite_workspace_authority.memory_staging".to_string(),
|
||||
items: valid_items,
|
||||
})
|
||||
}
|
||||
|
||||
fn memory_staging_entry_summary(entry: StagingEntry) -> MemoryStagingEntrySummary {
|
||||
MemoryStagingEntrySummary {
|
||||
id: entry.id.to_string(),
|
||||
|
||||
@@ -14,7 +14,7 @@ use chrono::{Duration, SecondsFormat, Utc};
|
||||
use futures::{SinkExt, StreamExt};
|
||||
use memory::backend::{
|
||||
MemoryBackendHttpResponse, MemoryBackendOperation, MemoryConsolidateStagingOperation,
|
||||
MemoryConsolidationOutput, execute_memory_backend_operation,
|
||||
MemoryConsolidationOutput,
|
||||
};
|
||||
use protocol::stream::{decode_method, encode_event};
|
||||
use serde::{Deserialize, Serialize};
|
||||
@@ -58,7 +58,8 @@ use crate::hosts::{
|
||||
WorkerSpawnResult, WorkerSpawnWorkingDirectoryRequest, WorkerSummary, WorkerWorkspaceSummary,
|
||||
};
|
||||
use crate::identity::WorkspaceIdentity;
|
||||
use crate::memory_staging::{MemoryStagingListResponse, list_memory_staging};
|
||||
use crate::memory_backend::execute_memory_backend_operation_with_authority;
|
||||
use crate::memory_staging::{MemoryStagingListResponse, list_memory_staging_from_authority};
|
||||
use crate::observation::{
|
||||
BackendObservationProxy, ObservationProxyError, RuntimeObservationClient,
|
||||
RuntimeObservationSource, RuntimeObservationSourceConfig,
|
||||
@@ -1458,9 +1459,10 @@ async fn scoped_list_memory_staging(
|
||||
Query(query): Query<MemoryStagingQuery>,
|
||||
) -> ApiResult<Json<MemoryStagingListResponse>> {
|
||||
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(list_memory_staging(&layout, query.limit)))
|
||||
Ok(Json(list_memory_staging_from_authority(
|
||||
&api.authority,
|
||||
query.limit,
|
||||
)?))
|
||||
}
|
||||
|
||||
async fn scoped_memory_backend_operation(
|
||||
@@ -1469,9 +1471,8 @@ async fn scoped_memory_backend_operation(
|
||||
Json(operation): Json<MemoryBackendOperation>,
|
||||
) -> ApiResult<Json<MemoryBackendHttpResponse>> {
|
||||
validate_workspace_scope(&api, &path.workspace_id)?;
|
||||
let memory_config = manifest::MemoryConfig::default();
|
||||
let layout = memory::WorkspaceLayout::resolve(&memory_config, &api.config.workspace_root);
|
||||
let response = match execute_memory_backend_operation(&layout, operation) {
|
||||
let response = match execute_memory_backend_operation_with_authority(&api.authority, operation)
|
||||
{
|
||||
Ok(result) => MemoryBackendHttpResponse::Ok { result },
|
||||
Err(error) => MemoryBackendHttpResponse::Error {
|
||||
message: sanitize_backend_error(&error.to_string()),
|
||||
|
||||
Reference in New Issue
Block a user