worker: add internal worker runner

This commit is contained in:
Keisuke Hirata 2026-07-18 05:59:02 +09:00
parent ff04becfab
commit 88b91a2c55
No known key found for this signature in database
6 changed files with 189 additions and 43 deletions

View File

@ -1,8 +1,8 @@
--- ---
title: 'Internal Worker runnerを実装する' title: 'Internal Worker runnerを実装する'
state: 'inprogress' state: 'closed'
created_at: '2026-07-17T20:47:11Z' created_at: '2026-07-17T20:47:11Z'
updated_at: '2026-07-17T20:47:48Z' updated_at: '2026-07-17T20:58:47Z'
assignee: null assignee: null
queued_by: 'yoi ticket' queued_by: 'yoi ticket'
queued_at: '2026-07-17T20:47:48Z' queued_at: '2026-07-17T20:47:48Z'

View File

@ -0,0 +1,9 @@
Implemented reusable Worker-internal runner and migrated memory extract to use it.
The new runner provides a Worker-owned path for isolated internal LLM jobs with caller-supplied prompt, input, limited tools, cache key, max turns, and usage capture. Memory extract now uses this runner while preserving the existing trigger and staging behavior.
Validation passed:
- `cargo fmt --check`
- `cargo test -p memory`
- `cargo test -p worker`
- `nix build .#yoi`

View File

@ -22,4 +22,61 @@ Ticket を `yoi ticket` が queued にしました。
State changed to `inprogress`. State changed to `inprogress`.
---
<!-- event: implementation_report author: hare at: 2026-07-17T20:58:47Z -->
## Implementation report
Implemented reusable Worker-internal runner and migrated memory extract to use it.
Changes:
- Added `crates/worker/src/internal_worker.rs` with `InternalWorkerSpec`, `InternalWorkerRunResult`, `InternalWorkerRunError`, and `run_internal_worker`.
- The runner owns isolated Engine creation, limited tool registration, cache key / max turns, usage capture, and foreground-history isolation by construction.
- Updated memory extract path to run through the internal runner with only the extract tool surface supplied by the caller.
- Kept existing extract behavior: post-run threshold trigger, transitional `write_extracted` path, and flat staging output.
- Left persistence / Protocol exposure / session-explore feature for later Tickets.
Validation:
- `cargo fmt --check`
- `cargo test -p memory`
- `cargo test -p worker`
- `nix build .#yoi`
---
<!-- event: state_changed author: "yoi ticket" at: 2026-07-17T20:58:47Z from: inprogress to: done reason: cli_state field: state -->
## State changed
State changed to `done`.
---
<!-- event: state_changed author: hare at: 2026-07-17T20:58:47Z from: done to: closed reason: closed field: state -->
## State changed
Ticket を closed にしました。
---
<!-- event: close author: hare at: 2026-07-17T20:58:47Z status: closed -->
## 完了
Implemented reusable Worker-internal runner and migrated memory extract to use it.
The new runner provides a Worker-owned path for isolated internal LLM jobs with caller-supplied prompt, input, limited tools, cache key, max turns, and usage capture. Memory extract now uses this runner while preserving the existing trigger and staging behavior.
Validation passed:
- `cargo fmt --check`
- `cargo test -p memory`
- `cargo test -p worker`
- `nix build .#yoi`
--- ---

View File

@ -0,0 +1,92 @@
//! Reusable runner for Worker-internal LLM jobs.
//!
//! Internal workers are isolated model/tool runs owned by the foreground
//! [`Worker`](crate::Worker). They do not append their harness prompt/input to
//! the foreground session history. Callers provide a deliberately limited tool
//! surface and decide how to apply the result.
use std::sync::{Arc, Mutex};
use llm_engine::llm_client::client::LlmClient;
use llm_engine::llm_client::event::UsageEvent;
use llm_engine::tool::ToolDefinition;
use llm_engine::{Engine, EngineError};
/// Specification for a single internal worker run.
pub(crate) struct InternalWorkerSpec {
/// Stable purpose label for audit/debug/persistence metadata.
pub purpose: &'static str,
/// System prompt for the isolated engine.
pub system_prompt: String,
/// Initial user input for the isolated engine.
pub input: String,
/// Client used by the internal engine.
pub client: Box<dyn LlmClient>,
/// Optional prompt-cache key.
pub cache_key: Option<String>,
/// Optional turn limit for the internal engine.
pub max_turns: Option<u32>,
/// Deliberately limited tool surface for this internal run.
pub tools: Vec<ToolDefinition>,
}
/// Result metadata for an internal worker run.
#[derive(Debug, Clone, Default)]
pub(crate) struct InternalWorkerRunResult {
pub purpose: &'static str,
/// Last usage event observed for this internal run.
pub usage: Option<UsageEvent>,
}
/// Error metadata for an internal worker run.
#[derive(Debug)]
pub(crate) struct InternalWorkerRunError {
pub purpose: &'static str,
pub source: EngineError,
/// Last usage event observed before the failure, if any.
pub usage: Option<UsageEvent>,
}
/// Run an isolated internal worker engine.
pub(crate) async fn run_internal_worker(
spec: InternalWorkerSpec,
) -> Result<InternalWorkerRunResult, InternalWorkerRunError> {
let InternalWorkerSpec {
purpose,
system_prompt,
input,
client,
cache_key,
max_turns,
tools,
} = spec;
let mut worker = Engine::new(client).system_prompt(system_prompt);
worker.set_cache_key(cache_key);
worker.set_max_turns(max_turns);
worker.register_tools(tools);
let usage_capture = Arc::new(Mutex::new(None));
let usage_capture_for_worker = usage_capture.clone();
worker.on_usage(move |event| {
*usage_capture_for_worker
.lock()
.expect("internal worker usage capture poisoned") = Some(event.clone());
});
let run_result = worker.run(input).await;
let usage = usage_capture
.lock()
.expect("internal worker usage capture poisoned")
.clone();
match run_result {
Ok(_output) => Ok(InternalWorkerRunResult { purpose, usage }),
Err(source) => Err(InternalWorkerRunError {
purpose,
source,
usage,
}),
}
}

