Release 0.2.0 #4

Merged
dguiducci merged 96 commits from main into release 2026-08-17 18:07:20 +01:00
35 changed files with 3160 additions and 89 deletions
Showing only changes of commit 0297fe71bd - Show all commits
Generated
+2
View File
@@ -1604,9 +1604,11 @@ checksum = "ec9d92d097f4749b64e8cc33d924d9f40a2d4eb91402b458014b781f5733d60f"
name = "honcho-client"
version = "0.1.0"
dependencies = [
"anyhow",
"reqwest 0.13.4",
"serde",
"serde_json",
"tokio",
"tracing",
]
+13 -1
View File
@@ -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<dyn ToolActivator>,
definition_override: Option<Value>,
}
impl ActivateToolsTool {
pub fn new(activator: Arc<dyn ToolActivator>) -> Self { Self { activator } }
pub fn new(activator: Arc<dyn ToolActivator>) -> 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": {
+3
View File
@@ -275,6 +275,9 @@ fn project_message(out: &mut Vec<Value>, 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
+417
View File
@@ -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<dyn SystemContextSource>,
/// 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<Arc<dyn ToolSet>>,
/// Model pin (bypasses AUTO). Strength is resolved by the host's selector.
pub model: Option<ModelHint>,
/// Per-child selector override (e.g. a different required strength, D14).
pub selector: Option<Arc<dyn ModelSelector>>,
/// Per-child assembler override (e.g. scoped DTL activation).
pub assembler: Option<Arc<dyn crate::context::ContextAssembler>>,
}
/// 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<String>,
pub add: Vec<Arc<dyn Tool>>,
}
impl ToolSelection {
pub fn inherit() -> Self { Self::default() }
pub fn minus(names: impl IntoIterator<Item = impl Into<String>>) -> Self {
Self { remove: names.into_iter().map(Into::into).collect(), add: Vec::new() }
}
pub fn plus(tools: Vec<Arc<dyn Tool>>) -> 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<AgentProfile>;
async fn list(&self, kind: AgentKind) -> Vec<AgentSummary>;
/// 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<dyn ToolSet>,
remove: Vec<String>,
add: Vec<Arc<dyn Tool>>,
}
impl ToolSet for FilteredToolSet {
fn defs(&self, model: &crate::model::ModelInfo) -> Vec<Value> {
let mut defs: Vec<Value> = 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<Arc<dyn Tool>> {
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<LoopManager>,
catalog: Arc<dyn AgentCatalog>,
store: Arc<dyn HistoryStore>,
max_depth: u32,
name: String,
definition_override: Option<Value>,
}
impl DelegateTool {
pub fn new(
manager: Arc<LoopManager>,
catalog: Arc<dyn AgentCatalog>,
store: Arc<dyn HistoryStore>,
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<String>) -> 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<ToolOutput, ToolFailure> {
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<dyn ToolSet> = match profile.toolset.clone() {
Some(ts) => ts,
None => {
let parent_tools = ctx
.extensions
.get::<SharedToolSet>()
.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<ToolOutput, ToolFailure> {
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<AgentProfile>,
}
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<AgentProfile> {
self.profiles
.iter()
.find(|p| p.id == id)
.cloned()
.ok_or_else(|| anyhow::anyhow!("unknown agent `{id}`"))
}
async fn list(&self, kind: AgentKind) -> Vec<AgentSummary> {
self.profiles
.iter()
.filter(|p| p.kind == kind)
.map(|p| AgentSummary { id: p.id.clone(), kind: p.kind, description: String::new() })
.collect()
}
}
+6
View File
@@ -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,
+1
View File
@@ -17,6 +17,7 @@ pub struct PendingCall {
pub name: String,
pub args: Value,
pub frame: FrameId,
pub parent_frame: Option<FrameId>,
pub agent: String,
/// Host free-form (source, permission group, …).
pub extensions: Extensions,
+22 -8
View File
@@ -19,6 +19,8 @@ pub struct Question {
pub suggested: Vec<String>,
/// 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<dyn HumanChannel>,
store: Arc<dyn HistoryStore>,
name: String,
}
impl AskUserTool {
pub fn new(channel: Arc<dyn HumanChannel>, store: Arc<dyn HistoryStore>) -> 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<String>) -> 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<ToolOutput, ToolFailure> {
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
+12 -6
View File
@@ -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<dyn ModelSelector> = params.selector.as_ref().unwrap_or(&deps.models);
// First selection of the turn.
let mut handle: ModelHandle = match deps.models.select(&params.model_hint, &[]).await {
let mut handle: ModelHandle = match selector.select(&params.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(&params.model_hint, &tried).await {
match selector.select(&params.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(),
};
+5
View File
@@ -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};
+27 -7
View File
@@ -76,6 +76,12 @@ pub struct LoopParams {
pub system: Arc<dyn SystemContextSource>,
pub tools: Arc<dyn ToolSet>,
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<Arc<dyn crate::model::ModelSelector>>,
/// Parent-linked cancellation (DelegateTool passes `ctx.cancel.child_token()`):
/// `None` = a fresh scope. Cancellation stays sticky down the tree.
pub token: Option<CancellationToken>,
pub live_input: Option<Arc<dyn LiveInput>>,
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<TurnHandle, StartError> {
{
let registry = self.registry.lock().unwrap();
if registry.contains_key(&params.conversation) {
return Err(StartError::AlreadyRunning);
}
}
self.spawn(params)
self.spawn_detached(params)
}
fn spawn_detached(&self, params: LoopParams) -> Result<TurnHandle, StartError> {
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<TurnHandle, StartError> {
+14 -4
View File
@@ -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<FrameId>;
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<Option<FrameRecord>>;
/// All active frames of a conversation (recovery: batch detection, cascade).
async fn active_frames(&self, conv: &ConversationId) -> crate::Result<Vec<FrameRecord>>;
async fn deepest_active(&self, conv: &ConversationId) -> crate::Result<Option<FrameRecord>>;
@@ -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<Option<StoredCall>>;
/// 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<Vec<StoredCall>>;
// ── summaries ──
+24
View File
@@ -69,6 +69,11 @@ impl HistoryStore for InMemoryStore {
Ok(())
}
async fn get_frame(&self, frame: FrameId) -> crate::Result<Option<FrameRecord>> {
let i = self.inner.lock().unwrap();
Ok(i.frames.get(&frame).cloned())
}
async fn active_frames(&self, conv: &ConversationId) -> crate::Result<Vec<FrameRecord>> {
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<Option<StoredCall>> {
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<Vec<StoredCall>> {
let i = self.inner.lock().unwrap();
Ok(i.messages
+6
View File
@@ -207,6 +207,12 @@ pub trait ToolSet: Send + Sync {
fn find(&self, name: &str) -> Option<Arc<dyn Tool>>;
}
/// Wrapper so `Arc<dyn ToolSet>` 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<dyn ToolSet>);
/// A trivial `ToolSet` from a list of tools (testing, simple hosts).
pub struct ToolRegistry {
tools: Vec<Arc<dyn Tool>>,
+138
View File
@@ -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<StaticSystemContext>,
}
#[async_trait]
impl AgentCatalog for TestCatalog {
async fn get(&self, id: &str, _child_frame: agent_loop::ids::FrameId) -> agent_loop::Result<AgentProfile> {
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<agent_loop::delegate::AgentSummary> {
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<dyn AgentCatalog> = Arc::new(TestCatalog {
context: Arc::new(StaticSystemContext::new("You are a researcher.")),
});
let delegate: Arc<dyn Tool> = 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<dyn AgentCatalog> = Arc::new(TestCatalog {
context: Arc::new(StaticSystemContext::new("worker")),
});
let delegate: Arc<dyn Tool> = 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<String> = 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;
+1
View File
@@ -9,6 +9,7 @@ pub type ToolFuture = Pin<Box<dyn std::future::Future<Output = anyhow::Result<St
/// A single LLM-callable tool injected by a specific interface (Telegram, Web, Cron, …).
///
/// The handler closure captures interface-specific state (e.g. `Arc<Bot>` + `ChatId`).
#[derive(Clone)]
pub struct InterfaceTool {
/// OpenAI-format tool definition sent to the LLM in the tools array.
pub definition: Value,
+5
View File
@@ -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"
+1 -1
View File
@@ -12,7 +12,7 @@
//! # Required secret
//!
//! Set before enabling the plugin:
//! ```
//! ```text
//! set_secret("HUGGINGFACE_TOKEN", "hf_...")
//! ```
//! Get a token at <https://huggingface.co/settings/tokens>.
@@ -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<String> = 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<Option<String>> = 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<sqlx::SqlitePool>,
/// 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<usize>,
/// 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<Arc<UserFs>>,
/// DTL activations (consulted only in non-Inline modes).
pub activation: Option<SkaldActivationSource>,
}
#[agent_loop::async_trait]
impl ContextAssembler for SkaldAssembler {
async fn build(
&self,
store: &Arc<dyn HistoryStore>,
input: &AssembleInput,
) -> agent_loop::Result<Vec<Value>> {
let mut out: Vec<Value> = 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(
"<scratchpad>\n \
<!-- Temporary notes shared by all agents in this session. Not persisted across sessions. -->\n"
);
for (k, v) in &scratch {
s.push_str(&format!(" <note key=\"{k}\">{v}</note>\n"));
}
s.push_str("</scratchpad>");
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<i64, Vec<Value>> =
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<MessageMetadata> = 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<Value> = 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<Value> = 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<MediaRef> = Vec::new();
for tc in &entry.calls {
if let Some(mj) = tc.extras["media"].as_str()
&& let Ok(mut v) = serde_json::from_str::<Vec<MediaRef>>(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<String> {
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::<chrono_tz::Tz>().ok())
.or_else(|| system_timezone().and_then(|s| s.parse::<chrono_tz::Tz>().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<Value>, text: String, media: Vec<Value>) {
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::<Vec<serde_json::Value>>(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::<Vec<_>>().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)")
}
}
}
@@ -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<ClarificationManager>,
session_id: i64,
agent_id: String,
source: String,
is_interactive: bool,
context_label: Arc<std::sync::RwLock<Option<String>>>,
}
impl SkaldHumanChannel {
pub fn new(
clarification: Arc<ClarificationManager>,
session_id: i64,
agent_id: impl Into<String>,
source: impl Into<String>,
is_interactive: bool,
context_label: Arc<std::sync::RwLock<Option<String>>>,
) -> 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<String, HumanGone> {
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<SqlitePool>,
sid: i64,
}
impl UpdateScratchpadTool {
pub fn new(pool: Arc<SqlitePool>, 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<ToolOutput, ToolFailure> {
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<ToolOutput, ToolFailure> {
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<dyn HumanChannel>, store: Arc<dyn agent_loop::store::HistoryStore>) -> 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<ToolOutput, ToolFailure> {
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<Arc<dyn Fn(Value) -> ToolFuture + Send + Sync>>,
}
impl ExecuteTaskAliasTool {
pub fn new(
delegate: DelegateTool,
definition: Value,
async_handler: Option<Arc<dyn Fn(Value) -> 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<ToolOutput, ToolFailure> {
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<dyn Fn(Value) -> 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<ToolOutput, ToolFailure> {
(self.handler)(args)
.await
.map(ToolOutput::Text)
.map_err(|e| ToolFailure::Failed(e.to_string()))
}
}
@@ -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<SqlitePool>,
shared_pool: Arc<SqlitePool>,
user_id: String,
session_id: i64,
source: String,
is_interactive: bool,
context_label: Arc<RwLock<Option<String>>>,
llm_manager: Arc<LlmManager>,
approval: Arc<ApprovalManager>,
clarification: Arc<ClarificationManager>,
mcp: Arc<dyn McpProvider>,
registry: Arc<ToolRegistry>,
/// Parent turn's derived def lists (the child's base derives from these).
base_defs: Vec<serde_json::Value>,
config_defs: Arc<Vec<serde_json::Value>>,
memory_tools: Vec<Arc<dyn crate::tools::Tool>>,
image_tools: Vec<Arc<dyn crate::tools::Tool>>,
core_tools: Vec<Arc<dyn crate::tools::Tool>>,
root_only: Vec<String>,
/// The delegate tool, injected post-construction (catalog ↔ delegate cycle).
delegate: RwLock<Option<Arc<DelegateTool>>>,
/// Per-turn assembler knobs shared with children.
datetime_config: DatetimeConfig,
max_history_messages: usize,
max_tool_result_chars: Option<usize>,
compactor_enabled: bool,
fs: Option<Arc<core_api::user_fs::UserFs>>,
project_root: Option<String>,
}
impl SkaldAgentCatalog {
#[allow(clippy::too_many_arguments)]
pub fn new(
pool: Arc<SqlitePool>,
shared_pool: Arc<SqlitePool>,
user_id: String,
session_id: i64,
source: String,
is_interactive: bool,
context_label: Arc<RwLock<Option<String>>>,
llm_manager: Arc<LlmManager>,
approval: Arc<ApprovalManager>,
clarification: Arc<ClarificationManager>,
mcp: Arc<dyn McpProvider>,
registry: Arc<ToolRegistry>,
base_defs: Vec<serde_json::Value>,
config_defs: Arc<Vec<serde_json::Value>>,
memory_tools: Vec<Arc<dyn crate::tools::Tool>>,
image_tools: Vec<Arc<dyn crate::tools::Tool>>,
root_only: Vec<String>,
datetime_config: DatetimeConfig,
max_history_messages: usize,
max_tool_result_chars: Option<usize>,
compactor_enabled: bool,
fs: Option<Arc<core_api::user_fs::UserFs>>,
project_root: Option<String>,
) -> 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<AgentProfile> {
// 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<serde_json::Value> = 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<RwLock<HashSet<String>>> = 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<Arc<dyn LoopTool>> = 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<dyn agent_loop::tool::ToolSet> = 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<dyn ContextAssembler> = 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<AgentSummary> {
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");
}
}
}
+151 -15
View File
@@ -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<RwLock<Option<RunContext>>>,
pre_approved: Arc<Mutex<HashSet<i64>>>,
auto_deny: Arc<AtomicBool>,
context_label: Arc<RwLock<Option<String>>>,
context_label: Arc<std::sync::RwLock<Option<String>>>,
/// For the `PendingWrite` diff: owner pool (user-memory), shared pool
/// (shared-memory), and the caller's fs view (host paths).
pool: Arc<SqlitePool>,
shared_pool: Arc<SqlitePool>,
fs: Option<SharedFs>,
}
impl ApprovalGate {
@@ -50,7 +58,10 @@ impl ApprovalGate {
run_context: Arc<RwLock<Option<RunContext>>>,
pre_approved: Arc<Mutex<HashSet<i64>>>,
auto_deny: Arc<AtomicBool>,
context_label: Arc<RwLock<Option<String>>>,
context_label: Arc<std::sync::RwLock<Option<String>>>,
pool: Arc<SqlitePool>,
shared_pool: Arc<SqlitePool>,
fs: Option<SharedFs>,
) -> 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<String> {
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<String> {
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(),
};
@@ -184,6 +184,30 @@ impl HistoryStore for SqliteHistory {
Ok(())
}
async fn get_frame(&self, frame: FrameId) -> agent_loop::Result<Option<FrameRecord>> {
let row = sqlx::query_as::<_, (i64, i64, String, Option<String>, i64, Option<i64>, Option<String>)>(
"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<Vec<FrameRecord>> {
let session_id = Self::session_id(conv)?;
let rows = sqlx::query_as::<_, (i64, i64, String, Option<String>, i64, Option<i64>)>(
@@ -319,6 +343,24 @@ impl HistoryStore for SqliteHistory {
Ok(())
}
async fn get_call(&self, id: ToolCallId) -> agent_loop::Result<Option<StoredCall>> {
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<Vec<StoredCall>> {
// 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.
@@ -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<HashMap<i64, Option<String>>>,
}
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;
}
}
@@ -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<dyn PendingUserInput>,
}
impl PendingLiveInput {
pub fn new(inner: Arc<dyn PendingUserInput>) -> Self { Self { inner } }
}
#[agent_loop::async_trait]
impl LiveInput for PendingLiveInput {
async fn drain(&self) -> Vec<NewMessage> {
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()
}
}
@@ -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;
@@ -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<String>) -> Option<String> {
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<SqlitePool>,
pub shared_pool: Arc<SqlitePool>,
pub fs: Option<SharedFs>,
}
/// 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<String> {
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<String> {
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,
}
}
@@ -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<String>,
/// Dynamic extra context (Honcho memory merged with per-turn overrides),
/// emitted as the dynamic tail.
pub extra_dynamic: Option<String>,
pub tail_reminder: Option<String>,
pub substitutions: HashMap<String, String>,
/// Owner pool (`user-memory/` notes).
pub pool: Arc<SqlitePool>,
/// Shared pool (`shared-memory/`, shared folders, user profile).
pub shared_pool: Arc<SqlitePool>,
pub user_id: String,
pub mcp: Arc<dyn McpProvider>,
/// Project root for `__PROJECT_ROOT__` expansion in `inject_memory`.
pub project_root: Option<String>,
}
#[agent_loop::async_trait]
impl SystemContextSource for AgentSystemContext {
async fn system_context(&self, _turn: &TurnInfo) -> agent_loop::Result<SystemContext> {
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<memory_file path=\"{display}\">\n{c}\n</memory_file>\n"
)),
None => static_content.push_str(&format!(
"\n<memory_file path=\"{display}\">\n(file not created yet)\n</memory_file>\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<skills_index path=\"{display}\">\n{c}\n</skills_index>\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>, 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<String> = 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__<name>__<tool>`:\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
}
}
+25 -3
View File
@@ -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<Value>,
config_defs: Vec<Value>,
config_defs: Arc<Vec<Value>>,
mcp: Arc<dyn McpProvider>,
grants: Arc<RwLock<HashSet<String>>>,
memory_tools: Vec<Arc<dyn crate::tools::Tool>>,
@@ -212,13 +212,15 @@ pub struct SkaldToolSet {
core_tools: Vec<Arc<dyn crate::tools::Tool>>,
/// Extra crate-native tools for find() (bridge-free).
native_tools: Vec<Arc<dyn LoopTool>>,
/// Records tools offered to the LLM each round (Security-groups UI).
discovery: Option<Arc<crate::tool_discovery::ToolDiscovery>>,
}
impl SkaldToolSet {
#[allow(clippy::too_many_arguments)]
pub fn new(
base_defs: Vec<Value>,
config_defs: Vec<Value>,
config_defs: Arc<Vec<Value>>,
mcp: Arc<dyn McpProvider>,
grants: Arc<RwLock<HashSet<String>>>,
memory_tools: Vec<Arc<dyn crate::tools::Tool>>,
@@ -236,13 +238,24 @@ impl SkaldToolSet {
interface_tools,
core_tools,
native_tools: Vec::new(),
discovery: None,
}
}
pub fn with_discovery(mut self, discovery: Arc<crate::tool_discovery::ToolDiscovery>) -> Self {
self.discovery = Some(discovery);
self
}
pub fn with_native(mut self, tool: Arc<dyn LoopTool>) -> Self {
self.native_tools.push(tool);
self
}
pub fn with_native_all(mut self, tools: Vec<Arc<dyn LoopTool>>) -> 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<RwLock<HashSet<String>>>) -> 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![],
@@ -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<ServerEvent>,
tools: Arc<ToolRegistry>,
mcp: Arc<dyn McpProvider>,
store: Arc<dyn HistoryStore>,
shared: Arc<std::sync::Mutex<TranslateShared>>,
}
/// 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<i64>,
/// Accumulated tool calls of the turn (done/failed only — mirrors the old
/// `all_tool_calls` accumulate rules).
pub tool_calls: Vec<core_api::bus::ToolCallEvent>,
}
impl EventTranslator {
pub fn new(
tx: mpsc::Sender<ServerEvent>,
tools: Arc<ToolRegistry>,
mcp: Arc<dyn McpProvider>,
store: Arc<dyn HistoryStore>,
) -> (Self, Arc<std::sync::Mutex<TranslateShared>>) {
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<Event<LoopEvent>>, 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<LoopEvent>) {
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<MessageMetadata> = 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()
}
@@ -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": {
@@ -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<dyn PendingUserInput>>,
tx: &mpsc::Sender<ServerEvent>,
) -> anyhow::Result<TurnOutcome> {
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<dyn ModelSelector> =
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<Arc<dyn LoopTool>> = 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<InterfaceTool> = 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<dyn ToolSet> = 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<Arc<dyn LiveInput>> =
pending_input.map(|p| Arc::new(PendingLiveInput::new(p.clone())) as Arc<dyn LiveInput>);
// ── 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<ToolCallEvent> = 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<InterfaceTool> {
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"]
}
}
})
}
@@ -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<String> {
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<String> {
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', " ")
}
+48 -41
View File
@@ -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<ClarificationManager>,
pub(super) event_bus: Arc<ChatEventBus>,
/// Human-readable label injected by background runners (e.g. "CronJob: Daily Digest").
pub(super) context_label: std::sync::RwLock<Option<String>>,
pub(super) context_label: Arc<std::sync::RwLock<Option<String>>>,
pub(super) memory_manager: Arc<MemoryManager>,
pub(super) image_generator_manager: Arc<ImageGeneratorManager>,
/// 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<AtomicBool>,
/// 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<std::collections::HashSet<i64>>,
pub(super) pre_approved: Arc<std::sync::Mutex<std::collections::HashSet<i64>>>,
/// Context compactor, shared across all sessions. `None` when compaction
/// is disabled (no `compaction` section in config).
pub(super) compactor: Option<Arc<ContextCompactor>>,
/// 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<Option<(Arc<agent_loop::manager::LoopManager>, 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<Option<RunContext>>,
pub(super) run_context: Arc<tokio::sync::RwLock<Option<RunContext>>>,
/// 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<i64>,
@@ -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()
}
+10
View File
@@ -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<Arc<dyn Tool>> {
self.tools.get(name).cloned()
}
/// Every registered tool (the execution side of a `SkaldToolSet`).
pub fn all_tools(&self) -> Vec<Arc<dyn Tool>> {
self.tools.values().cloned().collect()
}
/// Tool definitions for the root agent (depth = 0): excludes sub_agents_only tools.
pub fn openai_definitions(&self) -> Vec<Value> {
self.tools.values()