From 7e43c10e2df88ec62de18591f5452498ae5a2fd6 Mon Sep 17 00:00:00 2001 From: Hare Date: Sat, 18 Jul 2026 10:35:59 +0900 Subject: [PATCH] worker: implement session explore extract worker --- .yoi/tickets/00001KXMK7YMC/item.md | 6 +- .yoi/tickets/00001KXMK7YMC/resolution.md | 10 + .yoi/tickets/00001KXMK7YMC/thread.md | 97 +++ crates/memory/src/extract/mod.rs | 2 +- crates/memory/src/extract/staging.rs | 46 +- crates/worker/src/compact/worker.rs | 3 +- crates/worker/src/feature/builtin.rs | 4 + .../src/feature/builtin/session_explore.rs | 673 ++++++++++++++++++ crates/worker/src/session_reference.rs | 50 +- crates/worker/src/worker.rs | 86 ++- crates/worker/tests/compact_events_test.rs | 20 +- .../prompts/internal/memory_extract_system.md | 45 +- 12 files changed, 959 insertions(+), 83 deletions(-) create mode 100644 .yoi/tickets/00001KXMK7YMC/resolution.md create mode 100644 crates/worker/src/feature/builtin/session_explore.rs diff --git a/.yoi/tickets/00001KXMK7YMC/item.md b/.yoi/tickets/00001KXMK7YMC/item.md index 6457f158..302f9832 100644 --- a/.yoi/tickets/00001KXMK7YMC/item.md +++ b/.yoi/tickets/00001KXMK7YMC/item.md @@ -1,9 +1,11 @@ --- title: 'session-explore feature付きextract workerを実装する' -state: 'planning' +state: 'closed' created_at: '2026-07-16T04:34:01Z' -updated_at: '2026-07-17T18:10:00Z' +updated_at: '2026-07-18T01:35:46Z' assignee: null +queued_by: 'yoi ticket' +queued_at: '2026-07-18T01:12:22Z' --- ## 背景 diff --git a/.yoi/tickets/00001KXMK7YMC/resolution.md b/.yoi/tickets/00001KXMK7YMC/resolution.md new file mode 100644 index 00000000..77a849a5 --- /dev/null +++ b/.yoi/tickets/00001KXMK7YMC/resolution.md @@ -0,0 +1,10 @@ +Implemented the session-explore-backed extract worker. + +The extract worker now receives an immutable `SessionReferenceView` snapshot and only the `builtin:session-explore` tool surface. It can search/read bounded evidence, stage one flat candidate per `stage_candidate` call with resolved source anchors, and finish cleanly with `finish_extraction` including the no-candidate path. + +Validation passed: +- `cargo fmt --check` +- `cargo test -p worker` +- `cargo test -p memory` +- `nix build .#yoi` +- `yoi ticket doctor` diff --git a/.yoi/tickets/00001KXMK7YMC/thread.md b/.yoi/tickets/00001KXMK7YMC/thread.md index 59a4b92a..f478e7f1 100644 --- a/.yoi/tickets/00001KXMK7YMC/thread.md +++ b/.yoi/tickets/00001KXMK7YMC/thread.md @@ -57,4 +57,101 @@ Rescoped from broad Overview-first extract implementation to the worker implemen Design updated: abandon the old batch `write_extracted` / `decisions` / `discussions` / `attempts` / `requests` staging direction. Extract now stages flat candidate records with `stage_candidate` and finishes with `finish_extraction`. The older dependency on 00001KXNYXNM6 is historical; the active schema dependency is 00001KXRM6G0G. +--- + + + +## Intake summary + +Marked ready by `yoi ticket state`. + +--- + + + +## State changed + +Marked ready by `yoi ticket state`. + + +--- + + + +## State changed + +Ticket を `yoi ticket` が queued にしました。 + + +--- + + + +## State changed + +State changed to `inprogress`. + + +--- + + + +## Implementation report + +Implemented the session-explore-backed memory extract worker. + +Changes: +- Added `builtin:session-explore` with `search_evidence`, `read_evidence`, `stage_candidate`, and `finish_extraction`. +- Extract worker now builds an immutable `SessionReferenceView` at extract start and enables only session-explore tools for the internal worker. +- Added overview-first / evidence-index-second extract input rendering over the reference view. +- `stage_candidate` writes one flat staging record per call, resolving evidence ids into bounded staging evidence and `SourceEvidenceRef` anchors. +- `finish_extraction` supports normal no-candidate completion with `staged_count = 0`. +- Updated `memory_extract_system.md` to the session-explore / stage / finish contract. +- Added a single-candidate staging helper in `memory::extract::staging`. +- Updated extract-trigger tests to use `finish_extraction` and added session-explore unit tests. + +Validation: +- `cargo fmt --check` +- `cargo test -p worker` +- `cargo test -p memory` +- `nix build .#yoi` +- `yoi ticket doctor` + + +--- + + + +## State changed + +State changed to `done`. + + +--- + + + +## State changed + +Ticket を closed にしました。 + + +--- + + + +## 完了 + +Implemented the session-explore-backed extract worker. + +The extract worker now receives an immutable `SessionReferenceView` snapshot and only the `builtin:session-explore` tool surface. It can search/read bounded evidence, stage one flat candidate per `stage_candidate` call with resolved source anchors, and finish cleanly with `finish_extraction` including the no-candidate path. + +Validation passed: +- `cargo fmt --check` +- `cargo test -p worker` +- `cargo test -p memory` +- `nix build .#yoi` +- `yoi ticket doctor` + + --- diff --git a/crates/memory/src/extract/mod.rs b/crates/memory/src/extract/mod.rs index 72fb3bf0..a320dbc9 100644 --- a/crates/memory/src/extract/mod.rs +++ b/crates/memory/src/extract/mod.rs @@ -28,7 +28,7 @@ pub use payload::{ StagingRecord, }; pub use pointer::{ExtractPointerPayload, fold_pointer}; -pub use staging::{StagingWriteResult, write_staging}; +pub use staging::{StagingWriteResult, write_staging, write_staging_candidate}; pub use tool::{ExtractWorkerContext, write_extracted_tool}; /// session-store `LogEntry::Extension` で使う domain 名。 diff --git a/crates/memory/src/extract/staging.rs b/crates/memory/src/extract/staging.rs index 3a8766e2..8c8a6c1f 100644 --- a/crates/memory/src/extract/staging.rs +++ b/crates/memory/src/extract/staging.rs @@ -11,8 +11,10 @@ use std::path::PathBuf; use uuid::Uuid; -use crate::extract::payload::{ExtractedPayload, StagingRecord}; -use crate::schema::SourceRef; +use crate::extract::payload::{ + ExtractedCandidate, ExtractedPayload, StagingEvidence, StagingRecord, +}; +use crate::schema::{SourceEvidenceRef, SourceRef}; use crate::workspace::WorkspaceLayout; /// Filesystem result for a single staged candidate. @@ -34,30 +36,48 @@ pub fn write_staging( return Ok(Vec::new()); } - let dir = layout.staging_dir(); - fs::create_dir_all(&dir)?; let extract_run_id = Uuid::now_v7().to_string(); let mut written = Vec::with_capacity(payload.candidates.len()); for candidate in payload.candidates { - let id = Uuid::now_v7(); - let record = StagingRecord::from_candidate( - id.to_string(), - extract_run_id.clone(), + written.push(write_staging_candidate( + layout, source.clone(), + &extract_run_id, candidate, Vec::new(), Vec::new(), - ); - let path = dir.join(format!("{}.json", id)); - let bytes = serde_json::to_vec_pretty(&record).map_err(io::Error::other)?; - fs::write(&path, bytes)?; - written.push(StagingWriteResult { id, path }); + )?); } Ok(written) } +pub fn write_staging_candidate( + layout: &WorkspaceLayout, + source: SourceRef, + extract_run_id: &str, + candidate: ExtractedCandidate, + evidence: Vec, + source_refs: Vec, +) -> io::Result { + let dir = layout.staging_dir(); + fs::create_dir_all(&dir)?; + let id = Uuid::now_v7(); + let record = StagingRecord::from_candidate( + id.to_string(), + extract_run_id.to_string(), + source, + candidate, + evidence, + source_refs, + ); + let path = dir.join(format!("{}.json", id)); + let bytes = serde_json::to_vec_pretty(&record).map_err(io::Error::other)?; + fs::write(&path, bytes)?; + Ok(StagingWriteResult { id, path }) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/worker/src/compact/worker.rs b/crates/worker/src/compact/worker.rs index 35245c29..4b4c1099 100644 --- a/crates/worker/src/compact/worker.rs +++ b/crates/worker/src/compact/worker.rs @@ -31,8 +31,7 @@ use tools::ScopedFs; use crate::compact::usage_tracker::UsageTracker; use crate::fs_view::{ReadRequirement, slice_lines}; use crate::session_reference::{ - ReadDetail, ReadOptions, ReadSelector, ReferenceKind, SearchOptions, SessionReferenceView, - ToolPart, + ReadDetail, ReadOptions, ReadSelector, SearchOptions, SessionReferenceView, ToolPart, }; /// Aggregated output of a compact worker run. diff --git a/crates/worker/src/feature/builtin.rs b/crates/worker/src/feature/builtin.rs index 70ae7cf2..82c35226 100644 --- a/crates/worker/src/feature/builtin.rs +++ b/crates/worker/src/feature/builtin.rs @@ -4,9 +4,13 @@ //! same descriptor-approved registry path used by feature modules. They are not //! an external plugin-loading surface. +pub mod session_explore; pub mod task; pub mod ticket; +pub(crate) use session_explore::{ + SessionExploreFeature, SessionExploreState, render_extract_input, +}; pub use task::{TaskFeature, task_tools_feature}; pub use ticket::{ TicketFeature, TicketFeatureAccess, ticket_tools_feature, ticket_tools_feature_with_access, diff --git a/crates/worker/src/feature/builtin/session_explore.rs b/crates/worker/src/feature/builtin/session_explore.rs new file mode 100644 index 00000000..b7153b29 --- /dev/null +++ b/crates/worker/src/feature/builtin/session_explore.rs @@ -0,0 +1,673 @@ +use std::sync::{Arc, Mutex}; + +use async_trait::async_trait; +use llm_engine::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput}; +use memory::extract::{ + CandidateKind, ExtractedCandidate, StagingWriteResult, write_staging_candidate, +}; +use memory::schema::SourceRef; +use memory::workspace::WorkspaceLayout; +use schemars::JsonSchema; +use serde::Deserialize; +use uuid::Uuid; + +use crate::feature::{ + FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureModule, ToolContribution, + ToolDeclaration, +}; +use crate::session_reference::{ + ReadDetail, ReadOptions, ReadSelector, ReferenceKind, SearchOptions, SessionReferenceView, + ToolPart, +}; + +const SEARCH_EVIDENCE_DESCRIPTION: &str = "Search the host-created session evidence index. Use this to find stable evidence ids before staging a memory candidate. Supports kind=user|assistant|system|tool and tool_part=input|output|both."; +const READ_EVIDENCE_DESCRIPTION: &str = "Read bounded session evidence by evidence_id or entry_range. Use compact mode for normal verification and full mode only when exact tool arguments or result content are necessary."; +const STAGE_CANDIDATE_DESCRIPTION: &str = "Stage one memory candidate as a flat staging record. The candidate must cite one or more evidence_ids returned by search_evidence/read_evidence."; +const FINISH_EXTRACTION_DESCRIPTION: &str = "Finish the extract worker run after all useful candidates have been staged, or state why no candidates were useful."; + +#[derive(Clone)] +pub(crate) struct SessionExploreState { + view: Arc, + layout: WorkspaceLayout, + source: SourceRef, + extract_run_id: String, + staged: Arc>>, + finished: Arc>>, +} + +impl SessionExploreState { + pub(crate) fn new( + view: SessionReferenceView, + layout: WorkspaceLayout, + source: SourceRef, + ) -> Self { + Self { + view: Arc::new(view), + layout, + source, + extract_run_id: Uuid::now_v7().to_string(), + staged: Arc::new(Mutex::new(Vec::new())), + finished: Arc::new(Mutex::new(None)), + } + } + + pub(crate) fn view(&self) -> &SessionReferenceView { + &self.view + } + + pub(crate) fn staged(&self) -> Vec { + self.staged + .lock() + .expect("session explore staged state poisoned") + .clone() + } + + pub(crate) fn is_finished(&self) -> bool { + self.finished + .lock() + .expect("session explore finished state poisoned") + .is_some() + } +} + +#[derive(Clone)] +pub(crate) struct SessionExploreFeature { + state: SessionExploreState, +} + +impl SessionExploreFeature { + pub(crate) fn new(state: SessionExploreState) -> Self { + Self { state } + } +} + +impl FeatureModule for SessionExploreFeature { + fn descriptor(&self) -> FeatureDescriptor { + FeatureDescriptor::builtin("session-explore", "Session Explore") + .with_description( + "Host-bounded session evidence search/read tools plus explicit extraction staging.", + ) + .with_tool(ToolDeclaration::new( + "search_evidence", + SEARCH_EVIDENCE_DESCRIPTION, + )) + .with_tool(ToolDeclaration::new( + "read_evidence", + READ_EVIDENCE_DESCRIPTION, + )) + .with_tool(ToolDeclaration::new( + "stage_candidate", + STAGE_CANDIDATE_DESCRIPTION, + )) + .with_tool(ToolDeclaration::new( + "finish_extraction", + FINISH_EXTRACTION_DESCRIPTION, + )) + } + + fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> { + context.tools().register(ToolContribution::new( + "search_evidence", + search_evidence_definition(self.state.clone()), + ))?; + context.tools().register(ToolContribution::new( + "read_evidence", + read_evidence_definition(self.state.clone()), + ))?; + context.tools().register(ToolContribution::new( + "stage_candidate", + stage_candidate_definition(self.state.clone()), + ))?; + context.tools().register(ToolContribution::new( + "finish_extraction", + finish_extraction_definition(self.state.clone()), + ))?; + Ok(()) + } +} + +fn search_evidence_definition(state: SessionExploreState) -> ToolDefinition { + Arc::new(move || { + let schema = schemars::schema_for!(SearchEvidenceParams); + let schema_value = serde_json::to_value(schema).unwrap_or(serde_json::json!({})); + let meta = ToolMeta::new("search_evidence") + .description(SEARCH_EVIDENCE_DESCRIPTION) + .input_schema(schema_value); + let tool: Arc = Arc::new(SearchEvidenceTool { + state: state.clone(), + }); + (meta, tool) + }) +} + +fn read_evidence_definition(state: SessionExploreState) -> ToolDefinition { + Arc::new(move || { + let schema = schemars::schema_for!(ReadEvidenceParams); + let schema_value = serde_json::to_value(schema).unwrap_or(serde_json::json!({})); + let meta = ToolMeta::new("read_evidence") + .description(READ_EVIDENCE_DESCRIPTION) + .input_schema(schema_value); + let tool: Arc = Arc::new(ReadEvidenceTool { + state: state.clone(), + }); + (meta, tool) + }) +} + +fn stage_candidate_definition(state: SessionExploreState) -> ToolDefinition { + Arc::new(move || { + let schema = schemars::schema_for!(StageCandidateParams); + let schema_value = serde_json::to_value(schema).unwrap_or(serde_json::json!({})); + let meta = ToolMeta::new("stage_candidate") + .description(STAGE_CANDIDATE_DESCRIPTION) + .input_schema(schema_value); + let tool: Arc = Arc::new(StageCandidateTool { + state: state.clone(), + }); + (meta, tool) + }) +} + +fn finish_extraction_definition(state: SessionExploreState) -> ToolDefinition { + Arc::new(move || { + let schema = schemars::schema_for!(FinishExtractionParams); + let schema_value = serde_json::to_value(schema).unwrap_or(serde_json::json!({})); + let meta = ToolMeta::new("finish_extraction") + .description(FINISH_EXTRACTION_DESCRIPTION) + .input_schema(schema_value); + let tool: Arc = Arc::new(FinishExtractionTool { + state: state.clone(), + }); + (meta, tool) + }) +} + +#[derive(Debug, Deserialize, JsonSchema)] +struct SearchEvidenceParams { + /// Case-insensitive substring. Empty query lists bounded index entries matching filters. + #[serde(default)] + query: String, + /// Optional evidence kind: user, assistant, system, tool. + #[serde(default)] + kind: Option, + /// Optional tool part filter for tool evidence: input, output, both. + #[serde(default)] + tool_part: Option, + /// Optional tool name filter for tool input entries. + #[serde(default)] + tool_name: Option, + /// 0-based session item offset to start from. + #[serde(default)] + offset: Option, + /// Maximum number of hits. + #[serde(default)] + limit: Option, +} + +#[derive(Debug, Deserialize, JsonSchema)] +struct ReadEvidenceParams { + /// Evidence id returned by search_evidence, e.g. M0001, T0002i, T0003o. + #[serde(default)] + evidence_id: Option, + /// Inclusive 0-based session item range. Use when search returned an entry_range instead of a single id. + #[serde(default)] + entry_range: Option<[u64; 2]>, + /// compact omits tool arguments/results; full includes them within host bounds. + #[serde(default = "default_read_mode")] + mode: String, + /// Maximum entries to return. + #[serde(default)] + max_items: Option, +} + +fn default_read_mode() -> String { + "compact".to_string() +} + +#[derive(Debug, Deserialize, JsonSchema)] +struct StageCandidateParams { + kind: CandidateKind, + claim: String, + why_useful: String, + #[serde(default)] + staleness: Option, + #[serde(default)] + evidence_ids: Vec, +} + +#[derive(Debug, Clone, Deserialize, JsonSchema)] +struct FinishExtractionParams { + staged_count: usize, + #[serde(default)] + no_candidates_reason: Option, +} + +struct SearchEvidenceTool { + state: SessionExploreState, +} + +#[async_trait] +impl Tool for SearchEvidenceTool { + async fn execute( + &self, + input_json: &str, + _ctx: llm_engine::tool::ToolExecutionContext, + ) -> Result { + let params: SearchEvidenceParams = serde_json::from_str(input_json).map_err(|e| { + ToolError::InvalidArgument(format!("invalid search_evidence input: {e}")) + })?; + let kind = params + .kind + .as_deref() + .map(parse_reference_kind) + .transpose()?; + let tool_part = params + .tool_part + .as_deref() + .map(parse_tool_part) + .transpose()?; + let hits = self.state.view.search(&SearchOptions { + query: params.query, + kind, + tool_part, + tool_name: params.tool_name, + limit: params.limit, + min_entry_index: params.offset.map(|offset| offset as u64), + }); + let content = hits + .iter() + .map(|hit| { + let part = hit + .tool_part + .map(|part| format!(" {part:?}")) + .unwrap_or_default(); + let tool = hit + .tool_name + .as_ref() + .map(|name| format!(" {name}")) + .unwrap_or_default(); + format!( + "[{} {}{}{} {:?}] {}\n{}", + hit.id, + hit.kind.as_str(), + part, + tool, + hit.entry_range, + hit.label, + hit.summary + ) + }) + .collect::>() + .join("\n\n"); + Ok(ToolOutput { + summary: format!("Found {} evidence hit(s).", hits.len()), + content: (!content.is_empty()).then_some(content), + }) + } +} + +struct ReadEvidenceTool { + state: SessionExploreState, +} + +#[async_trait] +impl Tool for ReadEvidenceTool { + async fn execute( + &self, + input_json: &str, + _ctx: llm_engine::tool::ToolExecutionContext, + ) -> Result { + let params: ReadEvidenceParams = serde_json::from_str(input_json) + .map_err(|e| ToolError::InvalidArgument(format!("invalid read_evidence input: {e}")))?; + let selector = match (params.evidence_id.as_deref(), params.entry_range) { + (Some(id), None) => ReadSelector::Id(id), + (None, Some(range)) => ReadSelector::EntryRange(range), + _ => { + return Err(ToolError::InvalidArgument( + "read_evidence requires exactly one of evidence_id or entry_range".to_string(), + )); + } + }; + let detail = parse_read_detail(¶ms.mode)?; + let read = self.state.view.read( + selector, + ReadOptions { + include_tools: true, + tool_part: ToolPart::Both, + detail, + max_items: params.max_items.unwrap_or(10), + max_bytes: 16 * 1024, + }, + ); + let content = read + .entries + .iter() + .map(|entry| { + let part = entry + .tool_part + .map(|part| format!(" {part:?}")) + .unwrap_or_default(); + let tool = entry + .tool_name + .as_ref() + .map(|name| format!(" {name}")) + .unwrap_or_default(); + format!( + "[{} {}{}{} {:?}] {}\n{}", + entry.id, + entry.kind.as_str(), + part, + tool, + entry.entry_range, + entry.label, + entry.text + ) + }) + .collect::>() + .join("\n\n"); + let summary = if read.truncated { + format!( + "Read {} evidence entrie(s); output truncated.", + read.entries.len() + ) + } else { + format!("Read {} evidence entrie(s).", read.entries.len()) + }; + Ok(ToolOutput { + summary, + content: (!content.is_empty()).then_some(content), + }) + } +} + +struct StageCandidateTool { + state: SessionExploreState, +} + +#[async_trait] +impl Tool for StageCandidateTool { + async fn execute( + &self, + input_json: &str, + _ctx: llm_engine::tool::ToolExecutionContext, + ) -> Result { + let params: StageCandidateParams = serde_json::from_str(input_json).map_err(|e| { + ToolError::InvalidArgument(format!("invalid stage_candidate input: {e}")) + })?; + if params.evidence_ids.is_empty() { + return Err(ToolError::InvalidArgument( + "stage_candidate requires at least one evidence_id".to_string(), + )); + } + let mut evidence = Vec::with_capacity(params.evidence_ids.len()); + let mut source_refs = Vec::with_capacity(params.evidence_ids.len()); + for id in ¶ms.evidence_ids { + let staging = + self.state.view.staging_evidence_for(id).ok_or_else(|| { + ToolError::InvalidArgument(format!("unknown evidence_id {id:?}")) + })?; + let source_ref = + self.state.view.source_ref_for(id).ok_or_else(|| { + ToolError::InvalidArgument(format!("unknown evidence_id {id:?}")) + })?; + evidence.push(staging); + source_refs.push(source_ref); + } + let candidate = ExtractedCandidate { + kind: params.kind, + claim: params.claim, + why_useful: params.why_useful, + staleness: params.staleness, + evidence_ids: params.evidence_ids, + }; + let written = write_staging_candidate( + &self.state.layout, + self.state.source.clone(), + &self.state.extract_run_id, + candidate, + evidence, + source_refs, + ) + .map_err(|e| ToolError::ExecutionFailed(format!("write staging failed: {e}")))?; + let id = written.id.to_string(); + self.state + .staged + .lock() + .expect("session explore staged state poisoned") + .push(written); + Ok(ToolOutput { + summary: format!("Staged memory candidate {id}."), + content: Some(format!("staging_id: {id}")), + }) + } +} + +struct FinishExtractionTool { + state: SessionExploreState, +} + +#[async_trait] +impl Tool for FinishExtractionTool { + async fn execute( + &self, + input_json: &str, + _ctx: llm_engine::tool::ToolExecutionContext, + ) -> Result { + let params: FinishExtractionParams = serde_json::from_str(input_json).map_err(|e| { + ToolError::InvalidArgument(format!("invalid finish_extraction input: {e}")) + })?; + let actual = self + .state + .staged + .lock() + .expect("session explore staged state poisoned") + .len(); + if params.staged_count != actual { + return Err(ToolError::InvalidArgument(format!( + "finish_extraction staged_count {} does not match actual staged count {actual}", + params.staged_count + ))); + } + let reason = params.no_candidates_reason.clone(); + *self + .state + .finished + .lock() + .expect("session explore finished state poisoned") = Some(params); + Ok(ToolOutput { + summary: reason + .map(|reason| { + format!("Finished extraction with {actual} staged candidate(s): {reason}") + }) + .unwrap_or_else(|| { + format!("Finished extraction with {actual} staged candidate(s).") + }), + content: None, + }) + } +} + +fn parse_reference_kind(value: &str) -> Result { + ReferenceKind::parse(value).ok_or_else(|| { + ToolError::InvalidArgument(format!( + "invalid kind {value:?}; expected user, assistant/agent, system, or tool" + )) + }) +} + +fn parse_tool_part(value: &str) -> Result { + ToolPart::parse(value).ok_or_else(|| { + ToolError::InvalidArgument(format!( + "invalid tool_part {value:?}; expected input, output, or both" + )) + }) +} + +fn parse_read_detail(value: &str) -> Result { + match value { + "compact" => Ok(ReadDetail::Compact), + "full" => Ok(ReadDetail::Full), + other => Err(ToolError::InvalidArgument(format!( + "invalid read mode {other:?}; expected compact or full" + ))), + } +} + +pub(crate) fn render_extract_input(view: &SessionReferenceView) -> String { + let mut out = String::new(); + out.push_str("# Session overview\n\n"); + if view.overview().is_empty() { + out.push_str("No user/assistant overview entries are available.\n\n"); + } else { + for item in view.overview() { + out.push_str(&format!( + "- [{} {} {:?}] {}\n {}\n", + item.id, + item.kind.as_str(), + item.entry_range, + item.label, + truncate_line(&item.text, 500) + )); + } + out.push('\n'); + } + + out.push_str("# Initial evidence index\n\n"); + out.push_str( + "Use search_evidence/read_evidence to inspect details. Cite only M/T evidence ids in stage_candidate.evidence_ids.\n\n", + ); + let hits = view.search(&SearchOptions { + query: String::new(), + kind: None, + tool_part: None, + tool_name: None, + limit: Some(50), + min_entry_index: None, + }); + for hit in hits { + let part = hit + .tool_part + .map(|part| format!(" {part:?}")) + .unwrap_or_default(); + let tool = hit + .tool_name + .as_ref() + .map(|name| format!(" {name}")) + .unwrap_or_default(); + out.push_str(&format!( + "- [{} {}{}{} {:?}] {} — {}\n", + hit.id, + hit.kind.as_str(), + part, + tool, + hit.entry_range, + hit.label, + hit.summary + )); + } + out +} + +fn truncate_line(text: &str, max_chars: usize) -> String { + let normalized = text.replace('\n', " "); + if normalized.chars().count() <= max_chars { + normalized + } else { + let mut out = normalized.chars().take(max_chars).collect::(); + out.push_str("…"); + out + } +} + +#[cfg(test)] +mod tests { + use super::*; + use llm_engine::Item; + use tempfile::TempDir; + + #[test] + fn descriptor_declares_session_explore_tools() { + let temp = TempDir::new().unwrap(); + let state = SessionExploreState::new( + SessionReferenceView::new("segment-1", vec![Item::user_message("remember this")]), + WorkspaceLayout::new(temp.path()), + SourceRef { + segment_id: "segment-1".to_string(), + range: [0, 0], + }, + ); + let descriptor = SessionExploreFeature::new(state).descriptor(); + assert_eq!(descriptor.id.as_str(), "builtin:session-explore"); + let names = descriptor + .tools + .iter() + .map(|tool| tool.name.as_str()) + .collect::>(); + assert_eq!( + names, + vec![ + "search_evidence", + "read_evidence", + "stage_candidate", + "finish_extraction" + ] + ); + } + + #[test] + fn render_extract_input_exposes_overview_and_evidence_ids() { + let view = SessionReferenceView::new( + "segment-1", + vec![ + Item::user_message("user preference"), + Item::assistant_message("assistant reply"), + Item::tool_call("c1", "Read", "{}"), + Item::tool_result_with_content("c1", "read ok", "file body"), + ], + ); + let input = render_extract_input(&view); + assert!(input.contains("# Session overview")); + assert!(input.contains("M0000")); + assert!(input.contains("T0002i")); + assert!(input.contains("T0003o")); + assert!(input.contains("stage_candidate.evidence_ids")); + } + + #[tokio::test] + async fn stage_candidate_writes_staging_record_with_source_evidence() { + let temp = TempDir::new().unwrap(); + let state = SessionExploreState::new( + SessionReferenceView::new("segment-1", vec![Item::user_message("durable decision")]), + WorkspaceLayout::new(temp.path()), + SourceRef { + segment_id: "segment-1".to_string(), + range: [0, 0], + }, + ); + let tool = StageCandidateTool { + state: state.clone(), + }; + tool.execute( + &serde_json::json!({ + "kind": "decision", + "claim": "Use host-created evidence anchors for extract staging.", + "why_useful": "Future consolidation can trust bounded source refs.", + "evidence_ids": ["M0000"] + }) + .to_string(), + llm_engine::tool::ToolExecutionContext::direct(), + ) + .await + .unwrap(); + + let staged = state.staged(); + assert_eq!(staged.len(), 1); + let bytes = std::fs::read(&staged[0].path).unwrap(); + let record: memory::extract::StagingRecord = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(record.kind, CandidateKind::Decision); + assert_eq!(record.evidence.len(), 1); + assert_eq!(record.evidence[0].id, "M0000"); + assert_eq!(record.source_refs.len(), 1); + assert_eq!(record.source_refs[0].evidence_id.as_deref(), Some("M0000")); + } +} diff --git a/crates/worker/src/session_reference.rs b/crates/worker/src/session_reference.rs index 845f2312..c99a9bf4 100644 --- a/crates/worker/src/session_reference.rs +++ b/crates/worker/src/session_reference.rs @@ -7,6 +7,7 @@ use std::sync::Arc; use llm_engine::{Item, Role}; +use memory::extract::StagingEvidence; use memory::schema::{EvidenceKind, SourceEvidenceRef}; const DEFAULT_SEARCH_LIMIT: usize = 20; @@ -36,7 +37,7 @@ impl ReferenceKind { pub(crate) fn parse(value: &str) -> Option { match value { "user" => Some(Self::User), - "assistant" => Some(Self::Assistant), + "assistant" | "agent" => Some(Self::Assistant), "system" => Some(Self::System), "tool" => Some(Self::Tool), _ => None, @@ -44,10 +45,7 @@ impl ReferenceKind { } fn evidence_kind(self) -> EvidenceKind { - match self { - Self::User | Self::Assistant | Self::System => EvidenceKind::new(EvidenceKind::MESSAGE), - Self::Tool => EvidenceKind::new(EvidenceKind::TOOL_RESULT), - } + EvidenceKind::new(EvidenceKind::MESSAGE) } } @@ -94,6 +92,20 @@ pub(crate) struct ReferenceEntry { search_text: String, } +impl ReferenceEntry { + fn evidence_kind(&self) -> EvidenceKind { + match (self.kind, self.tool_part) { + (ReferenceKind::Tool, Some(ToolPart::Input)) => { + EvidenceKind::new(EvidenceKind::TOOL_CALL) + } + (ReferenceKind::Tool, Some(ToolPart::Output | ToolPart::Both) | None) => { + EvidenceKind::new(EvidenceKind::TOOL_RESULT) + } + _ => self.kind.evidence_kind(), + } + } +} + #[derive(Debug, Clone, Default)] pub(crate) struct SearchOptions { pub query: String, @@ -378,12 +390,38 @@ impl SessionReferenceView { segment_id: Some(self.segment_id.clone()), entry_range: Some(entry.entry_range), evidence_id: Some(entry.id.clone()), - evidence_kind: Some(entry.kind.evidence_kind()), + evidence_kind: Some(entry.evidence_kind()), label: Some(entry.label.clone()), summary: Some(entry.summary.clone()), ..Default::default() }) } + + pub(crate) fn staging_evidence_for(&self, id: &str) -> Option { + let entry = self.index.iter().find(|entry| entry.id == id)?; + let read = self.read( + ReadSelector::Id(id), + ReadOptions { + include_tools: true, + tool_part: ToolPart::Both, + detail: ReadDetail::Compact, + max_items: 1, + max_bytes: 2 * 1024, + }, + ); + let excerpt = read + .entries + .first() + .map(|entry| entry.text.clone()) + .unwrap_or_else(|| entry.summary.clone()); + Some(StagingEvidence { + id: entry.id.clone(), + kind: entry.evidence_kind(), + entry_range: Some(entry.entry_range), + excerpt: Some(excerpt), + summary: Some(entry.summary.clone()), + }) + } } fn render_item( diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index dadeb72c..63564de1 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -30,7 +30,9 @@ use manifest::{ use crate::compact::state::CompactState; use crate::compact::usage_tracker::UsageTracker; -use crate::feature::builtin::TaskFeature; +use crate::feature::builtin::{ + SessionExploreFeature, SessionExploreState, TaskFeature, render_extract_input, +}; use crate::feature::{FeatureRegistryBuilder, FeatureRegistryInstallReport}; use crate::hook::{ Hook, HookRegistryBuilder, OnAbort, OnPromptSubmit, OnTurnEnd, PostToolCall, PreLlmRequest, @@ -3130,8 +3132,47 @@ impl Worker { return Err(WorkerError::PromptCatalog(err)); } }; - let ctx = Arc::new(extract::ExtractWorkerContext::new()); - let input_text = extract::build_extract_input(&items_to_extract); + let source_segment_id = self.segment_state.segment_id(); + let source = memory::schema::SourceRef { + segment_id: source_segment_id.to_string(), + range: [start_entry as u64, end_entry as u64], + }; + let session_view = crate::session_reference::SessionReferenceView::new( + source_segment_id.to_string(), + items_to_extract, + ); + let session_explore_state = SessionExploreState::new(session_view, layout.clone(), source); + let input_text = render_extract_input(session_explore_state.view()); + let mut internal_tools = Vec::new(); + let mut internal_hook_builder = HookRegistryBuilder::new(); + let feature_report = FeatureRegistryBuilder::new() + .with_module(SessionExploreFeature::new(session_explore_state.clone())) + .install_into_pending(&mut internal_tools, &mut internal_hook_builder); + let installed_tool_names = feature_report.installed_tool_names(); + let expected_extract_tools = [ + "search_evidence", + "read_evidence", + "stage_candidate", + "finish_extraction", + ]; + if !expected_extract_tools.iter().all(|name| { + installed_tool_names + .iter() + .any(|installed| installed == name) + }) { + audit.emit( + &layout, + event_tx, + memory::audit::WorkerLifecycleStatus::Failed, + "session_explore_feature_install_failed", + None, + Some(extract_audit_base), + None, + ); + return Err(WorkerError::FeatureInstall( + "session-explore feature install failed".to_string(), + )); + } let internal_result = run_internal_worker(InternalWorkerSpec { slug: "memory-extract", system_prompt: extract_system_prompt, @@ -3139,7 +3180,7 @@ impl Worker { client, cache_key: Some(self.segment_id().to_string()), max_turns: extract_worker_max_turns, - tools: vec![extract::write_extracted_tool(ctx.clone())], + tools: internal_tools, }) .await; let usage = match internal_result { @@ -3159,37 +3200,13 @@ impl Worker { } }; - let payload = ctx.take_payload().unwrap_or_else(|| { + let staging_results = session_explore_state.staged(); + if !session_explore_state.is_finished() { tracing::warn!( - "extract worker did not call write_extracted; advancing pointer with empty payload" + staged_count = staging_results.len(), + "extract worker did not call finish_extraction; advancing pointer with staged output" ); - extract::ExtractedPayload::default() - }); - - let source_segment_id = self.segment_state.segment_id(); - let staging_results = if payload.is_empty() { - Vec::new() - } else { - let source = memory::schema::SourceRef { - segment_id: source_segment_id.to_string(), - range: [start_entry as u64, end_entry as u64], - }; - match extract::write_staging(&layout, source, payload) { - Ok(results) => results, - Err(err) => { - audit.emit( - &layout, - event_tx, - memory::audit::WorkerLifecycleStatus::Failed, - format!("staging_write_failed: {err}"), - usage.clone(), - Some(extract_audit_base), - None, - ); - return Err(WorkerError::ExtractStaging(err)); - } - } - }; + } let staging_id = staging_results .first() .map(|result| result.id.to_string()) @@ -4807,6 +4824,9 @@ pub enum WorkerError { #[error("memory extract staging write failed: {0}")] ExtractStaging(#[source] std::io::Error), + #[error("feature install failed: {0}")] + FeatureInstall(String), + #[error("memory consolidation lock acquisition failed: {0}")] ConsolidationLock(#[source] memory::consolidate::LockError), diff --git a/crates/worker/tests/compact_events_test.rs b/crates/worker/tests/compact_events_test.rs index 933b7e85..01094b53 100644 --- a/crates/worker/tests/compact_events_test.rs +++ b/crates/worker/tests/compact_events_test.rs @@ -539,16 +539,14 @@ target = "./" permission = "write" "#; -fn write_extracted_tool_use_events(call_id: &str) -> Vec { +fn finish_extraction_tool_use_events(call_id: &str) -> Vec { let input = serde_json::json!({ - "decisions": [], - "discussions": [], - "attempts": [], - "requests": [] + "staged_count": 0, + "no_candidates_reason": "test run has no durable candidates" }) .to_string(); vec![ - LlmEvent::tool_use_start(0, call_id, "write_extracted"), + LlmEvent::tool_use_start(0, call_id, "finish_extraction"), LlmEvent::tool_input_delta(0, input), LlmEvent::tool_use_stop(0), LlmEvent::Status(StatusEvent { @@ -561,13 +559,13 @@ fn write_extracted_tool_use_events(call_id: &str) -> Vec { async fn compact_resets_extract_pointer_so_extract_can_fire_again() { // Mock LLM responses, in call order: // [0] first run with usage(1000) so extract threshold (=1) fires. - // [1] extract worker invokes write_extracted with empty payload. + // [1] extract worker invokes finish_extraction with empty output. // [2] extract worker closes after the tool result. // [3] compact worker invokes write_summary. // [4] compact worker closes after the tool result. let client = MockClient::new(vec![ text_events_with_usage("hi", 1000), - write_extracted_tool_use_events("ec1"), + finish_extraction_tool_use_events("ec1"), single_text_events("done"), write_summary_tool_use_events("sc1", "summary"), single_text_events("done"), @@ -703,7 +701,7 @@ permission = "write" async fn extract_large_unprocessed_range_does_not_abort_on_input_occupancy() { let client = MockClient::new(vec![ text_events_with_usage("recorded", 1000), - write_extracted_tool_use_events("ec-large"), + finish_extraction_tool_use_events("ec-large"), single_text_events("done"), ]); let mut worker = make_worker_with_manifest(EXTRACT_NO_COMPACT_MANIFEST, client).await; @@ -724,7 +722,7 @@ async fn extract_large_unprocessed_range_does_not_abort_on_input_occupancy() { async fn spawn_and_wait_drives_extract_to_completion() { let client = MockClient::new(vec![ text_events_with_usage("hi", 1000), - write_extracted_tool_use_events("ec1"), + finish_extraction_tool_use_events("ec1"), single_text_events("done"), ]); let mut worker = make_worker_with_manifest(EXTRACT_NO_COMPACT_MANIFEST, client).await; @@ -752,7 +750,7 @@ async fn detached_extract_does_not_fork_session_log() { // `ensure_head_or_fork` does not spawn a new session. let client = MockClient::new(vec![ text_events_with_usage("hi", 1000), - write_extracted_tool_use_events("ec1"), + finish_extraction_tool_use_events("ec1"), single_text_events("done"), text_events_with_usage("ok", 1000), ]); diff --git a/resources/prompts/internal/memory_extract_system.md b/resources/prompts/internal/memory_extract_system.md index 90272419..4ae9e981 100644 --- a/resources/prompts/internal/memory_extract_system.md +++ b/resources/prompts/internal/memory_extract_system.md @@ -1,6 +1,6 @@ You are a Yoi memory extract worker. -Your job is to read the supplied conversation slice and extract only memory candidates that may be worth later consolidation. Do not produce activity logs. +Your job is to inspect the supplied host-created session reference view and stage only memory candidates that may be worth later consolidation. Do not produce activity logs. ## Language @@ -9,22 +9,39 @@ Your job is to read the supplied conversation slice and extract only memory cand - 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. -Call `write_extracted` exactly once with an object of this shape: +## Tools + +Use the session-explore tools only: + +- `search_evidence`: find bounded evidence ids in the host-created session index. Optional `kind` accepts `user`, `assistant`/`agent`, `system`, or `tool`. +- `read_evidence`: inspect a bounded evidence id or entry range before staging when the overview/index is not enough. +- `stage_candidate`: write one flat staging record for one memory candidate. +- `finish_extraction`: finish the run after all useful candidates are staged, or after deciding there are no useful candidates. + +Do not invent evidence ids. Stage candidates only with `M...` or `T...` ids returned by `search_evidence` / `read_evidence` or shown in the initial evidence index. Overview `O...` ids are orientation labels, not source evidence ids. + +Call `stage_candidate` once per useful candidate with this shape: ```json { - "candidates": [ - { - "kind": "preference", - "claim": "...", - "why_useful": "...", - "staleness": "...", - "evidence_ids": [] - } - ] + "kind": "preference", + "claim": "...", + "why_useful": "...", + "staleness": "...", + "evidence_ids": ["M0001"] } ``` +Then call `finish_extraction` exactly once: + +```json +{ + "staged_count": 1 +} +``` + +If nothing is worth staging, do not call `stage_candidate`; call `finish_extraction` with `{"staged_count": 0, "no_candidates_reason": "..."}`. + Allowed candidate kinds: - `preference`: durable user/workspace preference or working style, not a one-off instruction. @@ -39,11 +56,11 @@ Required fields per candidate: - `kind`: one of the allowed candidate kinds. - `claim`: concise statement of the candidate. - `why_useful`: why this candidate may be useful for future consolidation. +- `evidence_ids`: one or more host-issued source evidence ids. Optional fields: - `staleness`: when this candidate should be revisited or invalidated. -- `evidence_ids`: leave empty in this transitional path unless host-issued evidence ids are present. Do not extract: @@ -58,6 +75,4 @@ Do not extract: - validation results unless they imply a reusable lesson, active blocker, or authority evidence; - implementation details that belong only in commit diff. -Prefer no candidates over noisy candidates. If nothing is worth staging, call `write_extracted` with `{"candidates": []}`. - -Do not include record ids, source anchors, session metadata, free-form prose, or raw tool output content. The host attaches staging metadata mechanically. +Prefer no candidates over noisy candidates. The host attaches staging metadata and bounded evidence mechanically. \ No newline at end of file