View File

@ -16,6 +16,7 @@ mod shutdown_after_idle;
pub mod skill; pub mod skill;
pub mod spawn; pub mod spawn;
mod internal_worker;
mod interrupt_prep; mod interrupt_prep;
mod permission; mod permission;
mod ticket_event_notify; mod ticket_event_notify;

View File

@ -37,6 +37,7 @@ use crate::hook::{
PreToolCall, PreToolCall,
}; };
use crate::in_flight::InFlightEvents; use crate::in_flight::InFlightEvents;
use crate::internal_worker::{InternalWorkerSpec, run_internal_worker};
const COMPACTION_EXTENSION_DOMAIN: &str = "yoi.compaction"; const COMPACTION_EXTENSION_DOMAIN: &str = "yoi.compaction";
const COMPACTION_BLOCK_ID: &str = "compact"; const COMPACTION_BLOCK_ID: &str = "compact";
@ -3129,40 +3130,34 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
return Err(WorkerError::PromptCatalog(err)); return Err(WorkerError::PromptCatalog(err));
} }
}; };
let mut extract_worker = Engine::new(client).system_prompt(extract_system_prompt);
extract_worker.set_cache_key(Some(self.segment_id().to_string()));
extract_worker.set_max_turns(extract_worker_max_turns);
let usage_capture = Arc::new(Mutex::new(None));
let usage_capture_for_worker = usage_capture.clone();
extract_worker.on_usage(move |event| {
*usage_capture_for_worker
.lock()
.expect("memory extract usage capture poisoned") =
Some(usage_audit_from_event(event));
});
let ctx = Arc::new(extract::ExtractWorkerContext::new()); let ctx = Arc::new(extract::ExtractWorkerContext::new());
extract_worker.register_tool(extract::write_extracted_tool(ctx.clone()));
let input_text = extract::build_extract_input(&items_to_extract); let input_text = extract::build_extract_input(&items_to_extract);
if let Err(err) = extract_worker.run(input_text).await { let internal_result = run_internal_worker(InternalWorkerSpec {
let usage = usage_capture purpose: "memory_extract",
.lock() system_prompt: extract_system_prompt,
.expect("memory extract usage capture poisoned") input: input_text,
.clone(); client,
cache_key: Some(self.segment_id().to_string()),
max_turns: extract_worker_max_turns,
tools: vec![extract::write_extracted_tool(ctx.clone())],
})
.await;
let usage = match internal_result {
Ok(result) => result.usage.as_ref().map(usage_audit_from_event),
Err(err) => {
let usage = err.usage.as_ref().map(usage_audit_from_event);
audit.emit( audit.emit(
&layout, &layout,
event_tx, event_tx,
lifecycle_status_for_worker_error(&err), lifecycle_status_for_worker_error(&err.source),
format!("worker_failed: {err}"), format!("worker_failed: {}", err.source),
usage, usage,
Some(extract_audit_base), Some(extract_audit_base),
None, None,
); );
return Err(WorkerError::Engine(err)); return Err(WorkerError::Engine(err.source));
} }
};
let payload = ctx.take_payload().unwrap_or_else(|| { let payload = ctx.take_payload().unwrap_or_else(|| {
tracing::warn!( tracing::warn!(
@ -3182,16 +3177,12 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
match extract::write_staging(&layout, source, payload) { match extract::write_staging(&layout, source, payload) {
Ok(results) => results, Ok(results) => results,
Err(err) => { Err(err) => {
let usage = usage_capture
.lock()
.expect("memory extract usage capture poisoned")
.clone();
audit.emit( audit.emit(
&layout, &layout,
event_tx, event_tx,
memory::audit::WorkerLifecycleStatus::Failed, memory::audit::WorkerLifecycleStatus::Failed,
format!("staging_write_failed: {err}"), format!("staging_write_failed: {err}"),
usage, usage.clone(),
Some(extract_audit_base), Some(extract_audit_base),
None, None,
); );
@ -3230,10 +3221,6 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
.staging_paths .staging_paths
.push(result.path.display().to_string()); .push(result.path.display().to_string());
} }
let usage = usage_capture
.lock()
.expect("memory extract usage capture poisoned")
.clone();
let reason = if staging_id.is_empty() { let reason = if staging_id.is_empty() {
"completed_no_staging_output" "completed_no_staging_output"
} else { } else {