diff --git a/Cargo.lock b/Cargo.lock index b758415..51e7923 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1604,9 +1604,11 @@ checksum = "ec9d92d097f4749b64e8cc33d924d9f40a2d4eb91402b458014b781f5733d60f" name = "honcho-client" version = "0.1.0" dependencies = [ + "anyhow", "reqwest 0.13.4", "serde", "serde_json", + "tokio", "tracing", ] diff --git a/crates/agent-loop/src/activation.rs b/crates/agent-loop/src/activation.rs index 2bfbfc1..a8fd894 100644 --- a/crates/agent-loop/src/activation.rs +++ b/crates/agent-loop/src/activation.rs @@ -66,10 +66,19 @@ pub trait ToolActivator: Send + Sync { /// other — the defs re-read at the next round makes the new grants visible. pub struct ActivateToolsTool { activator: Arc, + definition_override: Option, } impl ActivateToolsTool { - pub fn new(activator: Arc) -> Self { Self { activator } } + pub fn new(activator: Arc) -> Self { + Self { activator, definition_override: None } + } + + /// Override the advertised definition (legacy parity). + pub fn with_definition(mut self, def: Value) -> Self { + self.definition_override = Some(def); + self + } } #[async_trait] @@ -77,6 +86,9 @@ impl Tool for ActivateToolsTool { fn name(&self) -> &str { "activate_tools" } fn definition(&self) -> Value { + if let Some(def) = &self.definition_override { + return def.clone(); + } json!({ "type": "function", "function": { diff --git a/crates/agent-loop/src/context.rs b/crates/agent-loop/src/context.rs index 475ee66..4f05942 100644 --- a/crates/agent-loop/src/context.rs +++ b/crates/agent-loop/src/context.rs @@ -275,6 +275,9 @@ fn project_message(out: &mut Vec, msg: &StoredMessage, result_limit: Opti before a result was recorded]" .to_string() } + crate::store::CallState::Failed => { + format!("Error: {}", call.result.as_deref().unwrap_or("unknown error")) + } _ => call.result.clone().unwrap_or_default(), }; if let Some(limit) = result_limit diff --git a/crates/agent-loop/src/delegate.rs b/crates/agent-loop/src/delegate.rs new file mode 100644 index 0000000..6bc7976 --- /dev/null +++ b/crates/agent-loop/src/delegate.rs @@ -0,0 +1,417 @@ +//! Sub-agents as a tool (blueprint §7, D2): the kernel never intercepts +//! anything — `delegate` is a tool like any other, dispatched through the +//! normal gate/hooks/execution path. A sync child is just a slow tool call the +//! parent awaits; a homogeneous batch of sync delegates fans out through the +//! kernel's generic concurrency (`concurrency_safe`). +//! +//! The crate ships the SYNC flow. Async delegation rides the host's +//! `AsyncExecutor` (phase-3 concern: Skald wires its durable cron executor +//! there); calling it here fails with a clear error. + +use std::sync::Arc; + +use serde_json::{Value, json}; + +use crate::async_trait; +use crate::context::SystemContextSource; +use crate::events::{EventSink, LoopEvent}; +use crate::ids::FrameId; +use crate::manager::{LoopManager, LoopParams, TurnMeta}; +use crate::model::{ModelHint, ModelSelector}; +use crate::store::{FrameSpec, HistoryStore, NewMessage}; +use crate::tool::{SharedToolSet, Tool, ToolCtx, ToolFailure, ToolOutput, ToolSet}; + +// ── AgentCatalog ───────────────────────────────────────────────────────────── + +/// The agent's kind (from the host's meta). Only `Task` agents are +/// dispatchable via `delegate`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AgentKind { + Chat, + Task, + System, +} + +/// A dispatchable agent. +#[derive(Clone)] +pub struct AgentProfile { + pub id: String, + pub kind: AgentKind, + /// The child's system context (its own prompt — never the parent's, B3). + pub context: Arc, + /// How the child's tool set derives from the parent's (ignored when + /// `toolset` is set). + pub tools: ToolSelection, + /// Full tool-set override (hosts whose children need a fresh registry + /// rather than a filtered view of the parent's — e.g. fresh grant sets). + pub toolset: Option>, + /// Model pin (bypasses AUTO). Strength is resolved by the host's selector. + pub model: Option, + /// Per-child selector override (e.g. a different required strength, D14). + pub selector: Option>, + /// Per-child assembler override (e.g. scoped DTL activation). + pub assembler: Option>, +} + +/// How a child's tool set derives from the parent's: strip `remove` by name, +/// then append `add`. +#[derive(Clone, Default)] +pub struct ToolSelection { + pub remove: Vec, + pub add: Vec>, +} + +impl ToolSelection { + pub fn inherit() -> Self { Self::default() } + pub fn minus(names: impl IntoIterator>) -> Self { + Self { remove: names.into_iter().map(Into::into).collect(), add: Vec::new() } + } + pub fn plus(tools: Vec>) -> Self { + Self { remove: Vec::new(), add: tools } + } +} + +/// Summary for catalog listings (a future `list_agents` tool). +#[derive(Debug, Clone)] +pub struct AgentSummary { + pub id: String, + pub kind: AgentKind, + pub description: String, +} + +#[async_trait] +pub trait AgentCatalog: Send + Sync { + /// Load a dispatchable profile, built for `child_frame` (already opened by + /// the DelegateTool — frame-scoped pieces like grants/activation anchor to + /// it). MUST reject non-`Task` kinds and unknown ids. + async fn get(&self, id: &str, child_frame: FrameId) -> crate::Result; + async fn list(&self, kind: AgentKind) -> Vec; + /// Frame-exit hook (host cleanup, e.g. deleting stack-scoped activations). + async fn on_child_closed(&self, _frame: crate::ids::FrameId) {} +} + +// ── FilteredToolSet ────────────────────────────────────────────────────────── + +/// The child's tool set: parent's minus `remove`, plus `add`. +pub struct FilteredToolSet { + inner: Arc, + remove: Vec, + add: Vec>, +} + +impl ToolSet for FilteredToolSet { + fn defs(&self, model: &crate::model::ModelInfo) -> Vec { + let mut defs: Vec = self + .inner + .defs(model) + .into_iter() + .filter(|d| { + let name = d["function"]["name"].as_str().unwrap_or(""); + !self.remove.iter().any(|r| r == name) + }) + .collect(); + defs.extend(self.add.iter().map(|t| t.definition())); + defs + } + + fn find(&self, name: &str) -> Option> { + if let Some(t) = self.add.iter().find(|t| t.name() == name) { + return Some(t.clone()); + } + if self.remove.iter().any(|r| r == name) { + return None; + } + self.inner.find(name) + } +} + +// ── DelegateTool ───────────────────────────────────────────────────────────── + +/// The shipped `delegate` tool. The parent loop simply awaits a slow tool — +/// nesting is reconstructed by subscribers from the `parent_frame` event tags. +#[derive(Clone)] +pub struct DelegateTool { + manager: Arc, + catalog: Arc, + store: Arc, + max_depth: u32, + name: String, + definition_override: Option, +} + +impl DelegateTool { + pub fn new( + manager: Arc, + catalog: Arc, + store: Arc, + max_depth: u32, + ) -> Self { + Self { manager, catalog, store, max_depth, name: "delegate".to_string(), definition_override: None } + } + + /// Register under a different wire name (Skald's legacy aliases + /// `execute_task` / `execute_subtask`, blueprint D11). + pub fn with_name(mut self, name: impl Into) -> Self { + self.name = name.into(); + self + } + + /// Override the advertised definition (legacy aliases keep their exact + /// legacy schema byte-for-byte). + pub fn with_definition(mut self, def: Value) -> Self { + self.definition_override = Some(def); + self + } + + /// The schema: `agent_id` + `prompt` required; `title`, `description`, + /// `mode` ("sync" — async rides the host executor), `client` accepted for + /// legacy compatibility. + fn schema(&self) -> Value { + json!({ + "type": "object", + "properties": { + "agent_id": { "type": "string", "description": "Id of the task agent to delegate to" }, + "prompt": { "type": "string", "description": "The full brief for the sub-agent" }, + "title": { "type": "string", "description": "Optional short title for the task" }, + "description": { "type": "string", "description": "Optional longer description" }, + "mode": { "type": "string", "enum": ["sync", "async"], + "description": "sync: wait for the result. async: host-scheduled (if wired)" }, + "client": { "type": "string", "description": "Optional model override" } + }, + "required": ["agent_id", "prompt"] + }) + } + + async fn run_sync(&self, agent_id: &str, prompt: &str, ctx: &ToolCtx) -> Result { + if agent_id == ctx.agent { + return Err(ToolFailure::Failed(format!( + "delegate: an agent cannot call itself (`{agent_id}`)" + ))); + } + + // Depth check (max recursion, from the parent frame). + let parent_frame = self + .store + .get_frame(ctx.frame) + .await + .map_err(|e| ToolFailure::Failed(format!("delegate: frame lookup failed: {e}")))? + .ok_or_else(|| ToolFailure::Failed("delegate: parent frame not found".into()))?; + let new_depth = parent_frame.spec.depth + 1; + if new_depth > self.max_depth { + return Err(ToolFailure::Failed(format!( + "delegate: maximum agent depth ({}) exceeded — refusing to recurse further", + self.max_depth + ))); + } + + let child_frame = self + .store + .open_frame(&ctx.conversation, Some(ctx.frame), FrameSpec { + agent: agent_id.to_string(), + prompt: Some(prompt.to_string()), + depth: new_depth, + parent_call: Some(ctx.call_id), + meta: Value::Null, + }) + .await + .map_err(|e| ToolFailure::Failed(format!("delegate: open frame failed: {e}")))?; + + // Profile AFTER the frame exists (frame-scoped pieces anchor to it). + // On rejection the frame is closed so nothing dangles. + let profile = match self.catalog.get(agent_id, child_frame).await { + Ok(p) => p, + Err(e) => { + let _ = self.store.close_frame(child_frame).await; + return Err(ToolFailure::Failed(format!("delegate: {e}"))); + } + }; + if profile.kind != AgentKind::Task { + let _ = self.store.close_frame(child_frame).await; + return Err(ToolFailure::Failed(format!( + "delegate: agent `{agent_id}` is not dispatchable (only task agents are)" + ))); + } + + self.store + .append(child_frame, NewMessage::agent(prompt)) + .await + .map_err(|e| ToolFailure::Failed(format!("delegate: append failed: {e}")))?; + + let events = EventSink::from_extensions(&ctx.extensions); + if let Some(ev) = &events { + ev.emit(child_frame, Some(ctx.frame), LoopEvent::AgentSpawned { + frame: child_frame, + agent: agent_id.to_string(), + depth: new_depth, + prompt_preview: preview_truncate(prompt, 500), + parent_call: ctx.call_id, + parent_agent: ctx.agent.clone(), + }); + } + + // The child's tool set: the profile's full override, or the parent's + // filtered per its ToolSelection. + let child_tools: Arc = match profile.toolset.clone() { + Some(ts) => ts, + None => { + let parent_tools = ctx + .extensions + .get::() + .ok_or_else(|| ToolFailure::Failed("delegate: no ToolSet in extensions".into()))?; + Arc::new(FilteredToolSet { + inner: parent_tools.0.clone(), + remove: profile.tools.remove.clone(), + add: profile.tools.add.clone(), + }) + } + }; + + let child = self + .manager + .start_loop(LoopParams { + conversation: ctx.conversation.clone(), + frame: child_frame, + parent_frame: Some(ctx.frame), + agent: agent_id.to_string(), + system: profile.context, + tools: child_tools, + model_hint: profile.model.unwrap_or_default(), + selector: profile.selector, + // Sticky /stop: the child rides the parent's cancellation tree. + token: Some(ctx.cancel.child_token()), + live_input: None, + extensions: ctx.extensions.clone(), + meta: TurnMeta::default(), + assembler: profile.assembler, + }) + .await + .map_err(|e| ToolFailure::Failed(format!("delegate: start loop failed: {e}")))?; + + let outcome = child.join().await; + + self.catalog.on_child_closed(child_frame).await; + let _ = self.store.close_frame(child_frame).await; + + let result_preview = |s: &str| preview_truncate(s, 500); + let emit_done = |text: &str| { + if let Some(ev) = &events { + ev.emit(child_frame, Some(ctx.frame), LoopEvent::AgentFinished { + frame: child_frame, + agent: agent_id.to_string(), + result_preview: result_preview(text), + parent_agent: ctx.agent.clone(), + }); + } + }; + + match outcome { + Ok(crate::kernel::TurnOutcome::Final { content, .. }) => { + emit_done(&content); + Ok(ToolOutput::Text(content)) + } + Ok(crate::kernel::TurnOutcome::Cancelled) => { + emit_done("⚠️ Cancelled."); + Ok(ToolOutput::Text(format!("Sub-agent `{agent_id}` was cancelled."))) + } + Ok(crate::kernel::TurnOutcome::Exhausted) => { + emit_done("⚠️ Exhausted tool-call rounds."); + Ok(ToolOutput::Text(format!( + "Sub-agent `{agent_id}` exceeded the tool-call round budget without producing a final answer." + ))) + } + Err(e) => { + emit_done(&format!("⚠️ Error: {e}")); + Err(ToolFailure::Failed(format!("Sub-agent `{agent_id}` failed: {e}"))) + } + } + } +} + +#[async_trait] +impl Tool for DelegateTool { + fn name(&self) -> &str { &self.name } + + fn definition(&self) -> Value { + if let Some(def) = &self.definition_override { + return def.clone(); + } + json!({ + "type": "function", + "function": { + "name": self.name, + "description": "Delegate a task to a sub-agent and wait for its result. \ + Use for focused, well-scoped work that benefits from a clean context.", + "parameters": self.schema(), + } + }) + } + + /// Sync delegates batch: a homogeneous fan-out runs them concurrently + /// (the kernel allocates ids in order first — results never mix). + fn concurrency_safe(&self, args: &Value) -> bool { + args["mode"].as_str() != Some("async") + } + + async fn call(&self, args: Value, ctx: &ToolCtx) -> Result { + let agent_id = args["agent_id"] + .as_str() + .ok_or_else(|| ToolFailure::Failed("delegate: missing required argument `agent_id`".into()))?; + let prompt = args["prompt"] + .as_str() + .ok_or_else(|| ToolFailure::Failed("delegate: missing required argument `prompt`".into()))?; + + match args["mode"].as_str() { + Some("async") => Err(ToolFailure::Failed( + "delegate: async mode rides the host's AsyncExecutor, which is not wired on this path" + .to_string(), + )), + _ => self.run_sync(agent_id, prompt, ctx).await, + } + } +} + +/// Truncate to `max` chars with an ellipsis (previews). +pub fn preview_truncate(s: &str, max: usize) -> String { + if s.chars().count() <= max { + return s.to_string(); + } + let cut: String = s.chars().take(max.saturating_sub(1)).collect(); + format!("{cut}…") +} + +/// A static catalog for tests and simple hosts. +pub struct StaticCatalog { + profiles: Vec, +} + +impl StaticCatalog { + pub fn new() -> Self { Self { profiles: Vec::new() } } + + pub fn with(mut self, profile: AgentProfile) -> Self { + self.profiles.push(profile); + self + } +} + +impl Default for StaticCatalog { + fn default() -> Self { Self::new() } +} + +#[async_trait] +impl AgentCatalog for StaticCatalog { + async fn get(&self, id: &str, _child_frame: FrameId) -> crate::Result { + self.profiles + .iter() + .find(|p| p.id == id) + .cloned() + .ok_or_else(|| anyhow::anyhow!("unknown agent `{id}`")) + } + + async fn list(&self, kind: AgentKind) -> Vec { + self.profiles + .iter() + .filter(|p| p.kind == kind) + .map(|p| AgentSummary { id: p.id.clone(), kind: p.kind, description: String::new() }) + .collect() + } +} diff --git a/crates/agent-loop/src/events.rs b/crates/agent-loop/src/events.rs index 3e7236a..d37059c 100644 --- a/crates/agent-loop/src/events.rs +++ b/crates/agent-loop/src/events.rs @@ -65,6 +65,8 @@ pub enum LoopEvent { id: ToolCallId, name: String, args: Value, + /// The approval request id in the host's registry (for UI resolution). + request_id: i64, }, // ── sub-agents (emitted by child loops; parent_frame in the tag) ── AgentSpawned { @@ -72,11 +74,15 @@ pub enum LoopEvent { agent: String, depth: u32, prompt_preview: String, + /// The parent frame's tool call that spawned this agent. + parent_call: ToolCallId, + parent_agent: String, }, AgentFinished { frame: FrameId, agent: String, result_preview: String, + parent_agent: String, }, AsyncResultReady { task: TaskId, diff --git a/crates/agent-loop/src/gate.rs b/crates/agent-loop/src/gate.rs index 4b8a378..e61d415 100644 --- a/crates/agent-loop/src/gate.rs +++ b/crates/agent-loop/src/gate.rs @@ -17,6 +17,7 @@ pub struct PendingCall { pub name: String, pub args: Value, pub frame: FrameId, + pub parent_frame: Option, pub agent: String, /// Host free-form (source, permission group, …). pub extensions: Extensions, diff --git a/crates/agent-loop/src/human.rs b/crates/agent-loop/src/human.rs index 7d2ba67..bc32620 100644 --- a/crates/agent-loop/src/human.rs +++ b/crates/agent-loop/src/human.rs @@ -19,6 +19,8 @@ pub struct Question { pub suggested: Vec, /// The tool call asking (for UI correlation). pub call: ToolCallId, + /// The frame asking (for event tagging). + pub frame: crate::ids::FrameId, } /// The human channel closed while waiting (WS down, user gone). @@ -46,23 +48,30 @@ pub trait HumanChannel: Send + Sync { pub struct AskUserTool { channel: Arc, store: Arc, + name: String, } impl AskUserTool { pub fn new(channel: Arc, store: Arc) -> Self { - Self { channel, store } + Self { channel, store, name: "ask_user".to_string() } + } + + /// Register under a legacy name (Skald's `ask_user_clarification`, D11). + pub fn with_name(mut self, name: impl Into) -> Self { + self.name = name.into(); + self } } #[async_trait] impl Tool for AskUserTool { - fn name(&self) -> &str { "ask_user" } + fn name(&self) -> &str { &self.name } fn definition(&self) -> Value { json!({ "type": "function", "function": { - "name": "ask_user", + "name": self.name, "description": "Ask the user a clarifying question and wait for the answer.", "parameters": { "type": "object", @@ -70,7 +79,9 @@ impl Tool for AskUserTool { "title": { "type": "string", "description": "Short title of the question" }, "question": { "type": "string", "description": "The question to ask" }, "suggested": { "type": "array", "items": { "type": "string" }, - "description": "Optional suggested answers" } + "description": "Optional suggested answers" }, + "suggested_answers": { "type": "array", "items": { "type": "string" }, + "description": "Optional suggested answers (legacy alias of `suggested`)" } }, "required": ["question"] } @@ -79,14 +90,17 @@ impl Tool for AskUserTool { } async fn call(&self, args: Value, ctx: &ToolCtx) -> Result { + let suggested = args["suggested"] + .as_array() + .or_else(|| args["suggested_answers"].as_array()) + .map(|a| a.iter().filter_map(|v| v.as_str().map(str::to_string)).collect()) + .unwrap_or_default(); let q = Question { title: args["title"].as_str().unwrap_or("Question").to_string(), question: args["question"].as_str().unwrap_or("").to_string(), - suggested: args["suggested"] - .as_array() - .map(|a| a.iter().filter_map(|v| v.as_str().map(str::to_string)).collect()) - .unwrap_or_default(), + suggested, call: ctx.call_id, + frame: ctx.frame, }; // Durability FIRST: the call must survive a crash as AwaitingHuman. self.store diff --git a/crates/agent-loop/src/kernel.rs b/crates/agent-loop/src/kernel.rs index 4d7e74f..2b39495 100644 --- a/crates/agent-loop/src/kernel.rs +++ b/crates/agent-loop/src/kernel.rs @@ -83,18 +83,23 @@ pub(crate) async fn run( events: events.clone(), }; - // ToolCtx extensions: host extensions + the event sink, so shipped tools - // (ask_user, activate_tools) can emit out-of-band. + // ToolCtx extensions: host extensions + the event sink + the turn's tool + // set, so shipped tools (ask_user, activate_tools, delegate) reach what + // they need. let tool_extensions = || { let mut ext = params.extensions.clone(); ext.insert(Arc::new(events.clone())); + ext.insert(Arc::new(crate::tool::SharedToolSet(params.tools.clone()))); ext }; events.emit(frame, parent, LoopEvent::TurnStarted); + // Per-loop selector override (sub-agents with their own strength, D14). + let selector: &Arc = params.selector.as_ref().unwrap_or(&deps.models); + // First selection of the turn. - let mut handle: ModelHandle = match deps.models.select(¶ms.model_hint, &[]).await { + let mut handle: ModelHandle = match selector.select(¶ms.model_hint, &[]).await { Ok(h) => h, Err(e) => { events.emit(frame, parent, LoopEvent::Error(format!("model selection failed: {e}"))); @@ -171,11 +176,11 @@ pub(crate) async fn run( match result { Ok(resp) => { - deps.models.report_success(&handle.id).await; + selector.report_success(&handle.id).await; break resp; } Err(e) => { - deps.models.report_failure(&handle.id, &e.to_string()).await; + selector.report_failure(&handle.id, &e.to_string()).await; let retriable = handle.model.is_retriable(&e); warn!(model = %handle.id, error = %e, retriable, "llm call failed"); if !retriable || tried.len() >= deps.retry.max_attempts { @@ -185,7 +190,7 @@ pub(crate) async fn run( }); return Err(anyhow!("llm call failed on {}: {e}", handle.id)); } - match deps.models.select(¶ms.model_hint, &tried).await { + match selector.select(¶ms.model_hint, &tried).await { Ok(next) => { events.emit(frame, parent, LoopEvent::ModelFallback { from: handle.id.clone(), @@ -527,6 +532,7 @@ async fn pre_execution( name: ptc.name.clone(), args: ptc.arguments.clone(), frame: params.frame, + parent_frame: params.parent_frame, agent: params.agent.clone(), extensions: params.extensions.clone(), }; diff --git a/crates/agent-loop/src/lib.rs b/crates/agent-loop/src/lib.rs index 1c9ca70..66d1b40 100644 --- a/crates/agent-loop/src/lib.rs +++ b/crates/agent-loop/src/lib.rs @@ -13,6 +13,7 @@ pub mod activation; pub mod context; +pub mod delegate; pub mod events; pub mod gate; pub mod hooks; @@ -46,6 +47,10 @@ pub mod prelude { AssembleInput, ContextAssembler, LinearAssembler, StaticSystemContext, SystemContext, SystemContextSource, TurnInfo, }; + pub use crate::delegate::{ + AgentCatalog, AgentKind, AgentProfile, AgentSummary, DelegateTool, FilteredToolSet, + StaticCatalog, ToolSelection, + }; pub use crate::events::{DeltaKind, Event, EventSink, LoopEvent}; pub use crate::gate::{AllowAll, DenyList, Gate, GateDecision, PendingCall}; pub use crate::hooks::{HookCtx, HookVerdict, LoopHooks}; diff --git a/crates/agent-loop/src/manager.rs b/crates/agent-loop/src/manager.rs index 5ba3384..dc2bd25 100644 --- a/crates/agent-loop/src/manager.rs +++ b/crates/agent-loop/src/manager.rs @@ -76,6 +76,12 @@ pub struct LoopParams { pub system: Arc, pub tools: Arc, pub model_hint: ModelHint, + /// Per-loop selector override (e.g. a sub-agent with its own strength, + /// blueprint D14). `None` = the manager's selector. + pub selector: Option>, + /// Parent-linked cancellation (DelegateTool passes `ctx.cancel.child_token()`): + /// `None` = a fresh scope. Cancellation stays sticky down the tree. + pub token: Option, pub live_input: Option>, pub extensions: Extensions, pub meta: TurnMeta, @@ -206,6 +212,8 @@ impl LoopManager { system: params.system, tools: params.tools, model_hint: params.model_hint, + selector: None, + token: None, live_input: params.live_input, extensions: params.extensions, meta: params.meta, @@ -215,14 +223,26 @@ impl LoopManager { // ── raw loops (DelegateTool, recovery, background runners) ── + /// Spawn a raw loop. Unlike `start_turn` this does NOT enforce the + /// one-loop-per-conversation rule and does NOT register in the live + /// registry: child loops (sub-agents, including concurrent batches) run + /// on the same conversation as their parent and are cancelled through + /// the parent's token tree (`child_token()`), not the registry. pub async fn start_loop(&self, params: LoopParams) -> Result { - { - let registry = self.registry.lock().unwrap(); - if registry.contains_key(¶ms.conversation) { - return Err(StartError::AlreadyRunning); - } - } - self.spawn(params) + self.spawn_detached(params) + } + + fn spawn_detached(&self, params: LoopParams) -> Result { + let conv = params.conversation.clone(); + let frame = params.frame; + let token = params.token.clone().unwrap_or_default(); + let events = self.sink(conv.clone()); + + let deps = self.deps.clone(); + let turn_token = token.clone(); + let join = tokio::spawn(async move { crate::kernel::run(deps, params, turn_token, events).await }); + + Ok(TurnHandle { conversation: conv, frame, cancel: token, join }) } fn spawn(&self, params: LoopParams) -> Result { diff --git a/crates/agent-loop/src/store.rs b/crates/agent-loop/src/store.rs index 06508a9..259990b 100644 --- a/crates/agent-loop/src/store.rs +++ b/crates/agent-loop/src/store.rs @@ -75,13 +75,16 @@ impl CallOutcome { } } - /// Text persisted as the call's result (what the model will read back). + /// Text persisted as the call's result. Kept RAW (the assembler formats + /// for the model: `Failed` results get their "Error:" prefix at + /// projection time, not here) so hosts with an existing schema (Skald's + /// `chat_llm_tools.result`) round-trip byte-identically. pub fn result_text(&self) -> String { match self { Self::Completed(out) => out.to_wire(), - Self::Failed(e) => format!("Error: {e}"), - Self::Cancelled => "Tool call cancelled by user.".to_string(), - Self::Rejected { reason } => format!("Tool call rejected: {reason}"), + Self::Failed(e) => e.clone(), + Self::Cancelled => "Cancelled by user.".to_string(), + Self::Rejected { reason } => reason.clone(), } } @@ -253,6 +256,8 @@ pub trait HistoryStore: Send + Sync { spec: FrameSpec, ) -> crate::Result; async fn close_frame(&self, frame: FrameId) -> crate::Result<()>; + /// One frame by id (DelegateTool depth checks, recovery). + async fn get_frame(&self, frame: FrameId) -> crate::Result>; /// All active frames of a conversation (recovery: batch detection, cascade). async fn active_frames(&self, conv: &ConversationId) -> crate::Result>; async fn deepest_active(&self, conv: &ConversationId) -> crate::Result>; @@ -273,6 +278,11 @@ pub trait HistoryStore: Send + Sync { async fn resolve_call(&self, id: ToolCallId, outcome: &CallOutcome) -> crate::Result<()>; /// Only `Running → AwaitingHuman`. async fn set_call_state(&self, id: ToolCallId, state: CallState) -> crate::Result<()>; + /// One call by id (translators enriching finish events, recovery). + async fn get_call(&self, id: ToolCallId) -> crate::Result>; + /// Merge host free-form extras into a call (Skald: diff preview, media). + /// Keys not understood by the store are ignored. + async fn set_call_extras(&self, id: ToolCallId, extras: Value) -> crate::Result<()>; async fn calls_in_state(&self, frame: FrameId, states: &[CallState]) -> crate::Result>; // ── summaries ── diff --git a/crates/agent-loop/src/store_memory.rs b/crates/agent-loop/src/store_memory.rs index add64a8..5cde8fd 100644 --- a/crates/agent-loop/src/store_memory.rs +++ b/crates/agent-loop/src/store_memory.rs @@ -69,6 +69,11 @@ impl HistoryStore for InMemoryStore { Ok(()) } + async fn get_frame(&self, frame: FrameId) -> crate::Result> { + let i = self.inner.lock().unwrap(); + Ok(i.frames.get(&frame).cloned()) + } + async fn active_frames(&self, conv: &ConversationId) -> crate::Result> { let i = self.inner.lock().unwrap(); Ok(i.frames.values().filter(|f| f.active && &f.conversation == conv).cloned().collect()) @@ -185,6 +190,25 @@ impl HistoryStore for InMemoryStore { Ok(()) } + async fn get_call(&self, id: ToolCallId) -> crate::Result> { + let i = self.inner.lock().unwrap(); + Ok(i.calls.values().flatten().find(|c| c.id == id).cloned()) + } + + async fn set_call_extras(&self, id: ToolCallId, extras: serde_json::Value) -> crate::Result<()> { + let mut i = self.inner.lock().unwrap(); + update_call(&mut i, id, |c| { + if let (Some(dst), Some(src)) = (c.extras.as_object_mut(), extras.as_object()) { + for (k, v) in src { + dst.insert(k.clone(), v.clone()); + } + } else { + c.extras = extras.clone(); + } + }); + Ok(()) + } + async fn calls_in_state(&self, frame: FrameId, states: &[CallState]) -> crate::Result> { let i = self.inner.lock().unwrap(); Ok(i.messages diff --git a/crates/agent-loop/src/tool.rs b/crates/agent-loop/src/tool.rs index 1ef2218..b93d4bd 100644 --- a/crates/agent-loop/src/tool.rs +++ b/crates/agent-loop/src/tool.rs @@ -207,6 +207,12 @@ pub trait ToolSet: Send + Sync { fn find(&self, name: &str) -> Option>; } +/// Wrapper so `Arc` can ride in [`Extensions`] (type-map keys +/// must be `Sized`). The kernel inserts one into every `ToolCtx`; shipped +/// tools that spawn child loops (delegate) inherit from it. +#[derive(Clone)] +pub struct SharedToolSet(pub Arc); + /// A trivial `ToolSet` from a list of tools (testing, simple hosts). pub struct ToolRegistry { tools: Vec>, diff --git a/crates/agent-loop/tests/kernel.rs b/crates/agent-loop/tests/kernel.rs index c638b9a..f138dfe 100644 --- a/crates/agent-loop/tests/kernel.rs +++ b/crates/agent-loop/tests/kernel.rs @@ -15,6 +15,7 @@ use agent_loop::store_memory::InMemoryStore; use agent_loop::testing::{self, FakeModel, Step}; use agent_loop::tool::{Tool, ToolCtx, ToolFailure, ToolOutput, ToolRegistry}; use agent_loop::context::StaticSystemContext; +use agent_loop::delegate::{AgentCatalog, AgentKind, AgentProfile, DelegateTool, ToolSelection}; use agent_loop::events::LoopEvent; use serde_json::{Value, json}; use tokio_util::sync::CancellationToken; @@ -496,3 +497,140 @@ async fn second_loop_on_same_conversation_rejected() { handle.cancel.cancel(); let _ = handle.join().await; } + +// ── delegate (sub-agents as a tool) ── + +struct TestCatalog { + context: Arc, +} + +#[async_trait] +impl AgentCatalog for TestCatalog { + async fn get(&self, id: &str, _child_frame: agent_loop::ids::FrameId) -> agent_loop::Result { + Ok(AgentProfile { + id: id.into(), + kind: AgentKind::Task, + context: self.context.clone(), + tools: ToolSelection::inherit(), + toolset: None, + model: None, + selector: None, + assembler: None, + }) + } + async fn list(&self, _kind: AgentKind) -> Vec { + Vec::new() + } +} + +#[tokio::test] +async fn sync_delegate_runs_child_loop_and_returns_result() { + let script = vec![ + Step::tool_calls("delegating", vec![testing::call("c1", "delegate", json!({"agent_id":"researcher","prompt":"find X"}))]), + Step::message("research says: X=42"), + Step::message("final answer with X=42"), + ]; + let store = Arc::new(InMemoryStore::new()); + let manager = Arc::new( + LoopManager::builder() + .models(Arc::new(agent_loop::model::SingleModel::new(FakeModel::new("m", script)))) + .store(store.clone()) + .build() + .unwrap(), + ); + let catalog: Arc = Arc::new(TestCatalog { + context: Arc::new(StaticSystemContext::new("You are a researcher.")), + }); + let delegate: Arc = Arc::new(DelegateTool::new(manager.clone(), catalog, manager.store(), 5)); + + let conv = ConversationId::new("d1"); + let frame = manager.open_root(&conv, FrameSpec::root("assistant")).await.unwrap(); + let p = TurnParams { + frame, + agent: "assistant".into(), + system: Arc::new(StaticSystemContext::new("root")), + tools: ToolRegistry::new().with_arc(delegate).into_toolset(), + model_hint: ModelHint::default(), + live_input: None, + extensions: Default::default(), + meta: TurnMeta::default(), + assembler: None, + }; + + let handle = manager.start_turn(conv.clone(), NewMessage::user("what is X?"), p).await.unwrap(); + let outcome = tokio::time::timeout(Duration::from_secs(5), handle.join()) + .await + .expect("delegate turn hung") + .unwrap(); + let TurnOutcome::Final { content, .. } = outcome else { panic!("expected Final, got {outcome:?}") }; + assert_eq!(content, "final answer with X=42"); + + // The parent's delegate call resolved Done with the CHILD's answer as result. + let done = store.calls_in_state(frame, &[CallState::Done]).await.unwrap(); + assert_eq!(done.len(), 1); + assert_eq!(done[0].result.as_deref(), Some("research says: X=42")); + + // The child frame exists, closed, with its Agent prompt + assistant answer. + let frames = store.active_frames(&conv).await.unwrap(); + assert!(frames.iter().all(|f| f.spec.depth == 0), "child frame must be closed"); + let history_all = store.load(frame).await.unwrap(); + assert!(history_all.iter().any(|m| m.role == agent_loop::store::Role::Assistant && m.content == "final answer with X=42")); +} + +#[tokio::test] +async fn delegate_batch_fans_out_concurrently() { + let script = vec![ + Step::tool_calls("", vec![ + testing::call("c1", "delegate", json!({"agent_id":"a1","prompt":"job one"})), + testing::call("c2", "delegate", json!({"agent_id":"a2","prompt":"job two"})), + ]), + Step::message("result one"), + Step::message("result two"), + Step::message("both done"), + ]; + let store = Arc::new(InMemoryStore::new()); + let manager = Arc::new( + LoopManager::builder() + .models(Arc::new(agent_loop::model::SingleModel::new(FakeModel::new("m", script)))) + .store(store.clone()) + .max_parallel_calls(2) + .build() + .unwrap(), + ); + let catalog: Arc = Arc::new(TestCatalog { + context: Arc::new(StaticSystemContext::new("worker")), + }); + let delegate: Arc = Arc::new(DelegateTool::new(manager.clone(), catalog, manager.store(), 5)); + + let conv = ConversationId::new("d2"); + let frame = manager.open_root(&conv, FrameSpec::root("assistant")).await.unwrap(); + let p = TurnParams { + frame, + agent: "assistant".into(), + system: Arc::new(StaticSystemContext::new("root")), + tools: ToolRegistry::new().with_arc(delegate).into_toolset(), + model_hint: ModelHint::default(), + live_input: None, + extensions: Default::default(), + meta: TurnMeta::default(), + assembler: None, + }; + + let handle = manager.start_turn(conv, NewMessage::user("do both"), p).await.unwrap(); + let outcome = tokio::time::timeout(Duration::from_secs(5), handle.join()) + .await + .expect("delegate batch hung") + .unwrap(); + assert!(matches!(outcome, TurnOutcome::Final { .. }), "got {outcome:?}"); + + // Both delegate calls resolved Done, each carrying one of the child results. + let done = store.calls_in_state(frame, &[CallState::Done]).await.unwrap(); + assert_eq!(done.len(), 2); + let results: HashSet = done.iter().filter_map(|c| c.result.clone()).collect(); + assert_eq!( + results, + ["result one".to_string(), "result two".to_string()].into_iter().collect() + ); +} + +use std::collections::HashSet; diff --git a/crates/core-api/src/interface_tool.rs b/crates/core-api/src/interface_tool.rs index 4c49fff..09465bb 100644 --- a/crates/core-api/src/interface_tool.rs +++ b/crates/core-api/src/interface_tool.rs @@ -9,6 +9,7 @@ pub type ToolFuture = Pin` + `ChatId`). +#[derive(Clone)] pub struct InterfaceTool { /// OpenAI-format tool definition sent to the LLM in the tools array. pub definition: Value, diff --git a/crates/honcho-client/Cargo.toml b/crates/honcho-client/Cargo.toml index 7f8c56d..509bb99 100644 --- a/crates/honcho-client/Cargo.toml +++ b/crates/honcho-client/Cargo.toml @@ -8,3 +8,8 @@ reqwest = { version = "0.13", default-features = false, features = ["rustls-no serde = { version = "1", features = ["derive"] } serde_json = "1" tracing = "0.1" + +[dev-dependencies] +# The crate-level doc example (`#[tokio::main]`) compiles under `cargo test`. +tokio = { version = "1", features = ["macros", "rt"] } +anyhow = "1" diff --git a/crates/plugin-tts-orpheus-3b/src/lib.rs b/crates/plugin-tts-orpheus-3b/src/lib.rs index cdaa3d5..cc43c3b 100644 --- a/crates/plugin-tts-orpheus-3b/src/lib.rs +++ b/crates/plugin-tts-orpheus-3b/src/lib.rs @@ -12,7 +12,7 @@ //! # Required secret //! //! Set before enabling the plugin: -//! ``` +//! ```text //! set_secret("HUGGINGFACE_TOKEN", "hf_...") //! ``` //! Get a token at . diff --git a/crates/skald-core/src/loop_adapters/assembler.rs b/crates/skald-core/src/loop_adapters/assembler.rs new file mode 100644 index 0000000..e735396 --- /dev/null +++ b/crates/skald-core/src/loop_adapters/assembler.rs @@ -0,0 +1,528 @@ +//! `SkaldAssembler` — Skald's history projection behind the crate's +//! `ContextAssembler` (port of `MessageBuilder::build`'s message-array half, +//! blueprint §10). Byte-parity with the current builder is the contract: +//! same layers, same tool-result texts, same DTL injections, same media rules. +//! +//! During phase 2 the old `MessageBuilder` still serves the legacy paths +//! (resume/recovery); the two are deleted together in phase 5. + +use std::sync::Arc; + +use agent_loop::activation::{ActivationSource, ToolRendering}; +use agent_loop::context::{AssembleInput, ContextAssembler}; +use agent_loop::store::{CallState, HistoryStore, Role}; +use core_api::message_meta::{MessageMetadata, attachments_block}; +use core_api::tool::MediaRef; +use core_api::user_fs::UserFs; +use serde_json::{Value, json}; + +use crate::compactor::SUMMARY_PREFIX; +use crate::config::DatetimeConfig; +use crate::loop_adapters::activation::SkaldActivationSource; +use crate::session::handler::media; +use crate::tools::tool_names as tn; + +/// Stand-in for a tool-call turn's `reasoning_content` when none was recorded +/// (DeepSeek's thinking mode 400s on replay without it). +const REASONING_ROUNDTRIP_PLACEHOLDER: &str = "(no reasoning recorded for this step)"; + +/// OS description (type + version), computed once. +fn os_description() -> &'static str { + static OS: std::sync::OnceLock = std::sync::OnceLock::new(); + OS.get_or_init(|| os_info::get().to_string()) +} + +/// System IANA timezone name, computed once. +fn system_timezone() -> Option<&'static str> { + static TZ: std::sync::OnceLock> = std::sync::OnceLock::new(); + TZ.get_or_init(|| iana_time_zone::get_timezone().ok()).as_deref() +} + +/// Skald's `ContextAssembler`: static system → scratchpad → summary → history +/// (with DTL + media) → dynamic tail (+datetime) → tail reminder. +pub struct SkaldAssembler { + /// Owner pool — scratchpad reads (keyed on `scratchpad_sid`). + pub pool: Arc, + /// Scratchpad scope (session_id, or the parent's for async sub-tasks). + pub scratchpad_sid: i64, + pub datetime_config: DatetimeConfig, + pub max_history_messages: usize, + pub max_tool_result_chars: Option, + /// The history window applies only when compaction is disabled. + pub compactor_enabled: bool, + /// The caller's fs view — media containment for inlining. `None` skips + /// media inlining entirely. + pub fs: Option>, + /// DTL activations (consulted only in non-Inline modes). + pub activation: Option, +} + +#[agent_loop::async_trait] +impl ContextAssembler for SkaldAssembler { + async fn build( + &self, + store: &Arc, + input: &AssembleInput, + ) -> agent_loop::Result> { + let mut out: Vec = Vec::new(); + + // ── 1. Static system message ────────────────────────────────────────── + let static_msg = if input.model.prompt_cache { + json!({ + "role": "system", + "content": [{ "type": "text", "text": input.system.base, "cache_control": { "type": "ephemeral" } }] + }) + } else { + json!({ "role": "system", "content": input.system.base }) + }; + out.push(static_msg); + + // ── 2. Scratchpad system message (before conversation) ──────────────── + let scratch = crate::db::scratchpad::for_session(&self.pool, self.scratchpad_sid).await?; + if !scratch.is_empty() { + let mut s = String::from( + "\n \ + \n" + ); + for (k, v) in &scratch { + s.push_str(&format!(" {v}\n")); + } + s.push_str(""); + out.push(json!({ "role": "system", "content": s })); + } + + // ── 3. Compaction summary + surviving history ───────────────────────── + let summary = store.latest_summary(input.frame).await?; + if let Some(s) = &summary { + out.push(json!({ + "role": "system", + "content": format!( + "{SUMMARY_PREFIX}\n\n{}\n\n\ + [End of context summary — the following messages are the most recent exchanges in full.]", + s.text + ) + })); + } + let mut history = match &summary { + Some(s) => store.load_since(input.frame, s.covered_up_to).await?, + None => store.load(input.frame).await?, + }; + + if !self.compactor_enabled && history.len() > self.max_history_messages { + history.drain(..history.len() - self.max_history_messages); + if matches!(history.first().map(|m| m.role), Some(Role::Assistant)) { + history.drain(..1); + } + } + + let current_turn_boundary = history + .iter() + .rposition(|e| matches!(e.role, Role::User | Role::Agent)); + + // Inline-media turn group: trailing assistant rows are the in-flight + // turn's own rounds; the current turn's user messages sit just before + // them. Older-turn media degrades to the textual path block. + let mut media_turn_start = history.len(); + while media_turn_start > 0 && matches!(history[media_turn_start - 1].role, Role::Assistant) { + media_turn_start -= 1; + } + while media_turn_start > 0 + && matches!(history[media_turn_start - 1].role, Role::User | Role::Agent) + { + media_turn_start -= 1; + } + + // DTL: tools activated at each assistant message (empty in Inline mode). + let activation_defs: std::collections::HashMap> = + match (&self.activation, input.model.tool_rendering) { + (Some(src), ToolRendering::Inline) => { + let _ = src; + Default::default() + } + (Some(src), _) => src + .activations(input.frame) + .await + .unwrap_or_default() + .into_iter() + .map(|a| (a.anchor.get(), a.defs)) + .collect(), + (None, _) => Default::default(), + }; + + // ── 4. Conversation history ─────────────────────────────────────────── + for (idx, entry) in history.iter().enumerate() { + let is_previous_turn = current_turn_boundary.is_some_and(|b| idx < b); + + match entry.role { + Role::System => {} + Role::User | Role::Agent => { + let metadata: Option = entry + .metadata + .as_ref() + .and_then(|v| serde_json::from_value(v.clone()).ok()); + let (text, media_parts) = match &metadata { + Some(meta) + if !meta.attachments.is_empty() + && idx >= media_turn_start + && self.fs.is_some() => + { + let fs = self.fs.as_deref().expect("guarded by is_some()"); + let partition = media::partition(&meta.attachments, &input.model.capabilities, fs).await; + ( + format!("{}{}", entry.content, attachments_block(&partition.rest)), + partition.parts, + ) + } + Some(meta) if !meta.attachments.is_empty() => ( + format!("{}{}", entry.content, attachments_block(&meta.attachments)), + Vec::new(), + ), + _ => (entry.content.clone(), Vec::new()), + }; + push_user_chunk(&mut out, text, media_parts); + } + Role::Assistant => { + if entry.calls.is_empty() { + let mut msg = json!({ "role": "assistant", "content": entry.content }); + if let Some(rc) = entry.reasoning.as_deref().filter(|s| !s.is_empty()) { + msg["reasoning_content"] = rc.into(); + msg["reasoning"] = rc.into(); + } + out.push(msg); + } else { + let tc_array: Vec = entry.calls + .iter() + .map(|tc| json!({ + "id": tc.provider_id, + "type": "function", + "function": { + "name": tc.name, + "arguments": serde_json::to_string(&tc.arguments) + .unwrap_or_else(|_| "{}".into()), + } + })) + .collect(); + + let mut msg = json!({ + "role": "assistant", + "content": entry.content, + "tool_calls": tc_array, + }); + // DeepSeek thinking mode: a tool-calling assistant turn must + // carry a NON-EMPTY reasoning_content on replay. + let rc = entry.reasoning.as_deref() + .filter(|s| !s.is_empty()) + .unwrap_or(REASONING_ROUNDTRIP_PLACEHOLDER); + msg["reasoning_content"] = rc.into(); + msg["reasoning"] = rc.into(); + out.push(msg); + + for tc in &entry.calls { + let result_content = match tc.state { + CallState::Done => tc.result.clone().unwrap_or_default(), + CallState::Failed => format!( + "Error: {}", + tc.result.as_deref().unwrap_or("unknown error") + ), + CallState::Rejected => tc.result.clone() + .unwrap_or_else(|| "User rejected this tool call.".to_string()), + CallState::Cancelled => tc.result.clone() + .unwrap_or_else(|| "Tool call was cancelled by the user.".to_string()), + // 'pending'/'running' left behind by a crash or a lost + // connection: the call really was interrupted mid-flight. + _ => "Error: tool call was interrupted (connection lost before user approval). Please retry the operation.".to_string(), + }; + + let result_content = self.maybe_hide_tool_result( + result_content, + is_previous_turn, + &tc.name, + &tc.arguments, + ); + + let mut tool_msg = json!({ + "role": "tool", + "tool_call_id": tc.provider_id, + "content": result_content, + }); + // Anthropic DTL: an `activate_tools` result becomes a set of + // `tool_reference`s. + if matches!(input.model.tool_rendering, ToolRendering::DeferredToolReference) + && tc.name == tn::ACTIVATE_TOOLS + && let Some(adefs) = activation_defs.get(&entry.id.get()) + { + let names: Vec = adefs.iter() + .filter_map(|d| d["function"]["name"].as_str()) + .map(|n| Value::String(n.to_string())) + .collect(); + if !names.is_empty() { + tool_msg["_tool_references"] = Value::Array(names); + } + } + out.push(tool_msg); + } + + // Tool-produced media of the current turn: inline as a + // synthetic `user` message right after the tool-result group. + if idx >= media_turn_start + && let Some(fs) = self.fs.as_deref() + { + let mut refs: Vec = Vec::new(); + for tc in &entry.calls { + if let Some(mj) = tc.extras["media"].as_str() + && let Ok(mut v) = serde_json::from_str::>(mj) + { + refs.append(&mut v); + } + } + if !refs.is_empty() { + let parts = media::inline_paths(&refs, &input.model.capabilities, fs).await; + if !parts.is_empty() { + out.push(json!({ "role": "user", "content": parts })); + } + } + } + + // Kimi K3 DTL: the tools activated at this assistant message, + // as a `system` message carrying a `tools` field, right after + // its tool-result group (append-only → cache-safe). + if matches!(input.model.tool_rendering, ToolRendering::SystemToolBlock) + && let Some(adefs) = activation_defs.get(&entry.id.get()) + && !adefs.is_empty() + { + out.push(json!({ "role": "system", "tools": adefs })); + } + } + } + } + } + + // ── 5. Dynamic tail (extra dynamic + datetime) ──────────────────────── + { + let datetime_line = self.datetime_line(); + let extra_dynamic = input.system.dynamic_tail.first().map(String::as_str); + let tail = match (extra_dynamic, datetime_line.as_deref()) { + (Some(dyn_ctx), Some(dt)) => Some(format!("{dyn_ctx}\n\n---\n{dt}")), + (Some(dyn_ctx), None) => Some(dyn_ctx.to_string()), + (None, Some(dt)) => Some(dt.to_string()), + (None, None) => None, + }; + if let Some(content) = tail { + out.push(json!({ "role": "system", "content": content })); + } + } + + // ── 6. Tail reminder ────────────────────────────────────────────────── + if let Some(reminder) = &input.system.tail_reminder { + out.push(json!({ "role": "system", "content": reminder })); + } + + Ok(out) + } +} + +impl SkaldAssembler { + /// The current date/time + OS + cwd block (empty when disabled). + fn datetime_line(&self) -> Option { + if !self.datetime_config.enabled { + return None; + } + let now_utc = chrono::Utc::now(); + let secs = now_utc.timestamp(); + + let secs = match self.datetime_config.round_minutes { + Some(m) if m > 0 => { + let bucket = (m as i64) * 60; + (secs / bucket) * bucket + } + _ => secs, + }; + + let tz = self.datetime_config.timezone.as_deref() + .and_then(|s| s.parse::().ok()) + .or_else(|| system_timezone().and_then(|s| s.parse::().ok())); + + let (formatted, tz_name) = match tz { + Some(tz) => { + use chrono::TimeZone as _; + let f = tz.timestamp_opt(secs, 0) + .single() + .map(|dt| dt.format("%Y-%m-%dT%H:%M:%S%:z").to_string()) + .unwrap_or_else(|| chrono::Local::now().format("%Y-%m-%dT%H:%M:%S%:z").to_string()); + (f, Some(tz.name().to_string())) + } + None => { + let f = chrono::DateTime::from_timestamp(secs, 0) + .map(|utc| utc.with_timezone(&chrono::Local).format("%Y-%m-%dT%H:%M:%S%:z").to_string()) + .unwrap_or_else(|| chrono::Local::now().format("%Y-%m-%dT%H:%M:%S%:z").to_string()); + (f, None) + } + }; + let date_line = match tz_name { + Some(name) => format!("Current date and time: {formatted} ({name})"), + None => format!("Current date and time: {formatted}"), + }; + + let cwd = "~"; + + Some(format!( + "{date_line}\nOperating system: {}\nWorking directory: {cwd}\n\ + Filesystem tools and execute_cmd resolve relative paths against your home directory.", + os_description() + )) + } + + /// Replaces an over-limit previous-turn result with an informative 1-liner. + fn maybe_hide_tool_result( + &self, + result: String, + is_previous_turn: bool, + tool_name: &str, + arguments: &Value, + ) -> String { + if !is_previous_turn { + return result; + } + let Some(limit) = self.max_tool_result_chars else { + return result; + }; + if result.len() <= limit { + return result; + } + summarize_tool_result(tool_name, arguments, &result) + } +} + +// ── Free helpers (ported verbatim from message_builder.rs) ───────────────────── + +/// Appends one user/agent chunk, coalescing with a preceding `user` message. +fn push_user_chunk(out: &mut Vec, text: String, media: Vec) { + fn text_part(t: &str) -> Value { + json!({ "type": "text", "text": t }) + } + + if let Some(last) = out.last_mut() + && last["role"] == "user" + { + if !last["content"].is_array() && media.is_empty() { + let prev = last["content"].as_str().unwrap_or("").to_string(); + last["content"] = Value::String(format!("{prev}\n\n{text}")); + return; + } + let mut parts = match last["content"].take() { + Value::Array(a) => a, + Value::String(s) => vec![text_part(&s)], + _ => Vec::new(), + }; + if let Some(tp) = parts.iter_mut().rev().find(|p| p["type"] == "text") { + let prev = tp["text"].as_str().unwrap_or("").to_string(); + tp["text"] = Value::String(format!("{prev}\n\n{text}")); + } else { + parts.insert(0, text_part(&text)); + } + parts.extend(media); + last["content"] = Value::Array(parts); + return; + } + if media.is_empty() { + out.push(json!({ "role": "user", "content": text })); + } else { + let mut parts = vec![text_part(&text)]; + parts.extend(media); + out.push(json!({ "role": "user", "content": parts })); + } +} + +/// Creates an informative 1-line summary of a tool call result. +fn summarize_tool_result(tool_name: &str, arguments: &Value, result: &str) -> String { + let args = arguments; + + let char_count = result.len(); + let line_count = if result.trim().is_empty() { 0 } else { result.lines().count() }; + + fn arg_str<'a>(args: &'a serde_json::Value, key: &str) -> &'a str { + args[key].as_str().unwrap_or("?") + } + + match tool_name { + tn::EXECUTE_CMD => { + let cmd = args["command"].as_str().unwrap_or(""); + let cmd_display = crate::session::handler::preview_truncate(cmd, 77); + let exit_code = result + .lines() + .next() + .and_then(|l| l.strip_prefix("exit: ")) + .unwrap_or("?"); + format!("[execute_cmd] ran `{cmd_display}` → exit {exit_code}, {line_count} lines output") + } + + "read_file" | "read_file_chunk" => { + let path = arg_str(args, "path"); + format!("[{tool_name}] read {path} ({char_count} chars)") + } + + "write_file" => { + let path = arg_str(args, "path"); + format!("[write_file] wrote to {path}") + } + + "edit_file" | "patch_file" => { + let path = arg_str(args, "path"); + format!("[{tool_name}] edited {path}") + } + + "list_dir" | "glob" => { + let path = args["path"].as_str() + .or_else(|| args["pattern"].as_str()) + .unwrap_or("?"); + format!("[{tool_name}] {path} ({char_count} chars)") + } + + "list_items" => { + let kind = arg_str(args, "type"); + format!("[list_items] {kind} ({char_count} chars)") + } + + "toggle_item" => { + let kind = arg_str(args, "kind"); + let id = arg_str(args, "id"); + let enabled = args["enabled"].as_bool().unwrap_or(false); + format!("[toggle_item] {kind} '{id}' → {}", if enabled { "enabled" } else { "disabled" }) + } + + tn::READ_NOTIFICATION => { + let count = serde_json::from_str::>(result) + .map(|v| v.len()) + .unwrap_or(0); + format!("[read_notification] {count} notification(s)") + } + + tn::EXECUTE_TASK | tn::EXECUTE_SUBTASK => { + let agent = arg_str(args, "agent_id"); + format!("[{tool_name}] → {agent} ({char_count} chars result)") + } + + tn::ACTIVATE_TOOLS => { + let groups = args["groups"] + .as_array() + .map(|a| a.iter().filter_map(|v| v.as_str()).collect::>().join(", ")) + .unwrap_or_else(|| "?".to_string()); + format!("[activate_tools] loaded: {groups}") + } + + _ if tool_name.starts_with("mcp__") => { + format!("[{tool_name}] ({char_count} chars result)") + } + + _ => { + let first_arg = args.as_object() + .and_then(|m| m.iter().next()) + .map(|(k, v)| { + let sv = crate::session::handler::preview_truncate(v.as_str().unwrap_or_default(), 40); + format!(" {k}={sv}") + }) + .unwrap_or_default(); + format!("[{tool_name}]{first_arg} ({char_count} chars result)") + } + } +} diff --git a/crates/skald-core/src/loop_adapters/builtins.rs b/crates/skald-core/src/loop_adapters/builtins.rs new file mode 100644 index 0000000..5f387ab --- /dev/null +++ b/crates/skald-core/src/loop_adapters/builtins.rs @@ -0,0 +1,273 @@ +//! Skald's side of the crate's built-in tools: the `HumanChannel` +//! (clarification manager + interactive `AgentQuestion`), scratchpad/todos +//! tools, and the legacy-name aliases (`execute_task` sync/async composition, +//! `ask_user_clarification`, interface tools). + +use std::sync::Arc; + +use agent_loop::async_trait; +use agent_loop::delegate::DelegateTool; +use agent_loop::events::{EventSink, LoopEvent}; +use agent_loop::human::{HumanChannel, HumanGone, Question}; +use agent_loop::tool::{Tool, ToolCtx, ToolFailure, ToolOutput}; +use serde_json::{Value, json}; +use sqlx::SqlitePool; + +use crate::clarification::ClarificationManager; +use core_api::interface_tool::ToolFuture; + +// ── SkaldHumanChannel ──────────────────────────────────────────────────────── + +/// The `ask_user` backend: registers in `ClarificationManager` (so the +/// question lands in the Inbox for EVERY session kind) and, for interactive +/// sessions, also emits `AgentQuestion` inline in the chat (via +/// `LoopEvent::Host`). Port of `dispatch_ask_user_clarification`. +pub struct SkaldHumanChannel { + clarification: Arc, + session_id: i64, + agent_id: String, + source: String, + is_interactive: bool, + context_label: Arc>>, +} + +impl SkaldHumanChannel { + pub fn new( + clarification: Arc, + session_id: i64, + agent_id: impl Into, + source: impl Into, + is_interactive: bool, + context_label: Arc>>, + ) -> Self { + Self { + clarification, + session_id, + agent_id: agent_id.into(), + source: source.into(), + is_interactive, + context_label, + } + } +} + +#[async_trait] +impl HumanChannel for SkaldHumanChannel { + async fn ask(&self, q: Question, events: &EventSink) -> Result { + let label = self.context_label.read().ok().and_then(|g| g.clone()); + let (request_id, rx) = self + .clarification + .register( + self.session_id, + &self.agent_id, + &self.source, + label.as_deref(), + &q.title, + &q.question, + q.suggested.clone(), + ) + .await; + + if self.is_interactive { + events.emit(q.frame, None, LoopEvent::Host(json!({ + "type": "agent_question", + "request_id": request_id, + "tool_call_id": q.call.get(), + "title": q.title, + "question": q.question, + "suggested_answers": q.suggested, + }))); + } + + // The answer arrives via WS (resolve_question) or the Inbox REST. A + // session-wide cancel (WS drop) closes the channel → HumanGone → the + // tool suspends and the call stays pending for resume. + rx.await.map_err(|_| HumanGone) + } +} + +// ── UpdateScratchpadTool ───────────────────────────────────────────────────── + +/// The session-scoped shared blackboard (port of `dispatch_update_scratchpad`). +pub struct UpdateScratchpadTool { + pool: Arc, + sid: i64, +} + +impl UpdateScratchpadTool { + pub fn new(pool: Arc, sid: i64) -> Self { Self { pool, sid } } +} + +#[async_trait] +impl Tool for UpdateScratchpadTool { + fn name(&self) -> &str { crate::tools::tool_names::UPDATE_SCRATCHPAD } + + fn definition(&self) -> Value { + crate::session::handler::update_scratchpad_tool_def() + } + + async fn call(&self, args: Value, _ctx: &ToolCtx) -> Result { + let key = args["key"].as_str().unwrap_or("").to_string(); + let value = args["value"].as_str().unwrap_or("").to_string(); + crate::db::scratchpad::upsert(&self.pool, self.sid, &key, &value) + .await + .map(|_| ToolOutput::Text(format!("Scratchpad updated: {key}"))) + .map_err(|e| ToolFailure::Failed(e.to_string())) + } +} + +// ── WriteTodosTool ─────────────────────────────────────────────────────────── + +/// Stateless checklist echo (port of `dispatch_write_todos`). +pub struct WriteTodosTool; + +#[async_trait] +impl Tool for WriteTodosTool { + fn name(&self) -> &str { crate::tools::tool_names::WRITE_TODOS } + + fn definition(&self) -> Value { + crate::session::handler::write_todos_tool_def() + } + + async fn call(&self, args: Value, _ctx: &ToolCtx) -> Result { + let items = args["todos"].as_array().ok_or_else(|| { + ToolFailure::Failed("`write_todos` requires a `todos` array. Re-send the full list, e.g. [{\"content\":\"...\",\"status\":\"pending\"}].".into()) + })?; + if items.is_empty() { + return Err(ToolFailure::Failed("`todos` is empty — send at least one item, or omit the call entirely.".into())); + } + + let mut lines = Vec::with_capacity(items.len()); + let (mut done, mut active, mut pending) = (0usize, 0usize, 0usize); + for item in items { + let content = item["content"].as_str().unwrap_or("").trim(); + if content.is_empty() { + continue; + } + let marker = match item["status"].as_str() { + Some("completed") => { done += 1; "x" } + Some("in_progress") => { active += 1; "~" } + _ => { pending += 1; " " } + }; + lines.push(format!("[{marker}] {content}")); + } + if lines.is_empty() { + return Err(ToolFailure::Failed("No valid todo items (every `content` was empty).".into())); + } + + Ok(ToolOutput::Text(format!( + "Todo list ({total}): {done} done, {active} in progress, {pending} pending\n{body}", + total = lines.len(), + body = lines.join("\n"), + ))) + } +} + +// ── SkaldAskUserTool ───────────────────────────────────────────────────────── + +/// The legacy `ask_user_clarification`: the crate's `AskUserTool` mechanics +/// (AwaitingHuman + Suspend) with Skald's exact legacy definition. +pub struct SkaldAskUserTool { + inner: agent_loop::human::AskUserTool, +} + +impl SkaldAskUserTool { + pub fn new(channel: Arc, store: Arc) -> Self { + Self { + inner: agent_loop::human::AskUserTool::new(channel, store) + .with_name(crate::tools::tool_names::ASK_USER_CLARIFICATION), + } + } +} + +#[async_trait] +impl Tool for SkaldAskUserTool { + fn name(&self) -> &str { crate::tools::tool_names::ASK_USER_CLARIFICATION } + + fn definition(&self) -> Value { + crate::session::handler::ask_user_clarification_tool_def() + } + + async fn call(&self, args: Value, ctx: &ToolCtx) -> Result { + self.inner.call(args, ctx).await + } +} + +// ── ExecuteTaskAliasTool ───────────────────────────────────────────────────── + +/// The legacy `execute_task`: `mode=sync` (or unspecified) delegates to the +/// crate's `DelegateTool`; `mode=async` rides the legacy interface-tool +/// handler (ChatHub's task injection) until phase 3 wires `CronExecutor`. +pub struct ExecuteTaskAliasTool { + delegate: DelegateTool, + definition: Value, + async_handler: Option ToolFuture + Send + Sync>>, +} + +impl ExecuteTaskAliasTool { + pub fn new( + delegate: DelegateTool, + definition: Value, + async_handler: Option ToolFuture + Send + Sync>>, + ) -> Self { + Self { delegate, definition, async_handler } + } +} + +#[async_trait] +impl Tool for ExecuteTaskAliasTool { + fn name(&self) -> &str { crate::tools::tool_names::EXECUTE_TASK } + + fn definition(&self) -> Value { self.definition.clone() } + + fn concurrency_safe(&self, args: &Value) -> bool { + args["mode"].as_str() != Some("async") + } + + async fn call(&self, args: Value, ctx: &ToolCtx) -> Result { + if args["mode"].as_str() == Some("async") { + let Some(handler) = &self.async_handler else { + return Err(ToolFailure::Failed( + "execute_task: async mode is not available in this session".into(), + )); + }; + return handler(args) + .await + .map(ToolOutput::Text) + .map_err(|e| ToolFailure::Failed(e.to_string())); + } + self.delegate.call(args, ctx).await + } +} + +// ── LegacyInterfaceTool ────────────────────────────────────────────────────── + +/// Wraps a ChatHub-provided `InterfaceTool` (definition + handler closure) as +/// a crate-native tool — interface tools keep their exact legacy behavior +/// during the migration. +pub struct LegacyInterfaceTool { + definition: Value, + handler: Arc ToolFuture + Send + Sync>, +} + +impl LegacyInterfaceTool { + pub fn new(it: core_api::interface_tool::InterfaceTool) -> Self { + Self { definition: it.definition, handler: it.handler } + } +} + +#[async_trait] +impl Tool for LegacyInterfaceTool { + fn name(&self) -> &str { + self.definition["function"]["name"].as_str().unwrap_or("") + } + + fn definition(&self) -> Value { self.definition.clone() } + + async fn call(&self, args: Value, _ctx: &ToolCtx) -> Result { + (self.handler)(args) + .await + .map(ToolOutput::Text) + .map_err(|e| ToolFailure::Failed(e.to_string())) + } +} diff --git a/crates/skald-core/src/loop_adapters/catalog.rs b/crates/skald-core/src/loop_adapters/catalog.rs new file mode 100644 index 0000000..c04e367 --- /dev/null +++ b/crates/skald-core/src/loop_adapters/catalog.rs @@ -0,0 +1,279 @@ +//! `SkaldAgentCatalog` — the crate's `AgentCatalog` over `agents/*` +//! (port of `build_sub_agent_config`, blueprint §10): builds the child's +//! profile — its own prompt (never the parent's, B3), derived tool set +//! (root-only strip + sub-agent augmentation + approval visibility), own +//! strength selector (D14), own DTL-scoped assembler and activator. + +use std::collections::HashSet; +use std::sync::{Arc, RwLock}; + +use agent_loop::context::ContextAssembler; +use agent_loop::delegate::{AgentCatalog, AgentKind, AgentProfile, AgentSummary, DelegateTool, ToolSelection}; +use agent_loop::ids::FrameId; +use agent_loop::model::ModelHint; +use agent_loop::tool::Tool as LoopTool; +use agent_loop::activation::ActivateToolsTool; +use sqlx::SqlitePool; + +use crate::approval::ApprovalManager; +use crate::clarification::ClarificationManager; +use crate::config::DatetimeConfig; +use crate::llm::LlmManager; +use crate::loop_adapters::activation::SkaldToolActivator; +use crate::loop_adapters::assembler::SkaldAssembler; +use crate::loop_adapters::builtins::{SkaldAskUserTool, SkaldHumanChannel}; +use crate::loop_adapters::history::SqliteHistory; +use crate::loop_adapters::selector::SkaldSelector; +use crate::loop_adapters::system::AgentSystemContext; +use crate::loop_adapters::toolset::SkaldToolSet; +use crate::mcp::McpProvider; +use crate::tools::ToolRegistry; +use crate::tools::tool_names as tn; + +/// Everything the catalog needs from the parent turn, captured at wiring time. +pub struct SkaldAgentCatalog { + pool: Arc, + shared_pool: Arc, + user_id: String, + session_id: i64, + source: String, + is_interactive: bool, + context_label: Arc>>, + llm_manager: Arc, + approval: Arc, + clarification: Arc, + mcp: Arc, + registry: Arc, + /// Parent turn's derived def lists (the child's base derives from these). + base_defs: Vec, + config_defs: Arc>, + memory_tools: Vec>, + image_tools: Vec>, + core_tools: Vec>, + root_only: Vec, + /// The delegate tool, injected post-construction (catalog ↔ delegate cycle). + delegate: RwLock>>, + /// Per-turn assembler knobs shared with children. + datetime_config: DatetimeConfig, + max_history_messages: usize, + max_tool_result_chars: Option, + compactor_enabled: bool, + fs: Option>, + project_root: Option, +} + +impl SkaldAgentCatalog { + #[allow(clippy::too_many_arguments)] + pub fn new( + pool: Arc, + shared_pool: Arc, + user_id: String, + session_id: i64, + source: String, + is_interactive: bool, + context_label: Arc>>, + llm_manager: Arc, + approval: Arc, + clarification: Arc, + mcp: Arc, + registry: Arc, + base_defs: Vec, + config_defs: Arc>, + memory_tools: Vec>, + image_tools: Vec>, + root_only: Vec, + datetime_config: DatetimeConfig, + max_history_messages: usize, + max_tool_result_chars: Option, + compactor_enabled: bool, + fs: Option>, + project_root: Option, + ) -> Self { + let core_tools = registry.all_tools(); + Self { + pool, + shared_pool, + user_id, + session_id, + source, + is_interactive, + context_label, + llm_manager, + approval, + clarification, + mcp, + registry, + base_defs, + config_defs, + memory_tools, + image_tools, + core_tools, + root_only, + delegate: RwLock::new(None), + datetime_config, + max_history_messages, + max_tool_result_chars, + compactor_enabled, + fs, + project_root, + } + } + + /// Post-construction wiring of the delegate (the catalog ↔ delegate cycle). + pub fn set_delegate(&self, delegate: DelegateTool) { + *self.delegate.write().unwrap() = Some(Arc::new(delegate)); + } +} + +#[agent_loop::async_trait] +impl AgentCatalog for SkaldAgentCatalog { + async fn get(&self, id: &str, child_frame: FrameId) -> agent_loop::Result { + // Only `task` agents are dispatchable (rejects chat/system/unknown). + let meta = crate::agents::load_task_meta(id) + .map_err(|e| anyhow::anyhow!("{e}"))?; + + // The child's own strength drives its selector (D14) — never the + // parent's resolved client. + let selector = Arc::new(SkaldSelector::new(self.llm_manager.clone(), meta.strength)); + let model = meta.client.as_deref().map(ModelHint::name); + + // The child's system context: its own prompt, no per-turn extras. + let context = Arc::new(AgentSystemContext { + agent_id: id.to_string(), + extra_static: None, + extra_dynamic: None, + tail_reminder: None, + substitutions: Default::default(), + pool: self.pool.clone(), + shared_pool: self.shared_pool.clone(), + user_id: self.user_id.clone(), + mcp: self.mcp.clone(), + project_root: self.project_root.clone(), + }); + + // The child's def list: parent's base minus root-only minus the + // re-derived augmentations (added back natively below), plus + // sub-agents-only tools, through the approval visibility filter. + let mut child_defs: Vec = self + .base_defs + .iter() + .filter(|d| { + let name = d["function"]["name"].as_str().unwrap_or(""); + !self.root_only.iter().any(|n| n == name) + && name != tn::ASK_USER_CLARIFICATION + && name != tn::EXECUTE_SUBTASK + && name != tn::EXECUTE_TASK + }) + .cloned() + .collect(); + child_defs.extend(self.registry.openai_definitions_sub_agents_only()); + { + let group_rules = crate::db::approval_rules::list_for_group(&self.shared_pool, None) + .await + .unwrap_or_default(); + child_defs.retain(|def| { + let name = def["function"]["name"].as_str().unwrap_or(""); + self.approval.is_tool_visible(&group_rules, name) + }); + } + + // Native child tools: clarification, sub-delegation (depth permitting), + // and the frame-scoped activate_tools with a FRESH grant set. + let child_grants: Arc>> = Arc::new(RwLock::new( + crate::db::activated_tools::list_refs_stack(&self.pool, child_frame.get()) + .await + .unwrap_or_default() + .into_iter() + .collect(), + )); + + let mut native: Vec> = Vec::new(); + { + let channel = Arc::new(SkaldHumanChannel::new( + self.clarification.clone(), + self.session_id, + id, + &self.source, + self.is_interactive, + self.context_label.clone(), + )); + native.push(Arc::new(SkaldAskUserTool::new( + channel, + Arc::new(SqliteHistory::new(self.pool.clone())), + ))); + } + // `execute_subtask` only while the child can still recurse. + let delegate = self.delegate.read().unwrap().clone(); + if let Some(d) = delegate { + native.push(Arc::new(d.as_ref().clone().with_name(tn::EXECUTE_SUBTASK))); + } + native.push(Arc::new(ActivateToolsTool::new(Arc::new(SkaldToolActivator::new( + self.pool.clone(), + self.mcp.clone(), + child_grants.clone(), + self.session_id, + Some(child_frame.get()), + ))))); + + let toolset: Arc = Arc::new( + SkaldToolSet::new( + child_defs, + self.config_defs.clone(), + self.mcp.clone(), + child_grants, + self.memory_tools.clone(), + self.image_tools.clone(), + Vec::new(), + self.core_tools.clone(), + ) + .with_native_all(native), + ); + + let assembler: Arc = Arc::new(SkaldAssembler { + pool: self.pool.clone(), + scratchpad_sid: self.session_id, + datetime_config: self.datetime_config.clone(), + max_history_messages: self.max_history_messages, + max_tool_result_chars: self.max_tool_result_chars, + compactor_enabled: self.compactor_enabled, + fs: self.fs.clone(), + activation: Some(crate::loop_adapters::activation::SkaldActivationSource::new( + self.pool.clone(), + self.mcp.clone(), + self.config_defs.clone(), + self.session_id, + Some(child_frame.get()), + )), + }); + + Ok(AgentProfile { + id: id.to_string(), + kind: AgentKind::Task, + context, + tools: ToolSelection::inherit(), + model, + selector: Some(selector), + assembler: Some(assembler), + toolset: Some(toolset), + }) + } + + async fn list(&self, kind: AgentKind) -> Vec { + if kind != AgentKind::Task { + return Vec::new(); + } + crate::agents::discover() + .unwrap_or_default() + .into_iter() + .filter(|a| matches!(a.agent_type, crate::agents::AgentType::Task)) + .map(|a| AgentSummary { id: a.id, kind, description: a.description }) + .collect() + } + + async fn on_child_closed(&self, frame: FrameId) { + // Stack-scoped activations are ephemeral — deleted on frame exit. + if let Err(e) = crate::db::activated_tools::delete_for_stack(&self.pool, frame.get()).await { + tracing::warn!(frame = %frame, error = %e, "catalog: failed to delete stack activations"); + } + } +} diff --git a/crates/skald-core/src/loop_adapters/gate.rs b/crates/skald-core/src/loop_adapters/gate.rs index 11f2cf1..ce42ced 100644 --- a/crates/skald-core/src/loop_adapters/gate.rs +++ b/crates/skald-core/src/loop_adapters/gate.rs @@ -12,16 +12,19 @@ use std::collections::HashSet; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Arc, Mutex, RwLock}; +use std::sync::{Arc, Mutex}; use agent_loop::events::{EventSink, LoopEvent}; use agent_loop::gate::{Gate, GateDecision, PendingCall}; use agent_loop::store::{CallState, HistoryStore}; +use core_api::user_fs::SharedFs; +use sqlx::SqlitePool; +use tokio::sync::RwLock; use crate::approval::{ApprovalManager, GateResult}; use crate::run_context::RunContext; use crate::session::handler::ApprovalDecision; -use crate::tools::{ToolRegistry, is_file_read_tool, is_file_write_tool}; +use crate::tools::{ToolRegistry, is_file_read_tool, is_file_write_tool, tool_names as tn}; /// Everything the gate needs that the current loop keeps on the handler. /// Shared by reference so phase-2 wiring shares the same cells. @@ -35,7 +38,12 @@ pub struct ApprovalGate { run_context: Arc>>, pre_approved: Arc>>, auto_deny: Arc, - context_label: Arc>>, + context_label: Arc>>, + /// For the `PendingWrite` diff: owner pool (user-memory), shared pool + /// (shared-memory), and the caller's fs view (host paths). + pool: Arc, + shared_pool: Arc, + fs: Option, } impl ApprovalGate { @@ -50,7 +58,10 @@ impl ApprovalGate { run_context: Arc>>, pre_approved: Arc>>, auto_deny: Arc, - context_label: Arc>>, + context_label: Arc>>, + pool: Arc, + shared_pool: Arc, + fs: Option, ) -> Self { Self { approval, @@ -63,8 +74,130 @@ impl ApprovalGate { pre_approved, auto_deny, context_label, + pool, + shared_pool, + fs, } } + + /// Reads the current content of a file for the `PendingWrite` diff, routed + /// exactly like the fs-tools (memory notes → the right pool, everything + /// else → the caller's host workspace, containment-checked). + async fn read_current_content(&self, path: &str) -> Option { + use crate::tools::fs::{MemScope, classify_memory, resolve_host_path}; + if let Some(m) = classify_memory(path) { + let pool = match m.scope { + MemScope::User => &self.pool, + MemScope::Shared => &self.shared_pool, + }; + return crate::db::memory_docs::get(pool, &m.rel) + .await.ok().flatten().map(|d| d.content); + } + let fs = self.fs.as_ref()?; + let abs = resolve_host_path(&fs.load(), path).ok()?; + tokio::fs::read_to_string(&abs).await.ok() + } + + /// Computes what a file would look like after the tool runs, without + /// writing it. `None` if indeterminable (e.g. edit on a missing file). + async fn compute_new_content(&self, name: &str, args: &serde_json::Value) -> Option { + match name { + "write_file" => args["content"].as_str().map(|s| s.to_string()), + "edit_file" => { + let path = args["path"].as_str()?; + let old_text = args["old"].as_str()?; + let new_text = args["new"].as_str()?; + let current = self.read_current_content(path).await?; + if current.contains(old_text) { + Some(current.replacen(old_text, new_text, 1)) + } else { + None + } + } + "insert_at_line" => { + let path = args["path"].as_str()?; + let line_num = args["line"].as_u64()? as usize; + let new_text = args["content"].as_str()?; + let placement = args["placement"].as_str().unwrap_or("after"); + if line_num == 0 { return None; } + let current = self.read_current_content(path).await?; + let mut lines: Vec<&str> = current.split('\n').collect(); + let idx = (line_num - 1).min(lines.len().saturating_sub(1)); + let insert_idx = if placement == "before" { idx } else { idx + 1 }; + let new_lines: Vec<&str> = new_text.split('\n').collect(); + for (i, l) in new_lines.iter().enumerate() { + lines.insert(insert_idx + i, l); + } + Some(lines.join("\n")) + } + "replace_lines" => { + let path = args["path"].as_str()?; + let from_line = args["from_line"].as_u64()? as usize; + let to_line = args["to_line"].as_u64()? as usize; + let new_text = args["new"].as_str()?; + if from_line == 0 || to_line < from_line { return None; } + let current = self.read_current_content(path).await?; + let mut lines: Vec<&str> = current.lines().collect(); + let total = lines.len(); + if from_line > total { return None; } + let to_clamped = to_line.min(total); + let new_lines: Vec<&str> = new_text.lines().collect(); + lines.splice((from_line - 1)..to_clamped, new_lines); + let has_trailing = current.ends_with('\n'); + let mut result = lines.join("\n"); + if has_trailing { result.push('\n'); } + Some(result) + } + _ => None, + } + } + + /// Emits the approval event for the tool kind: `PendingWrite` (via + /// `LoopEvent::Host`) for file-write tools and `execute_cmd`, + /// `ApprovalRequired` otherwise (port of `emit_approval_event`). + async fn emit_approval_event( + &self, + events: &EventSink, + call: &PendingCall, + request_id: i64, + ) { + let name = call.name.as_str(); + if is_file_write_tool(name) { + let path = call.args["path"].as_str().unwrap_or("").to_string(); + let (old_content, new_content) = tokio::join!( + self.read_current_content(&path), + self.compute_new_content(name, &call.args), + ); + if let Some(new_content) = new_content { + events.emit(call.frame, call.parent_frame, LoopEvent::Host(serde_json::json!({ + "type": "pending_write", + "request_id": request_id, + "tool_call_id": call.id.get(), + "path": path, + "old_content": old_content, + "new_content": new_content, + }))); + return; + } + } else if name == tn::EXECUTE_CMD { + let cmd = call.args["command"].as_str().unwrap_or(""); + events.emit(call.frame, call.parent_frame, LoopEvent::Host(serde_json::json!({ + "type": "pending_write", + "request_id": request_id, + "tool_call_id": call.id.get(), + "path": "$ execute_cmd", + "old_content": serde_json::Value::Null, + "new_content": format!("$ {cmd}"), + }))); + return; + } + events.emit(call.frame, call.parent_frame, LoopEvent::ApprovalRequired { + id: call.id, + name: call.name.clone(), + args: call.args.clone(), + request_id, + }); + } } #[agent_loop::async_trait] @@ -95,7 +228,7 @@ impl Gate for ApprovalGate { // (never overrides a Deny). if matches!(gate, GateResult::Require) { let path = call.args["path"].as_str().unwrap_or(""); - let guard = self.run_context.read().map(|g| g.clone()).unwrap_or_default(); + let guard = self.run_context.read().await.clone(); let dflt = RunContext::default(); let rc = guard.as_ref().unwrap_or(&dflt); let pre_allowed = if is_file_read_tool(&call.name) { @@ -144,12 +277,7 @@ impl Gate for ApprovalGate { category, ) .await; - events.emit(call.frame, None, LoopEvent::ApprovalRequired { - id: call.id, - name: call.name.clone(), - args: call.args.clone(), - }); - let _ = request_id; + self.emit_approval_event(events, call, request_id).await; match approve_rx.await { Ok(ApprovalDecision::Approved) => GateDecision::Allow, @@ -225,10 +353,13 @@ mod tests { 1, "web", None, - Arc::new(RwLock::new(None)), + Arc::new(tokio::sync::RwLock::new(None)), Arc::new(Mutex::new(HashSet::new())), Arc::new(AtomicBool::new(false)), - Arc::new(RwLock::new(None)), + Arc::new(std::sync::RwLock::new(None)), + pool.clone(), + pool.clone(), + None, ); let (bus, _) = tokio::sync::broadcast::channel(16); let events = EventSink::new(ConversationId::new("session:1"), bus); @@ -237,6 +368,7 @@ mod tests { name: "some_tool".into(), args: json!({}), frame: FrameId(frame.id), + parent_frame: None, agent: "assistant".into(), extensions: Extensions::new(), }; @@ -287,10 +419,13 @@ mod tests { 1, "cron", // background source: auto-deny None, - Arc::new(RwLock::new(None)), + Arc::new(tokio::sync::RwLock::new(None)), Arc::new(Mutex::new(HashSet::new())), Arc::new(AtomicBool::new(true)), - Arc::new(RwLock::new(None)), + Arc::new(std::sync::RwLock::new(None)), + pool.clone(), + pool.clone(), + None, ); let (bus, _) = tokio::sync::broadcast::channel(16); let events = EventSink::new(ConversationId::new("session:1"), bus); @@ -299,6 +434,7 @@ mod tests { name: "some_tool".into(), args: json!({}), frame: FrameId(frame.id), + parent_frame: None, agent: "assistant".into(), extensions: Extensions::new(), }; diff --git a/crates/skald-core/src/loop_adapters/history.rs b/crates/skald-core/src/loop_adapters/history.rs index 9ab9d22..e7d16b6 100644 --- a/crates/skald-core/src/loop_adapters/history.rs +++ b/crates/skald-core/src/loop_adapters/history.rs @@ -184,6 +184,30 @@ impl HistoryStore for SqliteHistory { Ok(()) } + async fn get_frame(&self, frame: FrameId) -> agent_loop::Result> { + let row = sqlx::query_as::<_, (i64, i64, String, Option, i64, Option, Option)>( + "SELECT id, session_id, agent_id, agent_prompt, depth, parent_tool_call_id, terminated_at + FROM chat_sessions_stack + WHERE id = ?", + ) + .bind(frame.get()) + .fetch_optional(&*self.pool) + .await?; + Ok(row.map(|(id, sid, agent, prompt, depth, parent_call, terminated)| FrameRecord { + id: FrameId(id), + conversation: ConversationId::new(format!("session:{sid}")), + parent: None, + spec: FrameSpec { + agent, + prompt, + depth: depth as u32, + parent_call: parent_call.map(ToolCallId), + meta: Value::Null, + }, + active: terminated.is_none(), + })) + } + async fn active_frames(&self, conv: &ConversationId) -> agent_loop::Result> { let session_id = Self::session_id(conv)?; let rows = sqlx::query_as::<_, (i64, i64, String, Option, i64, Option)>( @@ -319,6 +343,24 @@ impl HistoryStore for SqliteHistory { Ok(()) } + async fn get_call(&self, id: ToolCallId) -> agent_loop::Result> { + Ok(chat_llm_tools::get(&self.pool, id.get()).await?.map(Self::stored_call)) + } + + async fn set_call_extras(&self, id: ToolCallId, extras: Value) -> agent_loop::Result<()> { + // Map the known extras onto the dedicated columns (preview, media); + // unknown keys are dropped (the table has no generic blob). + if extras.get("preview_old").is_some() || extras.get("preview_new").is_some() { + let old = extras["preview_old"].as_str(); + let new = extras["preview_new"].as_str(); + chat_llm_tools::set_preview(&self.pool, id.get(), old, new).await?; + } + if let Some(media) = extras["media"].as_str() { + chat_llm_tools::set_media(&self.pool, id.get(), media).await?; + } + Ok(()) + } + async fn calls_in_state(&self, frame: FrameId, states: &[CallState]) -> agent_loop::Result> { // All calls of the frame, filtered in Rust: a frame's call set is // bounded, and a static query keeps sqlx's dynamic-SQL audit happy. diff --git a/crates/skald-core/src/loop_adapters/hooks.rs b/crates/skald-core/src/loop_adapters/hooks.rs new file mode 100644 index 0000000..8ae6447 --- /dev/null +++ b/crates/skald-core/src/loop_adapters/hooks.rs @@ -0,0 +1,61 @@ +//! Skald's `LoopHooks`: the file-write diff preview bracket (pre: capture the +//! old content; post: capture the new one and persist via `set_call_extras`). +//! Port of the `execute_tool_call` preview bracketing (blueprint §10). + +use std::collections::HashMap; +use std::sync::Mutex; + +use agent_loop::events::PendingToolCall; +use agent_loop::hooks::{HookCtx, LoopHooks}; +use agent_loop::store::CallOutcome; +use serde_json::json; + +use crate::loop_adapters::preview::{PreviewContext, cap_preview, read_current_content}; +use crate::tools::is_file_write_tool; + +/// Captures before/after snapshots around file-write tools so the diff +/// renders inline and survives a reload. +pub struct SkaldWritePreviewHook { + ctx: PreviewContext, + /// old-content captured in `pre_tool_call`, consumed in `post_tool_call`. + pending: Mutex>>, +} + +impl SkaldWritePreviewHook { + pub fn new(ctx: PreviewContext) -> Self { + Self { ctx, pending: Mutex::new(HashMap::new()) } + } +} + +#[agent_loop::async_trait] +impl LoopHooks for SkaldWritePreviewHook { + async fn pre_tool_call(&self, call: &mut PendingToolCall, _ctx: &HookCtx) -> agent_loop::hooks::HookVerdict { + if is_file_write_tool(&call.name) + && let Some(path) = call.arguments["path"].as_str() + { + let old = cap_preview(read_current_content(&self.ctx, path).await); + self.pending.lock().unwrap().insert(call.id.get(), old); + } + agent_loop::hooks::HookVerdict::Allow + } + + async fn post_tool_call(&self, call: &PendingToolCall, outcome: &CallOutcome, ctx: &HookCtx) { + let Some(old) = self.pending.lock().unwrap().remove(&call.id.get()) else { + return; + }; + let Some(path) = call.arguments["path"].as_str() else { + return; + }; + // `new` is captured only on success — a failed/cancelled write shows + // no diff (the file may not exist in its intended form). + let new = if matches!(outcome, CallOutcome::Completed(_)) { + cap_preview(read_current_content(&self.ctx, path).await) + } else { + None + }; + let _ = ctx + .store + .set_call_extras(call.id, json!({ "preview_old": old, "preview_new": new })) + .await; + } +} diff --git a/crates/skald-core/src/loop_adapters/live_input.rs b/crates/skald-core/src/loop_adapters/live_input.rs new file mode 100644 index 0000000..ce8c1fa --- /dev/null +++ b/crates/skald-core/src/loop_adapters/live_input.rs @@ -0,0 +1,38 @@ +//! `PendingUserInput` → the crate's `LiveInput` (D10 pull-based live input). + +use std::sync::Arc; + +use agent_loop::manager::LiveInput; +use agent_loop::store::NewMessage; + +use crate::session::handler::PendingUserInput; + +/// Drains the source's inbox into the running turn: one `NewMessage` per +/// queued user message, attachments/command metadata preserved. +pub struct PendingLiveInput { + inner: Arc, +} + +impl PendingLiveInput { + pub fn new(inner: Arc) -> Self { Self { inner } } +} + +#[agent_loop::async_trait] +impl LiveInput for PendingLiveInput { + async fn drain(&self) -> Vec { + self.inner + .drain_user() + .await + .into_iter() + .map(|m| { + let mut msg = NewMessage::user(m.content); + if let Some(meta) = m.metadata + && let Ok(v) = serde_json::to_value(meta) + { + msg.metadata = Some(v); + } + msg + }) + .collect() + } +} diff --git a/crates/skald-core/src/loop_adapters/mod.rs b/crates/skald-core/src/loop_adapters/mod.rs index 2d39df6..2364218 100644 --- a/crates/skald-core/src/loop_adapters/mod.rs +++ b/crates/skald-core/src/loop_adapters/mod.rs @@ -16,7 +16,16 @@ //! `activated_tools` table and the MCP provider (D15). pub mod activation; +pub mod assembler; +pub mod builtins; +pub mod catalog; pub mod gate; pub mod history; +pub mod hooks; +pub mod live_input; +pub mod preview; pub mod selector; +pub mod system; pub mod toolset; +pub mod translate; +pub mod wiring; diff --git a/crates/skald-core/src/loop_adapters/preview.rs b/crates/skald-core/src/loop_adapters/preview.rs new file mode 100644 index 0000000..6535ae7 --- /dev/null +++ b/crates/skald-core/src/loop_adapters/preview.rs @@ -0,0 +1,100 @@ +//! File-write diff preview, shared by the approval gate (pre-approval diff) +//! and the write-preview hook (executed-write diff). Routes memory-vs-disk +//! exactly like the fs-tools. + +use std::sync::Arc; + +use core_api::user_fs::SharedFs; +use sqlx::SqlitePool; + +use crate::tools::fs::{MemScope, classify_memory, resolve_host_path}; + +/// Max bytes captured per side of a file-write diff preview. Beyond this the +/// side is dropped (`None`) so a huge file never bloats a row or a WS payload. +pub const MAX_PREVIEW_BYTES: usize = 256 * 1024; + +/// Drops a captured snapshot over the size cap (a truncated snapshot would +/// render a misleading diff). +pub fn cap_preview(s: Option) -> Option { + s.filter(|c| c.len() <= MAX_PREVIEW_BYTES) +} + +/// The pieces a preview read needs: owner pool (user-memory), shared pool +/// (shared-memory), and the caller's fs view (host paths). +#[derive(Clone)] +pub struct PreviewContext { + pub pool: Arc, + pub shared_pool: Arc, + pub fs: Option, +} + +/// Reads the current content of a file for a diff, routed exactly like the +/// fs-tools. A resolve failure or a missing note/file yields `None` +/// (rendered as "new file"). +pub async fn read_current_content(ctx: &PreviewContext, path: &str) -> Option { + if let Some(m) = classify_memory(path) { + let pool = match m.scope { + MemScope::User => &ctx.pool, + MemScope::Shared => &ctx.shared_pool, + }; + return crate::db::memory_docs::get(pool, &m.rel) + .await.ok().flatten().map(|d| d.content); + } + let fs = ctx.fs.as_ref()?; + let abs = resolve_host_path(&fs.load(), path).ok()?; + tokio::fs::read_to_string(&abs).await.ok() +} + +/// Computes what a file would look like after the tool runs, without writing +/// it. `None` if indeterminable (e.g. edit on a missing file). +pub async fn compute_new_content(ctx: &PreviewContext, name: &str, args: &serde_json::Value) -> Option { + match name { + "write_file" => args["content"].as_str().map(|s| s.to_string()), + "edit_file" => { + let path = args["path"].as_str()?; + let old_text = args["old"].as_str()?; + let new_text = args["new"].as_str()?; + let current = read_current_content(ctx, path).await?; + if current.contains(old_text) { + Some(current.replacen(old_text, new_text, 1)) + } else { + None + } + } + "insert_at_line" => { + let path = args["path"].as_str()?; + let line_num = args["line"].as_u64()? as usize; + let new_text = args["content"].as_str()?; + let placement = args["placement"].as_str().unwrap_or("after"); + if line_num == 0 { return None; } + let current = read_current_content(ctx, path).await?; + let mut lines: Vec<&str> = current.split('\n').collect(); + let idx = (line_num - 1).min(lines.len().saturating_sub(1)); + let insert_idx = if placement == "before" { idx } else { idx + 1 }; + let new_lines: Vec<&str> = new_text.split('\n').collect(); + for (i, l) in new_lines.iter().enumerate() { + lines.insert(insert_idx + i, l); + } + Some(lines.join("\n")) + } + "replace_lines" => { + let path = args["path"].as_str()?; + let from_line = args["from_line"].as_u64()? as usize; + let to_line = args["to_line"].as_u64()? as usize; + let new_text = args["new"].as_str()?; + if from_line == 0 || to_line < from_line { return None; } + let current = read_current_content(ctx, path).await?; + let mut lines: Vec<&str> = current.lines().collect(); + let total = lines.len(); + if from_line > total { return None; } + let to_clamped = to_line.min(total); + let new_lines: Vec<&str> = new_text.lines().collect(); + lines.splice((from_line - 1)..to_clamped, new_lines); + let has_trailing = current.ends_with('\n'); + let mut result = lines.join("\n"); + if has_trailing { result.push('\n'); } + Some(result) + } + _ => None, + } +} diff --git a/crates/skald-core/src/loop_adapters/system.rs b/crates/skald-core/src/loop_adapters/system.rs new file mode 100644 index 0000000..72113a8 --- /dev/null +++ b/crates/skald-core/src/loop_adapters/system.rs @@ -0,0 +1,186 @@ +//! `AgentSystemContext` — Skald's agent prompt as a `SystemContextSource` +//! (the static half of the old `MessageBuilder::build`, blueprint §10): +//! AGENT.md + `inject_memory` files + skills index + `extra_system` + +//! `__MCP_LIST__` / `__SHARED_FOLDERS__` / `__USER_PROFILE__` / custom +//! substitutions. The dynamic tail (Honcho memory, per-turn overrides) rides +//! as `dynamic_tail`; the datetime line and scratchpad stay assembler-side. + +use std::collections::HashMap; +use std::sync::Arc; + +use agent_loop::context::{SystemContext, SystemContextSource, TurnInfo}; +use sqlx::SqlitePool; + +use crate::mcp::McpProvider; + +/// Registry of installed skills, relative to Skald's process cwd. Injected +/// into agents that have `inject_skills` enabled (the default). +const SKILLS_INDEX_PATH: &str = "skills/index.md"; + +/// The static system content of one agent, resolved per turn. +pub struct AgentSystemContext { + pub agent_id: String, + /// Static extra context (interface formatting rules, e.g. Telegram HTML). + pub extra_static: Option, + /// Dynamic extra context (Honcho memory merged with per-turn overrides), + /// emitted as the dynamic tail. + pub extra_dynamic: Option, + pub tail_reminder: Option, + pub substitutions: HashMap, + /// Owner pool (`user-memory/` notes). + pub pool: Arc, + /// Shared pool (`shared-memory/`, shared folders, user profile). + pub shared_pool: Arc, + pub user_id: String, + pub mcp: Arc, + /// Project root for `__PROJECT_ROOT__` expansion in `inject_memory`. + pub project_root: Option, +} + +#[agent_loop::async_trait] +impl SystemContextSource for AgentSystemContext { + async fn system_context(&self, _turn: &TurnInfo) -> agent_loop::Result { + let mut static_content = crate::agents::load_prompt(&self.agent_id)?; + + let meta = crate::agents::load_meta(&self.agent_id)?; + if !meta.inject_memory.is_empty() { + static_content.push_str( + "\n\n---\nThe following memory files have been loaded automatically. \ + You can edit them with `edit_file` or `write_file` using the path shown.\n" + ); + for mem_path in &meta.inject_memory { + let (content, display) = self.load_inject_memory(mem_path).await; + match content { + Some(c) => static_content.push_str(&format!( + "\n\n{c}\n\n" + )), + None => static_content.push_str(&format!( + "\n\n(file not created yet)\n\n" + )), + } + } + } + + // Skills index — injected unless the agent opts out. Skipped silently + // when no skills are installed. + if meta.inject_skills { + let (abs, display) = self.resolve_memory_path(SKILLS_INDEX_PATH); + if let Ok(c) = tokio::fs::read_to_string(&abs).await { + static_content.push_str(&format!( + "\n\n---\nInstalled skills you can use (read the linked `SKILL.md` before running a skill):\n\ + \n\n{c}\n\n" + )); + } + } + + if let Some(extra) = &self.extra_static { + static_content.push_str("\n\n---\n"); + static_content.push_str(extra); + } + + if static_content.contains("__MCP_LIST__") { + static_content = static_content.replace("__MCP_LIST__", &self.render_mcp_list()); + } + if static_content.contains("__SHARED_FOLDERS__") { + static_content = static_content.replace( + "__SHARED_FOLDERS__", + &crate::session::handler::message_builder::render_shared_folders_section( + &self.shared_pool, + &self.user_id, + ) + .await?, + ); + } + if static_content.contains("__USER_PROFILE__") { + static_content = static_content.replace( + "__USER_PROFILE__", + &crate::session::handler::message_builder::render_user_profile_section( + &self.shared_pool, + &self.user_id, + ) + .await?, + ); + } + + for (key, value) in &self.substitutions { + let sentinel = format!("__{key}__"); + if static_content.contains(sentinel.as_str()) { + static_content = static_content.replace(sentinel.as_str(), value); + } + } + + Ok(SystemContext { + base: static_content, + extra_static: Vec::new(), + dynamic_tail: self.extra_dynamic.clone().into_iter().collect(), + tail_reminder: self.tail_reminder.clone(), + }) + } +} + +impl AgentSystemContext { + /// Loads an `inject_memory` entry, returning `(content, display_path)`. + /// Virtual memory paths read from SQLite; everything else is a disk read. + async fn load_inject_memory(&self, mem_path: &str) -> (Option, String) { + use crate::tools::fs::{MemScope, classify_memory}; + if let Some(m) = classify_memory(mem_path) { + let pool = match m.scope { + MemScope::User => &self.pool, + MemScope::Shared => &self.shared_pool, + }; + let content = crate::db::memory_docs::get(pool, &m.rel) + .await.ok().flatten().map(|d| d.content); + return (content, mem_path.to_string()); + } + let (abs, display) = self.resolve_memory_path(mem_path); + (tokio::fs::read_to_string(&abs).await.ok(), display) + } + + fn resolve_memory_path(&self, mem_path: &str) -> (std::path::PathBuf, String) { + let display = if mem_path.contains("__PROJECT_ROOT__") { + match &self.project_root { + Some(root) => mem_path.replace("__PROJECT_ROOT__", root), + None => { + tracing::warn!( + mem_path, + "inject_memory entry references __PROJECT_ROOT__ but this session has no project root; skipping" + ); + return (std::path::PathBuf::from(mem_path), mem_path.to_string()); + } + } + } else { + mem_path.to_string() + }; + let abs = crate::tools::fs::resolve(&display) + .unwrap_or_else(|_| std::path::PathBuf::from(&display)); + (abs, display) + } + + /// The **static** catalogue of loadable MCP servers (identical regardless + /// of which are active — cache-prefix stability). + fn render_mcp_list(&self) -> String { + let all_servers: std::collections::BTreeSet = self.mcp.tools() + .into_iter() + .map(|t| t.server_name) + .collect(); + + if all_servers.is_empty() { + return String::new(); + } + + let descriptions = self.mcp.server_descriptions(); + + let mut out = String::from( + "## MCP servers\n\nConnectors you can load with `activate_tools([\"name\"])`. \ + Once loaded, a server's tools are callable as `mcp____`:\n\n", + ); + out.push_str("| Server | Description |\n|--------|-------------|\n"); + for name in &all_servers { + let desc = descriptions.get(name) + .and_then(|d| d.as_deref()) + .unwrap_or("—"); + out.push_str(&format!("| `{name}` | {desc} |\n")); + } + out + } +} diff --git a/crates/skald-core/src/loop_adapters/toolset.rs b/crates/skald-core/src/loop_adapters/toolset.rs index 4b72e06..330fbc7 100644 --- a/crates/skald-core/src/loop_adapters/toolset.rs +++ b/crates/skald-core/src/loop_adapters/toolset.rs @@ -201,7 +201,7 @@ impl LoopTool for McpToolBridge { /// activated at round N are visible at round N+1 for free. pub struct SkaldToolSet { base_defs: Vec, - config_defs: Vec, + config_defs: Arc>, mcp: Arc, grants: Arc>>, memory_tools: Vec>, @@ -212,13 +212,15 @@ pub struct SkaldToolSet { core_tools: Vec>, /// Extra crate-native tools for find() (bridge-free). native_tools: Vec>, + /// Records tools offered to the LLM each round (Security-groups UI). + discovery: Option>, } impl SkaldToolSet { #[allow(clippy::too_many_arguments)] pub fn new( base_defs: Vec, - config_defs: Vec, + config_defs: Arc>, mcp: Arc, grants: Arc>>, memory_tools: Vec>, @@ -236,13 +238,24 @@ impl SkaldToolSet { interface_tools, core_tools, native_tools: Vec::new(), + discovery: None, } } + pub fn with_discovery(mut self, discovery: Arc) -> Self { + self.discovery = Some(discovery); + self + } + pub fn with_native(mut self, tool: Arc) -> Self { self.native_tools.push(tool); self } + + pub fn with_native_all(mut self, tools: Vec>) -> Self { + self.native_tools.extend(tools); + self + } } /// Tags an OpenAI tool definition as deferred (Anthropic tool search). @@ -285,6 +298,15 @@ impl ToolSet for SkaldToolSet { defs.extend(self.image_tools.iter().map(|t| t.openai_definition())); defs.extend(self.interface_tools.iter().map(|t| t.definition.clone())); defs.extend(self.native_tools.iter().map(|t| t.definition())); + // Dedup by name (first wins): the host's base/interface defs already + // carry the built-ins (scratchpad/todos/ask_user/activate_tools), and + // the native aliases provide the same names for find() — the wire must + // never carry duplicates (OpenAI-compat APIs 400 on them). + let mut seen = std::collections::HashSet::new(); + defs.retain(|d| seen.insert(d["function"]["name"].as_str().unwrap_or("").to_string())); + if let Some(discovery) = &self.discovery { + discovery.observe(&defs); + } defs } @@ -365,7 +387,7 @@ mod tests { fn toolset(grants: Arc>>) -> SkaldToolSet { SkaldToolSet::new( vec![serde_json::json!({"type":"function","function":{"name":"read_file","parameters":{}}})], - vec![serde_json::json!({"type":"function","function":{"name":"cron_list","parameters":{}}})], + Arc::new(vec![serde_json::json!({"type":"function","function":{"name":"cron_list","parameters":{}}})]), fake_mcp("gmail", &["send"]), grants, vec![], diff --git a/crates/skald-core/src/loop_adapters/translate.rs b/crates/skald-core/src/loop_adapters/translate.rs new file mode 100644 index 0000000..3a2375d --- /dev/null +++ b/crates/skald-core/src/loop_adapters/translate.rs @@ -0,0 +1,307 @@ +//! The `LoopEvent → ServerEvent` translator (blueprint §10): ONE subscriber of +//! the loop manager's bus, forwarding to the session's WS channel with the +//! host enrichments the frontend expects (display meta, diff previews, file +//! changes). Byte-parity with the old `TurnEmitter` sequence is the contract. + +use std::sync::Arc; + +use agent_loop::events::{DeltaKind, Event, LoopEvent}; +use agent_loop::store::{CallOutcome, HistoryStore}; +use core_api::message_meta::MessageMetadata; +use serde_json::Value; +use tokio::sync::mpsc; + +use crate::events::{ServerEvent, TokenDeltaKind}; +use crate::mcp::McpProvider; +use crate::tools::{ToolRegistry, is_file_write_tool}; + +/// Forwards one conversation's loop events to the session's WS `tx`. +pub struct EventTranslator { + tx: mpsc::Sender, + tools: Arc, + mcp: Arc, + store: Arc, + shared: Arc>, +} + +/// Turn state the wiring reads back after join (ChatEvent publication). +#[derive(Default)] +pub struct TranslateShared { + /// The user message id that opened the turn. + pub user_message_id: Option, + /// Accumulated tool calls of the turn (done/failed only — mirrors the old + /// `all_tool_calls` accumulate rules). + pub tool_calls: Vec, +} + +impl EventTranslator { + pub fn new( + tx: mpsc::Sender, + tools: Arc, + mcp: Arc, + store: Arc, + ) -> (Self, Arc>) { + let shared = Arc::new(std::sync::Mutex::new(TranslateShared::default())); + (Self { tx, tools, mcp, store, shared: shared.clone() }, shared) + } + + /// Subscribe and forward until `stop` is cancelled (the turn's end). + pub fn spawn(self, mut rx: tokio::sync::broadcast::Receiver>, stop: tokio_util::sync::CancellationToken) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + loop { + tokio::select! { + _ = stop.cancelled() => break, + ev = rx.recv() => { + match ev { + Ok(ev) => self.forward(ev).await, + Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => { + tracing::warn!(skipped = n, "event translator lagged; some events were dropped"); + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + } + } + } + } + }) + } + + async fn emit(&self, ev: ServerEvent) { + self.tx.send(ev).await.ok(); + } + + pub async fn forward(&self, ev: Event) { + let is_root = ev.parent_frame.is_none(); + match ev.inner { + LoopEvent::TurnStarted | LoopEvent::RoundStarted { .. } | LoopEvent::AsyncResultReady { .. } => {} + + LoopEvent::UserMessage { message_id, content, synthetic, metadata } => { + // The turn-opening user message (root, non-synthetic) is + // recorded for the wiring's ChatEvent publication. + if is_root && !synthetic { + let mut g = self.shared.lock().unwrap(); + if g.user_message_id.is_none() { + g.user_message_id = Some(message_id.get()); + } + } + if synthetic { + return; + } + let meta: Option = metadata + .as_ref() + .and_then(|v| serde_json::from_value(v.clone()).ok()); + let attachments = meta.as_ref().map(|m| m.attachments.clone()).unwrap_or_default(); + // A custom slash command persists its expanded template (for + // LLM replay) but the bubble shows the typed command. + let echo = meta + .and_then(|m| m.command.map(|c| c.display)) + .unwrap_or(content); + self.emit(ServerEvent::UserMessage { message_id: message_id.get(), content: echo, attachments }).await; + } + + LoopEvent::TokenDelta { kind, text } => { + let kind = match kind { + DeltaKind::Content => TokenDeltaKind::Content, + DeltaKind::Reasoning => TokenDeltaKind::Reasoning, + }; + self.emit(ServerEvent::TokenDelta { kind, delta: text }).await; + } + + LoopEvent::Thinking { message_id, content, usage, reasoning } => { + self.emit(ServerEvent::Thinking { + message_id: message_id.get(), + content, + input_tokens: usage.input_tokens, + output_tokens: usage.output_tokens, + reasoning_content: reasoning, + }).await; + } + + LoopEvent::Done { message_id, content, usage, reasoning } => { + if !is_root { + return; // a child's completion rides AgentFinished + } + self.emit(ServerEvent::Done { + message_id: message_id.get(), + stack_id: ev.frame.get(), + content, + input_tokens: usage.input_tokens, + output_tokens: usage.output_tokens, + reasoning_content: reasoning, + }).await; + } + + LoopEvent::Truncated { output_tokens } => { + if is_root { + self.emit(ServerEvent::Truncated { output_tokens }).await; + } + } + + LoopEvent::ToolCallStarted { id, message_id, name, args } => { + let (display_name, icon) = self.ui_meta(&name, &args); + let label_short = self.tools.describe_call(&name, &args, core_api::tool::ToolDescriptionLength::Short); + let label_full = self.tools.describe_call(&name, &args, core_api::tool::ToolDescriptionLength::Full); + let path = self.tools.target_path(&name, &args); + self.emit(ServerEvent::ToolStart { + tool_call_id: id.get(), + message_id: message_id.get(), + name, + arguments: args, + display_name, + icon, + label_short, + label_full, + path, + }).await; + } + + LoopEvent::ToolCallFinished { id, outcome } => match outcome { + CallOutcome::Completed(out) => { + let stored = self.store.get_call(id).await.ok().flatten(); + if let Some(c) = stored.as_ref() { + self.shared.lock().unwrap().tool_calls.push(core_api::bus::ToolCallEvent { + name: c.name.clone(), + arguments: Some(serde_json::to_string(&c.arguments).unwrap_or_default()), + result: Some(out.to_wire()), + status: "done".to_string(), + }); + } + let (preview_old, preview_new) = stored + .as_ref() + .map(|c| ( + c.extras["preview_old"].as_str().map(str::to_string), + c.extras["preview_new"].as_str().map(str::to_string), + )) + .unwrap_or((None, None)); + self.emit(ServerEvent::ToolDone { + tool_call_id: id.get(), + result: out.to_wire(), + result_type: out.kind().to_string(), + preview_old, + preview_new, + }).await; + // A successful file-write asks clients holding the file to reload. + if let Some(c) = stored + && is_file_write_tool(&c.name) + && let Some(p) = c.arguments["path"].as_str() + { + self.emit(ServerEvent::FileChanged { path: crate::approval::normalize_path(p) }).await; + } + } + CallOutcome::Failed(error) => { + let stored = self.store.get_call(id).await.ok().flatten(); + if let Some(c) = stored.as_ref() { + self.shared.lock().unwrap().tool_calls.push(core_api::bus::ToolCallEvent { + name: c.name.clone(), + arguments: Some(serde_json::to_string(&c.arguments).unwrap_or_default()), + result: Some(error.clone()), + status: "failed".to_string(), + }); + } + self.emit(ServerEvent::ToolError { tool_call_id: id.get(), error }).await; + } + CallOutcome::Cancelled => { + self.emit(ServerEvent::ToolCancelled { tool_call_id: id.get() }).await; + } + CallOutcome::Rejected { reason } => { + self.emit(ServerEvent::ToolRejected { tool_call_id: id.get(), reason }).await; + } + }, + + LoopEvent::ApprovalRequired { id, name, args, request_id } => { + self.emit(ServerEvent::ApprovalRequired { + request_id, + tool_call_id: id.get(), + tool_name: name, + arguments: args, + }).await; + } + + LoopEvent::AgentSpawned { frame, agent, depth, prompt_preview, parent_call, parent_agent } => { + self.emit(ServerEvent::AgentStart { + stack_id: frame.get(), + parent_tool_call_id: parent_call.get(), + agent_id: agent, + parent_agent_id: parent_agent, + depth: depth as i64, + prompt_preview, + }).await; + } + + LoopEvent::AgentFinished { frame, agent, result_preview, parent_agent } => { + self.emit(ServerEvent::AgentDone { + stack_id: frame.get(), + agent_id: agent, + parent_agent_id: parent_agent, + result_preview, + }).await; + } + + LoopEvent::ModelFallback { from, to, reason } => { + self.emit(ServerEvent::ModelFallback { from, to, reason: first_line(&reason) }).await; + } + + LoopEvent::LlmFailed { tried, last_error } => { + self.emit(ServerEvent::LlmFailed { tried, last_error }).await; + } + + LoopEvent::Compacted { .. } => {} + + LoopEvent::Error(message) => { + self.emit(ServerEvent::Error { message }).await; + } + + LoopEvent::Cancelled => { + if is_root { + self.emit(ServerEvent::Error { message: "Cancelled by user.".to_string() }).await; + } + } + + LoopEvent::Host(v) => self.forward_host(v).await, + } + } + + /// Host-escaped events (blueprint §4.9): `pending_write` from the + /// approval gate, `agent_question` from the human channel. + async fn forward_host(&self, v: Value) { + match v["type"].as_str() { + Some("pending_write") => { + self.emit(ServerEvent::PendingWrite { + request_id: v["request_id"].as_i64().unwrap_or_default(), + tool_call_id: v["tool_call_id"].as_i64().unwrap_or_default(), + path: v["path"].as_str().unwrap_or_default().to_string(), + old_content: v["old_content"].as_str().map(str::to_string), + new_content: v["new_content"].as_str().unwrap_or_default().to_string(), + }).await; + } + Some("agent_question") => { + self.emit(ServerEvent::AgentQuestion { + request_id: v["request_id"].as_i64().unwrap_or_default(), + tool_call_id: v["tool_call_id"].as_i64().unwrap_or_default(), + title: v["title"].as_str().unwrap_or_default().to_string(), + question: v["question"].as_str().unwrap_or_default().to_string(), + suggested_answers: v["suggested_answers"] + .as_array() + .map(|a| a.iter().filter_map(|s| s.as_str().map(str::to_string)).collect()) + .unwrap_or_default(), + }).await; + } + _ => {} + } + } + + /// `(display_name, icon)` for a tool card, with the MCP friendly-name + /// override (mirrors `tool_ui_meta`). + fn ui_meta(&self, name: &str, args: &Value) -> (String, String) { + let mut meta = self.tools.display_meta(name, args); + if let Some((server, tool)) = crate::mcp::parse_mcp_tool_name(name) + && let Some(friendly) = self.mcp.tool_display_name(server, tool) + { + meta.display_name = friendly; + } + (meta.display_name, meta.icon) + } +} + +fn first_line(s: &str) -> String { + s.lines().next().unwrap_or(s).to_string() +} diff --git a/crates/skald-core/src/loop_adapters/wiring.rs b/crates/skald-core/src/loop_adapters/wiring.rs new file mode 100644 index 0000000..e69de29 diff --git a/crates/skald-core/src/session/handler/config.rs b/crates/skald-core/src/session/handler/config.rs index d1b4552..c952f3e 100644 --- a/crates/skald-core/src/session/handler/config.rs +++ b/crates/skald-core/src/session/handler/config.rs @@ -8,7 +8,7 @@ use super::{ChatSessionHandler, update_scratchpad_tool_def, write_todos_tool_def use super::interface_tools::{AgentRunConfig, InterfaceTool, ToolFuture}; /// Returns an `activate_tools` OpenAI tool definition. -pub(super) fn activate_tools_tool_def() -> Value { +pub(crate) fn activate_tools_tool_def() -> Value { serde_json::json!({ "type": "function", "function": { diff --git a/crates/skald-core/src/session/handler/kernel_turn.rs b/crates/skald-core/src/session/handler/kernel_turn.rs new file mode 100644 index 0000000..f360d5d --- /dev/null +++ b/crates/skald-core/src/session/handler/kernel_turn.rs @@ -0,0 +1,376 @@ +//! Kernel-driven root turn (phase 2, blueprint §14): `handle_message` builds +//! the turn's `TurnParams` from its fields and drives the `agent-loop` kernel +//! instead of `run_agent_turn`. The translator (`EventTranslator`) is the ONE +//! bus subscriber producing the session's `ServerEvent`s. +//! +//! Sub-agents run on the same kernel via `DelegateTool` (sync); async +//! `execute_task` still rides the legacy interface handler until phase 3. +//! Recovery/resume stays on the old path until phase 3 as well. + +use std::sync::Arc; + +use agent_loop::activation::ActivateToolsTool; +use agent_loop::delegate::DelegateTool; +use agent_loop::ids::ConversationId; +use agent_loop::manager::{LiveInput, LoopManager, TurnMeta, TurnParams}; +use agent_loop::model::{ModelHint, ModelSelector}; +use agent_loop::store::{HistoryStore, NewMessage}; +use agent_loop::tool::{Extensions, Tool as LoopTool, ToolSet}; +use core_api::interface_tool::InterfaceTool; +use core_api::message_meta::MessageMetadata; +use serde_json::Value; +use tokio::sync::mpsc; +use tokio_util::sync::CancellationToken; +use tracing::info; + +use crate::chat_event_bus::ToolCallEvent; +use crate::events::ServerEvent; +use crate::loop_adapters::activation::{SkaldActivationSource, SkaldToolActivator}; +use crate::loop_adapters::assembler::SkaldAssembler; +use crate::loop_adapters::builtins::{ + ExecuteTaskAliasTool, LegacyInterfaceTool, SkaldAskUserTool, SkaldHumanChannel, + UpdateScratchpadTool, WriteTodosTool, +}; +use crate::loop_adapters::catalog::SkaldAgentCatalog; +use crate::loop_adapters::gate::ApprovalGate; +use crate::loop_adapters::history::SqliteHistory; +use crate::loop_adapters::hooks::SkaldWritePreviewHook; +use crate::loop_adapters::live_input::PendingLiveInput; +use crate::loop_adapters::preview::PreviewContext; +use crate::loop_adapters::selector::SkaldSelector; +use crate::loop_adapters::system::AgentSystemContext; +use crate::loop_adapters::toolset::{CallerUserId, SkaldToolSet}; +use crate::loop_adapters::translate::EventTranslator; +use crate::tools::tool_names as tn; + +use super::interface_tools::AgentRunConfig; +use super::{ChatSessionHandler, MAX_AGENT_DEPTH, PendingUserInput, TurnOutcome}; + +/// Special-cased names handled natively (never legacy-wrapped). +const NATIVE_NAMES: &[&str] = &[tn::ACTIVATE_TOOLS, tn::EXECUTE_TASK]; + +impl ChatSessionHandler { + /// Runs the root turn on the `agent-loop` kernel. Same observable contract + /// as `run_agent_turn` on the root: events over `tx`, `TurnOutcome` back. + pub(super) async fn run_kernel_turn( + &self, + stack_id: i64, + config: &AgentRunConfig, + user_content: &str, + is_synthetic: bool, + metadata: Option<&MessageMetadata>, + pending_input: Option<&Arc>, + tx: &mpsc::Sender, + ) -> anyhow::Result { + let pool = self.db.clone(); + let shared_pool = self.shared_pool.clone(); + let conv = ConversationId::new(format!("session:{}", self.session_id)); + + // ── Store ── + let store = Arc::new(SqliteHistory::new(pool.clone())); + + // ── Selector (root strength from the agent meta, D14) ── + let strength = crate::agents::load_meta(&config.agent_id) + .ok() + .and_then(|m| m.strength); + let selector: Arc = + Arc::new(SkaldSelector::new(self.llm_manager.clone(), strength)); + + // ── Gate ── + let group_id = self.tool_group_id().await; + let gate = ApprovalGate::new( + self.approval.clone(), + store.clone(), + self.tools.clone(), + self.session_id, + &self.source, + group_id, + self.run_context.clone(), + self.pre_approved.clone(), + self.auto_deny_approvals.clone(), + self.context_label.clone(), + pool.clone(), + shared_pool.clone(), + Some(self.fs.clone()), + ); + + // ── Hooks ── + let preview_hook = Arc::new(SkaldWritePreviewHook::new(PreviewContext { + pool: pool.clone(), + shared_pool: shared_pool.clone(), + fs: Some(self.fs.clone()), + })); + + // ── Manager ── + let manager = Arc::new( + LoopManager::builder() + .models(selector) + .store(store.clone()) + .gate_arc(Arc::new(gate)) + .hook(preview_hook) + .max_rounds(self.max_tool_rounds) + .max_parallel_calls(self.max_parallel_subagents) + .build()?, + ); + + // ── Catalog + delegate ── + let config_defs = Arc::new(config.config_tool_defs.clone()); + let catalog = Arc::new(SkaldAgentCatalog::new( + pool.clone(), + shared_pool.clone(), + self.user_id.clone(), + self.session_id, + self.source.clone(), + self.is_interactive, + self.context_label.clone(), + self.llm_manager.clone(), + self.approval.clone(), + self.clarification.clone(), + self.mcp.clone(), + self.tools.clone(), + config.base_tool_defs.clone(), + config_defs.clone(), + config.memory_tools.clone(), + config.image_tools.clone(), + config.root_only_tool_names.clone(), + self.datetime_config.clone(), + self.max_history_messages, + self.max_tool_result_chars, + self.compactor.is_some(), + Some(self.fs.load()), + self.run_context.read().await.as_ref().and_then(|rc| rc.project_root.clone()), + )); + let delegate = DelegateTool::new(manager.clone(), catalog.clone(), store.clone(), MAX_AGENT_DEPTH as u32); + catalog.set_delegate(delegate.clone()); + + // ── Tool set ── + let mut native: Vec> = Vec::new(); + // activate_tools (root scope — shares the config's grant set so the + // next round sees the new tools, exactly like today). + native.push(Arc::new( + ActivateToolsTool::new(Arc::new(SkaldToolActivator::new( + pool.clone(), + self.mcp.clone(), + config.active_mcp_grants.clone(), + self.session_id, + None, + ))) + .with_definition(super::config::activate_tools_tool_def()), + )); + // execute_task: sync → DelegateTool; async → the legacy interface handler. + { + let et = native_interface(config, tn::EXECUTE_TASK); + let (def, handler) = match et { + Some(it) => (it.definition.clone(), Some(it.handler.clone())), + None => (legacy_execute_task_def(), None), + }; + native.push(Arc::new(ExecuteTaskAliasTool::new( + delegate.clone().with_name(tn::EXECUTE_TASK), + def, + handler, + ))); + } + native.push(Arc::new(SkaldAskUserTool::new( + Arc::new(SkaldHumanChannel::new( + self.clarification.clone(), + self.session_id, + &config.agent_id, + &self.source, + self.is_interactive, + self.context_label.clone(), + )), + store.clone(), + ))); + native.push(Arc::new(UpdateScratchpadTool::new(pool.clone(), self.scratchpad_sid()))); + native.push(Arc::new(WriteTodosTool)); + + // Legacy interface tools (per-surface, minus the native ones). + let legacy: Vec = config + .interface_tools + .iter() + .filter(|it| { + let name = it.definition["function"]["name"].as_str().unwrap_or(""); + !NATIVE_NAMES.contains(&name) + }) + .cloned() + .collect(); + for it in &legacy { + native.push(Arc::new(LegacyInterfaceTool::new(it.clone()))); + } + + let mut toolset = SkaldToolSet::new( + config.base_tool_defs.clone(), + config_defs.clone(), + self.mcp.clone(), + config.active_mcp_grants.clone(), + config.memory_tools.clone(), + config.image_tools.clone(), + legacy, + self.tools.all_tools(), + ) + .with_discovery(self.tool_discovery.clone()); + for t in native { + toolset = toolset.with_native(t); + } + let tools: Arc = Arc::new(toolset); + + // ── System context ── + let system = Arc::new(AgentSystemContext { + agent_id: config.agent_id.clone(), + extra_static: config.extra_system.clone(), + extra_dynamic: config.extra_system_dynamic.clone(), + tail_reminder: config.tail_reminder.clone(), + substitutions: config.system_substitutions.clone(), + pool: pool.clone(), + shared_pool: shared_pool.clone(), + user_id: self.user_id.clone(), + mcp: self.mcp.clone(), + project_root: self.run_context.read().await.as_ref().and_then(|rc| rc.project_root.clone()), + }); + + // ── Assembler ── + let assembler = Arc::new(SkaldAssembler { + pool: pool.clone(), + scratchpad_sid: self.scratchpad_sid(), + datetime_config: self.datetime_config.clone(), + max_history_messages: self.max_history_messages, + max_tool_result_chars: self.max_tool_result_chars, + compactor_enabled: self.compactor.is_some(), + fs: Some(self.fs.load()), + activation: Some(SkaldActivationSource::new( + pool.clone(), + self.mcp.clone(), + config_defs.clone(), + self.session_id, + None, + )), + }); + + // ── Extensions (tool bridge context) ── + let mut extensions = Extensions::new(); + extensions.insert(pool.clone()); + extensions.insert(self.fs.load()); + extensions.insert(Arc::new(CallerUserId(self.user_id.clone()))); + + // ── Live input ── + let live_input: Option> = + pending_input.map(|p| Arc::new(PendingLiveInput::new(p.clone())) as Arc); + + // ── Translator ── + let (translator, shared) = EventTranslator::new( + tx.clone(), + self.tools.clone(), + self.mcp.clone(), + store.clone(), + ); + let stop = CancellationToken::new(); + let translator_task = translator.spawn(manager.events(), stop.clone()); + + // ── Frame + turn ── + let frame = store + .open_frame(&conv, None, agent_loop::store::FrameSpec::root(&config.agent_id)) + .await?; + // The frame opened at session provisioning is the one the old path + // used — assert the mapping (defensive; remove once bedded in). + debug_assert_eq!(frame.get(), stack_id); + + let msg = NewMessage { + role: agent_loop::store::Role::User, + content: user_content.to_string(), + synthetic: is_synthetic, + reasoning: None, + metadata: metadata.and_then(|m| serde_json::to_value(m).ok()), + }; + let params = TurnParams { + frame, + agent: config.agent_id.clone(), + system, + tools, + model_hint: ModelHint::name(config.client_name.clone()), + live_input, + extensions, + meta: TurnMeta { + synthetic: is_synthetic, + interactive: self.is_interactive, + ..TurnMeta::default() + }, + assembler: Some(assembler), + }; + + // Register for /stop, then drive. + *self.kernel_live.lock().unwrap() = Some((manager.clone(), conv.clone())); + let handle = manager.start_turn(conv.clone(), msg, params).await + .map_err(|e| anyhow::anyhow!("kernel turn failed to start: {e}"))?; + let outcome = handle.join().await; + *self.kernel_live.lock().unwrap() = None; + stop.cancel(); + let _ = translator_task.await; + + let shared_state = std::mem::take(&mut *shared.lock().unwrap()); + + match outcome? { + agent_loop::kernel::TurnOutcome::Final { content, message_id, usage, reasoning } => { + let tool_calls: Vec = shared_state.tool_calls; + info!( + session_id = self.session_id, + user_message_id = ?shared_state.user_message_id, + "kernel turn final" + ); + Ok(TurnOutcome::Final { + content, + message_id: message_id.get(), + input_tokens: usage.input_tokens, + output_tokens: usage.output_tokens, + truncated: usage.truncated, + reasoning_content: reasoning, + tool_calls, + }) + } + agent_loop::kernel::TurnOutcome::Cancelled => Ok(TurnOutcome::Cancelled), + agent_loop::kernel::TurnOutcome::Exhausted => Ok(TurnOutcome::Exhausted), + } + } + + /// `/stop` for the kernel-driven turn: cancels the live loop (the legacy + /// `current_cancel` path keeps covering resume/recovery). + pub(super) fn cancel_kernel_turn(&self) { + let live = self.kernel_live.lock().unwrap().clone(); + if let Some((manager, conv)) = live { + manager.cancel(&conv); + } + } +} + +/// Finds an interface tool by name in the run config. +fn native_interface(config: &AgentRunConfig, name: &str) -> Option { + config + .interface_tools + .iter() + .find(|it| it.definition["function"]["name"].as_str() == Some(name)) + .cloned() +} + +/// Fallback definition for `execute_task` when no interface handler was +/// injected (non-interactive sessions): mirrors the injected one. +fn legacy_execute_task_def() -> Value { + serde_json::json!({ + "type": "function", + "function": { + "name": tn::EXECUTE_TASK, + "description": "Execute a task with a sub-agent. mode=sync waits for the result; \ + mode=async schedules it in the background.", + "parameters": { + "type": "object", + "properties": { + "agent_id": { "type": "string" }, + "prompt": { "type": "string" }, + "title": { "type": "string" }, + "description": { "type": "string" }, + "mode": { "type": "string", "enum": ["sync", "async"] }, + "client": { "type": "string" } + }, + "required": ["agent_id", "prompt"] + } + } + }) +} diff --git a/crates/skald-core/src/session/handler/message_builder.rs b/crates/skald-core/src/session/handler/message_builder.rs index dc1fbd8..7af541d 100644 --- a/crates/skald-core/src/session/handler/message_builder.rs +++ b/crates/skald-core/src/session/handler/message_builder.rs @@ -700,12 +700,39 @@ impl MessageBuilder { // ── Free helpers ────────────────────────────────────────────────────────────── +/// `__SHARED_FOLDERS__` section, resolved from the registry (shared with the +/// `agent-loop` adapter's system-context source). +pub(crate) async fn render_shared_folders_section( + shared_pool: &SqlitePool, + user_id: &str, +) -> anyhow::Result { + let rows = crate::db::shared_folders::agent_view(shared_pool, user_id).await?; + Ok(render_shared_folders_table(&rows)) +} + +/// `__USER_PROFILE__` block, resolved from the registry (shared with the +/// `agent-loop` adapter's system-context source). +pub(crate) async fn render_user_profile_section( + shared_pool: &SqlitePool, + user_id: &str, +) -> anyhow::Result { + let user = crate::db::users::get(shared_pool, user_id).await?; + let locale = crate::i18n::resolve_locale( + shared_pool, + user.as_ref().and_then(|u| u.locale.as_deref()), + ).await; + Ok(render_user_profile_block( + user.as_ref(), + &locale, + chrono::Utc::now().date_naive(), + )) +} + /// Renders the shared-folders section body as a Markdown table — one row per /// folder the user belongs to, naming the folder's other members so the model /// knows exactly who sees what is written there. An empty membership yields an /// explicit "not a member" line so the model does not go probing `shared/` paths. -fn render_shared_folders_table(rows: &[crate::db::shared_folders::SharedFolderAccess]) -> String { - /// A free-text cell: single line, pipes escaped (they would split the table). +fn render_shared_folders_table(rows: &[crate::db::shared_folders::SharedFolderAccess]) -> String { /// A free-text cell: single line, pipes escaped (they would split the table). fn cell(s: &str) -> String { s.trim().replace('|', "\\|").replace('\n', " ") } diff --git a/crates/skald-core/src/session/handler/mod.rs b/crates/skald-core/src/session/handler/mod.rs index 59b2e18..9241be4 100644 --- a/crates/skald-core/src/session/handler/mod.rs +++ b/crates/skald-core/src/session/handler/mod.rs @@ -34,6 +34,7 @@ mod config; mod dispatch; mod emitter; mod gate; +mod kernel_turn; mod interface_tools; mod llm_call; mod llm_loop; @@ -43,7 +44,6 @@ mod messages; mod outcome; mod resume; -use emitter::TurnEmitter; pub use interface_tools::{InterfaceTool, ToolFuture}; @@ -54,7 +54,7 @@ pub const DEFAULT_MAX_TOOL_ROUNDS: usize = 20; /// Bounds fan-out so a large batch does not trigger provider rate-limit storms. pub const DEFAULT_MAX_PARALLEL_SUBAGENTS: usize = 4; -pub(super) const MAX_AGENT_DEPTH: i64 = 5; +pub(crate) const MAX_AGENT_DEPTH: i64 = 5; /// A queued user message to be appended to history mid-turn (drained from the /// source inbox at a round boundary). @@ -117,14 +117,14 @@ pub(super) enum TurnOutcome { /// lands inside a multi-byte UTF-8 character (e.g. an em-dash or emoji straddling /// the cut point), which is exactly how a well-formed sub-agent result once /// unwound a whole turn. Used for every event/log preview. -pub(super) fn preview_truncate(s: &str, max_chars: usize) -> String { +pub(crate) fn preview_truncate(s: &str, max_chars: usize) -> String { match s.char_indices().nth(max_chars) { Some((byte_idx, _)) => format!("{}…", &s[..byte_idx]), None => s.to_string(), } } -pub(super) fn update_scratchpad_tool_def() -> Value { +pub(crate) fn update_scratchpad_tool_def() -> Value { json!({ "type": "function", "function": { @@ -154,7 +154,7 @@ pub(super) fn update_scratchpad_tool_def() -> Value { /// agent's own tool-result history. Because conversation history is per-stack, /// it is never visible to sub-agents or to the caller — no DB storage needed. /// The agent re-sends the whole list (TodoWrite-style) on every update. -pub(super) fn write_todos_tool_def() -> Value { +pub(crate) fn write_todos_tool_def() -> Value { json!({ "type": "function", "function": { @@ -192,7 +192,7 @@ pub(super) fn write_todos_tool_def() -> Value { /// to `dispatch_sub_agent` (the InterfaceTool handler is never reached), so only /// the definition is needed here. `agent_id` is required because /// `dispatch_sub_agent` rejects calls without it. -fn execute_subtask_tool_def() -> Value { +pub(crate) fn execute_subtask_tool_def() -> Value { json!({ "type": "function", "function": { @@ -215,7 +215,7 @@ fn execute_subtask_tool_def() -> Value { }) } -fn ask_user_clarification_tool_def() -> Value { +pub(crate) fn ask_user_clarification_tool_def() -> Value { json!({ "type": "function", "function": { @@ -306,7 +306,7 @@ pub struct ChatSessionHandler { pub(super) clarification: Arc, pub(super) event_bus: Arc, /// Human-readable label injected by background runners (e.g. "CronJob: Daily Digest"). - pub(super) context_label: std::sync::RwLock>, + pub(super) context_label: Arc>>, pub(super) memory_manager: Arc, pub(super) image_generator_manager: Arc, /// Prevents concurrent handle_message calls on the same session. @@ -322,21 +322,24 @@ pub struct ChatSessionHandler { /// When true, any tool call that would require human approval is automatically /// denied instead of blocking. Used by TicManager and other headless runners /// that cannot process approval requests. - pub(super) auto_deny_approvals: AtomicBool, + pub(super) auto_deny_approvals: Arc, /// Tool-call ids the user already approved via a resolve endpoint after a restart /// (no live oneshot to unblock). The next resume's approval gate skips re-gating /// these so a post-restart approve dispatches the tool without a second prompt. - pub(super) pre_approved: std::sync::Mutex>, + pub(super) pre_approved: Arc>>, /// Context compactor, shared across all sessions. `None` when compaction /// is disabled (no `compaction` section in config). pub(super) compactor: Option>, + /// The live kernel-driven turn (manager + conversation) for `/stop` + /// routing (phase 2). `None` between turns / on legacy paths. + pub(super) kernel_live: std::sync::Mutex, agent_loop::ids::ConversationId)>>, /// Input token count from the most recently completed turn, stored /// atomically so the next `handle_message` call can decide whether to /// compact before processing the new message. Zero means unknown /// (provider did not report usage on the first turn). pub(super) last_input_tokens: AtomicU32, /// Active RunContext for this session. `None` means the "default" group is used implicitly. - pub(super) run_context: tokio::sync::RwLock>, + pub(super) run_context: Arc>>, /// When set, scratchpad reads/writes use this session_id instead of `self.session_id`. /// Used by async sub-tasks to share the parent's scratchpad. pub(super) scratchpad_session_id: std::sync::OnceLock, @@ -395,14 +398,15 @@ impl ChatSessionHandler { memory_manager, image_generator_manager, compactor, - context_label: std::sync::RwLock::new(None), + context_label: Arc::new(std::sync::RwLock::new(None)), processing: Mutex::new(()), current_cancel: std::sync::Mutex::new(CancellationToken::new()), - auto_deny_approvals: AtomicBool::new(false), - pre_approved: std::sync::Mutex::new(std::collections::HashSet::new()), + auto_deny_approvals: Arc::new(AtomicBool::new(false)), + pre_approved: Arc::new(std::sync::Mutex::new(std::collections::HashSet::new())), last_input_tokens: AtomicU32::new(0), - run_context: tokio::sync::RwLock::new(run_context), + run_context: Arc::new(tokio::sync::RwLock::new(run_context)), scratchpad_session_id: std::sync::OnceLock::new(), + kernel_live: std::sync::Mutex::new(None), } } @@ -453,6 +457,7 @@ impl ChatSessionHandler { /// sub-agent recursion: the token is never reset mid-turn. pub fn cancel(&self) { self.current_cancel.lock().unwrap().cancel(); + self.cancel_kernel_turn(); } /// True if a turn is currently in flight (the `processing` mutex is held for @@ -547,7 +552,7 @@ impl ChatSessionHandler { let token = CancellationToken::new(); *self.current_cancel.lock().unwrap() = token.clone(); let pool = &self.db; - let em = TurnEmitter::new(&tx); + let user_content = content.to_string(); // saved for the ChatEvent publication // Retrieve memory context (Honcho or other backend) for this turn. // Kept SEPARATE from extra_system_context (the static part) so it can be @@ -629,43 +634,30 @@ impl ChatSessionHandler { } } - let user_content = content.to_string(); // save before TurnOutcome::Final shadows `content` - let user_message_id = chat_history::append_with_metadata(pool, stack.id, &chat_history::Role::User, content, is_synthetic, None, metadata.as_ref()).await?; - - // Telnet-style echo: the bubble appears only once the message is persisted. - // Synthetic turns (TIC/notification) never produce a user bubble. - if !is_synthetic { - let attachments = metadata.as_ref().map(|m| m.attachments.clone()).unwrap_or_default(); - // A custom slash command persists its expanded template (for LLM replay) - // but the bubble must show the typed command — emit `display` when present. - let echo = metadata.as_ref() - .and_then(|m| m.command.as_ref()) - .map(|c| c.display.clone()) - .unwrap_or_else(|| user_content.clone()); - em.user_message(user_message_id, echo, attachments).await; - } - // Resume any tool calls left pending from a previous interrupted session. // They are re-gated (rules may have changed) and executed before the LLM runs. + // (Runs before the kernel turn, which appends the user message itself — + // resumed results belong to the previous turn and land first.) self.resume_pending_tools(stack.id, &config, &token, &tx).await?; - let outcome = self.run_agent_turn(stack.id, &config, &token, &tx, pending_input.as_ref()).await?; + let outcome = self.run_kernel_turn( + stack.id, &config, content, is_synthetic, metadata.as_ref(), pending_input.as_ref(), &tx, + ).await?; match outcome { - TurnOutcome::Final { content, message_id, input_tokens, output_tokens, truncated, reasoning_content, tool_calls } => { + TurnOutcome::Final { content, message_id, input_tokens, output_tokens, truncated: _, reasoning_content: _, tool_calls } => { // Persist token count so the *next* handle_message call knows // whether to compact before running the LLM loop. if let Some(t) = input_tokens { self.last_input_tokens.store(t, Ordering::Relaxed); } info!(session_id = self.session_id, stack_id = stack.id, ?input_tokens, ?output_tokens, "handle_message done"); - if truncated { - warn!(session_id = self.session_id, ?output_tokens, "response truncated (max_tokens)"); - em.truncated(output_tokens).await; - } - em.done(message_id, stack.id, content.clone(), input_tokens, output_tokens, reasoning_content).await; + // NB: the WS echo (UserMessage), the Done and — when cut off — + // the Truncated events were already emitted by the kernel's + // event translator during the turn. // Publish both messages to the event bus now that both are in the DB. + let user_message_id = shared_user_message_id(&self.db, stack.id, message_id).await; let now = chrono::Utc::now(); self.event_bus.user_message(ChatEvent { session_id: self.session_id, @@ -698,14 +690,29 @@ impl ChatSessionHandler { } TurnOutcome::Cancelled => { info!(session_id = self.session_id, "handle_message cancelled by user"); - em.error("Cancelled by user.".to_string()).await; + // The "Cancelled by user." error event was already emitted by + // the translator (root LoopEvent::Cancelled). Err(anyhow::anyhow!("Turn cancelled by user")) } TurnOutcome::Exhausted => { error!(session_id = self.session_id, max_rounds = self.max_tool_rounds, "tool-call loop exhausted without final answer"); - em.error(format!("Exceeded {} tool-call rounds without a final answer.", self.max_tool_rounds)).await; + tx.send(ServerEvent::Error { + message: format!("Exceeded {} tool-call rounds without a final answer.", self.max_tool_rounds), + }).await.ok(); Err(anyhow::anyhow!("tool-call loop exhausted after {} rounds without a final answer", self.max_tool_rounds)) } } } } + +/// The user message of the current turn: the latest User row before the final +/// assistant message (used for the ChatEvent publication). +async fn shared_user_message_id(pool: &sqlx::SqlitePool, stack_id: i64, _final_id: i64) -> i64 { + let history = chat_history::for_stack(pool, stack_id).await.unwrap_or_default(); + history + .iter() + .rev() + .find(|m| matches!(m.role, chat_history::Role::User | chat_history::Role::Agent)) + .map(|m| m.id) + .unwrap_or_default() +} diff --git a/crates/skald-core/src/tools/mod.rs b/crates/skald-core/src/tools/mod.rs index d74321c..98c3e68 100644 --- a/crates/skald-core/src/tools/mod.rs +++ b/crates/skald-core/src/tools/mod.rs @@ -108,6 +108,16 @@ impl ToolRegistry { self.tools.insert(tool.name().to_string(), tool); } + /// A tool by name (the execution side — used by the agent-loop adapters). + pub fn get_tool(&self, name: &str) -> Option> { + self.tools.get(name).cloned() + } + + /// Every registered tool (the execution side of a `SkaldToolSet`). + pub fn all_tools(&self) -> Vec> { + self.tools.values().cloned().collect() + } + /// Tool definitions for the root agent (depth = 0): excludes sub_agents_only tools. pub fn openai_definitions(&self) -> Vec { self.tools.values()