Merge pull request #1645 from nearai/staging-promote/0341fcc9-23558273569

chore: promote staging to staging-promote/6daa2f15-23538193544 (2026-03-25 18:47 UTC)
This commit is contained in:
Henry Park
2026-03-25 15:13:23 -07:00
committed by GitHub
43 changed files with 1157 additions and 88 deletions
+19 -5
View File
@@ -12,6 +12,7 @@ jobs:
tests:
name: Tests (${{ matrix.name }})
runs-on: ubuntu-latest
timeout-minutes: 45
strategy:
fail-fast: false
matrix:
@@ -40,11 +41,14 @@ jobs:
- name: Build WASM channels (for integration tests)
run: ./scripts/build-wasm-extensions.sh --channels
- name: Run Tests
run: cargo test ${{ matrix.flags }} -- --nocapture
run: |
timeout --signal=INT --kill-after=30s 40m \
cargo test ${{ matrix.flags }} -- --nocapture
heavy-integration-tests:
name: Heavy Integration Tests
runs-on: ubuntu-latest
timeout-minutes: 20
steps:
- name: Checkout repository
uses: actions/checkout@v6
@@ -58,9 +62,13 @@ jobs:
- name: Build Telegram WASM channel
run: cargo build --manifest-path channels-src/telegram/Cargo.toml --target wasm32-wasip2 --release
- name: Run thread scheduling integration tests
run: cargo test --no-default-features --features libsql,integration --test e2e_thread_scheduling -- --nocapture
run: |
timeout --signal=INT --kill-after=30s 15m \
cargo test --no-default-features --features libsql,integration --test e2e_thread_scheduling -- --nocapture
- name: Run Telegram thread-scope regression test
run: cargo test --features integration --test telegram_auth_integration test_private_messages_use_chat_id_as_thread_scope -- --exact
run: |
timeout --signal=INT --kill-after=30s 10m \
cargo test --features integration --test telegram_auth_integration test_private_messages_use_chat_id_as_thread_scope -- --exact
telegram-tests:
name: Telegram Channel Tests
@@ -68,6 +76,7 @@ jobs:
github.event_name != 'pull_request' ||
github.base_ref != 'staging'
runs-on: ubuntu-latest
timeout-minutes: 15
steps:
- name: Checkout repository
uses: actions/checkout@v6
@@ -75,7 +84,9 @@ jobs:
uses: dtolnay/rust-toolchain@stable
- uses: Swatinem/rust-cache@v2
- name: Run Telegram Channel Tests
run: cargo test --manifest-path channels-src/telegram/Cargo.toml -- --nocapture
run: |
timeout --signal=INT --kill-after=30s 10m \
cargo test --manifest-path channels-src/telegram/Cargo.toml -- --nocapture
windows-build:
name: Windows Build (${{ matrix.name }})
@@ -110,6 +121,7 @@ jobs:
github.event_name != 'pull_request' ||
github.base_ref != 'staging'
runs-on: ubuntu-latest
timeout-minutes: 30
steps:
- name: Checkout repository
uses: actions/checkout@v6
@@ -125,7 +137,9 @@ jobs:
- name: Build all WASM extensions against current WIT
run: ./scripts/build-wasm-extensions.sh
- name: Instantiation test (host linker compatibility)
run: cargo test --all-features wit_compat -- --nocapture
run: |
timeout --signal=INT --kill-after=30s 20m \
cargo test --all-features wit_compat -- --nocapture
bench-compile:
name: Benchmark Compilation
+55
View File
@@ -7,6 +7,32 @@
use serde::{Deserialize, Serialize};
/// A single tool decision in a reasoning update (SSE DTO).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ToolDecisionDto {
pub tool_name: String,
pub rationale: String,
}
impl ToolDecisionDto {
/// Parse a list of tool decisions from a JSON array value.
pub fn from_json_array(value: &serde_json::Value) -> Vec<Self> {
value
.as_array()
.map(|arr| {
arr.iter()
.filter_map(|d| {
Some(Self {
tool_name: d.get("tool_name")?.as_str()?.to_string(),
rationale: d.get("rationale")?.as_str()?.to_string(),
})
})
.collect()
})
.unwrap_or_default()
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum AppEvent {
@@ -163,6 +189,23 @@ pub enum AppEvent {
#[serde(skip_serializing_if = "Option::is_none")]
message: Option<String>,
},
/// Agent reasoning update (why it chose specific tools).
#[serde(rename = "reasoning_update")]
ReasoningUpdate {
narrative: String,
decisions: Vec<ToolDecisionDto>,
#[serde(skip_serializing_if = "Option::is_none")]
thread_id: Option<String>,
},
/// Reasoning update for a sandbox job.
#[serde(rename = "job_reasoning")]
JobReasoning {
job_id: String,
narrative: String,
decisions: Vec<ToolDecisionDto>,
},
}
impl AppEvent {
@@ -191,6 +234,8 @@ impl AppEvent {
Self::Suggestions { .. } => "suggestions",
Self::TurnCost { .. } => "turn_cost",
Self::ExtensionStatus { .. } => "extension_status",
Self::ReasoningUpdate { .. } => "reasoning_update",
Self::JobReasoning { .. } => "job_reasoning",
}
}
}
@@ -311,6 +356,16 @@ mod tests {
status: String::new(),
message: None,
},
AppEvent::ReasoningUpdate {
narrative: String::new(),
decisions: vec![],
thread_id: None,
},
AppEvent::JobReasoning {
job_id: String::new(),
narrative: String::new(),
decisions: vec![],
},
];
for variant in &variants {
+1 -1
View File
@@ -3,5 +3,5 @@
mod event;
mod util;
pub use event::AppEvent;
pub use event::{AppEvent, ToolDecisionDto};
pub use util::truncate_preview;
+79 -4
View File
@@ -16,6 +16,7 @@ use crate::agent::context_monitor::ContextMonitor;
use crate::agent::heartbeat::spawn_heartbeat;
use crate::agent::routine_engine::{RoutineEngine, spawn_cron_ticker};
use crate::agent::self_repair::{DefaultSelfRepair, RepairResult, SelfRepair};
use crate::agent::session::ThreadState;
use crate::agent::session_manager::SessionManager;
use crate::agent::submission::{Submission, SubmissionParser, SubmissionResult};
use crate::agent::{HeartbeatConfig as AgentHeartbeatConfig, Router, Scheduler, SchedulerDeps};
@@ -84,6 +85,15 @@ fn resolve_owner_scope_notification_user(
trimmed_option(explicit_user).or_else(|| trimmed_option(owner_fallback))
}
fn is_single_message_repl(message: &IncomingMessage) -> bool {
message.channel == "repl"
&& message
.metadata
.get("single_message_mode")
.and_then(|value| value.as_bool())
.unwrap_or(false)
}
async fn resolve_channel_notification_user(
extension_manager: Option<&Arc<ExtensionManager>>,
channel: Option<&str>,
@@ -1140,9 +1150,14 @@ impl Agent {
&& let Submission::UserInput { ref content } = submission
&& let Some(engine) = self.routine_engine().await
{
let single_message_repl = is_single_message_repl(message);
// Use post-hook content so that BeforeInbound hooks that rewrite
// input are respected by event trigger matching.
let fired = engine.check_event_triggers(message, content).await;
let fired = if single_message_repl {
engine.check_event_triggers_and_wait(message, content).await
} else {
engine.check_event_triggers(message, content).await
};
if fired > 0 {
tracing::debug!(
channel = %message.channel,
@@ -1150,10 +1165,16 @@ impl Agent {
fired,
"Consumed inbound user message with matching event-triggered routine(s)"
);
return Ok(Some(String::new()));
return if single_message_repl {
Ok(None)
} else {
Ok(Some(String::new()))
};
}
}
let session_for_empty_exit = Arc::clone(&session);
// Process based on submission type
let result = match submission {
Submission::UserInput { content } => {
@@ -1250,6 +1271,28 @@ impl Agent {
command,
message.channel
);
// /reasoning is special-cased here (not in handle_system_command)
// because it needs the session + thread_id to read turn reasoning
// data, which handle_system_command's signature doesn't provide.
if command == "reasoning" {
let result = self
.handle_reasoning_command(&args, &session, thread_id)
.await;
return match result {
SubmissionResult::Response { content } => Ok(Some(content)),
SubmissionResult::Ok { message } => Ok(message),
SubmissionResult::Error { message } => {
Ok(Some(format!("Error: {}", message)))
}
_ => {
if is_single_message_repl(message) {
Ok(None)
} else {
Ok(Some(String::new()))
}
}
};
}
// Authorization checks (including restart channel check) are enforced in handle_system_command
self.handle_system_command(&command, &args, &message.channel)
.await
@@ -1309,7 +1352,26 @@ impl Agent {
Ok(Some(content))
}
}
SubmissionResult::Ok { message } => Ok(message),
SubmissionResult::Ok {
message: output_message,
} => {
let should_exit =
if output_message.as_deref() == Some("") && is_single_message_repl(message) {
let sess = session_for_empty_exit.lock().await;
sess.threads
.get(&thread_id)
.map(|thread| thread.state != ThreadState::AwaitingApproval)
.unwrap_or(true)
} else {
false
};
if should_exit {
Ok(None)
} else {
Ok(output_message)
}
}
SubmissionResult::Error { message } => Ok(Some(format!("Error: {}", message))),
SubmissionResult::Interrupted => Ok(Some("Interrupted.".into())),
SubmissionResult::NeedApproval { .. } => {
@@ -1325,7 +1387,7 @@ impl Agent {
#[cfg(test)]
mod tests {
use super::{
chat_tool_execution_metadata, resolve_routine_notification_user,
chat_tool_execution_metadata, is_single_message_repl, resolve_routine_notification_user,
should_fallback_routine_notification, truncate_for_preview,
};
use crate::channels::IncomingMessage;
@@ -1487,4 +1549,17 @@ mod tests {
assert!(should_fallback_routine_notification(&error)); // safety: test-only assertion
}
#[test]
fn single_message_repl_detection_requires_repl_channel_and_metadata_flag() {
let repl = IncomingMessage::new("repl", "owner-scope", "hello")
.with_metadata(serde_json::json!({ "single_message_mode": true }));
let gateway = IncomingMessage::new("gateway", "owner-scope", "hello")
.with_metadata(serde_json::json!({ "single_message_mode": true }));
let plain_repl = IncomingMessage::new("repl", "owner-scope", "hello");
assert!(is_single_message_repl(&repl)); // safety: test-only assertion
assert!(!is_single_message_repl(&gateway)); // safety: test-only assertion
assert!(!is_single_message_repl(&plain_repl)); // safety: test-only assertion
}
}
+1
View File
@@ -414,6 +414,7 @@ mod tests {
id: "call_1".to_string(),
name: "echo".to_string(),
arguments: serde_json::json!({}),
reasoning: None,
};
let delegate = MockDelegate::new(vec![
tool_calls_output(vec![tool_call]),
+89
View File
@@ -465,6 +465,94 @@ impl Agent {
}
}
/// Handle `/reasoning [N|all]` — show reasoning history for the active thread.
pub(super) async fn handle_reasoning_command(
&self,
args: &[String],
session: &Arc<Mutex<Session>>,
thread_id: Uuid,
) -> SubmissionResult {
// Clone the turn data we need, then drop the session lock.
let turns_snapshot: Vec<(
usize,
Option<String>,
Vec<crate::agent::session::TurnToolCall>,
)>;
{
let sess = session.lock().await;
let thread = match sess.threads.get(&thread_id) {
Some(t) => t,
None => return SubmissionResult::error("No active thread."),
};
if thread.turns.is_empty() {
return SubmissionResult::ok_with_message("No turns yet.");
}
// Parse argument: default=last turn, "all"=all turns, N=specific turn (1-based).
let selected: Vec<&crate::agent::session::Turn> = match args.first().map(|s| s.as_str())
{
Some("all") => thread.turns.iter().collect(),
Some(n) => match n.parse::<usize>() {
Ok(0) => return SubmissionResult::error("Turn numbers start at 1."),
Ok(num) if num > thread.turns.len() => {
return SubmissionResult::error(format!(
"Turn {} does not exist (max: {}).",
num,
thread.turns.len()
));
}
Ok(num) => vec![&thread.turns[num - 1]],
Err(_) => return SubmissionResult::error("Usage: /reasoning [N|all]"),
},
None => {
// Default: last turn that has tool calls
match thread.turns.iter().rev().find(|t| !t.tool_calls.is_empty()) {
Some(t) => vec![t],
None => {
return SubmissionResult::ok_with_message("No turns with tool calls.");
}
}
}
};
turns_snapshot = selected
.into_iter()
.map(|t| (t.turn_number, t.narrative.clone(), t.tool_calls.clone()))
.collect();
}
// Session lock is now dropped — format output without holding it.
let mut output = String::new();
for (turn_number, narrative, tool_calls) in &turns_snapshot {
output.push_str(&format!("--- Turn {} ---\n", turn_number + 1));
if let Some(narrative) = narrative {
output.push_str(&format!("Reasoning: {}\n", narrative));
}
if tool_calls.is_empty() {
output.push_str(" (no tool calls)\n");
} else {
for tc in tool_calls {
let status = if tc.error.is_some() {
"error"
} else if tc.result.is_some() {
"ok"
} else {
"pending"
};
output.push_str(&format!(" {} [{}]", tc.name, status));
if let Some(ref rationale) = tc.rationale {
output.push_str(&format!("{}", rationale));
}
output.push('\n');
}
}
output.push('\n');
}
SubmissionResult::response(output.trim_end())
}
/// Handle system commands that bypass thread-state checks entirely.
pub(super) async fn handle_system_command(
&self,
@@ -480,6 +568,7 @@ impl Agent {
" /version Show version info\n",
" /tools List available tools\n",
" /debug Toggle debug mode\n",
" /reasoning [N|all] Show agent reasoning for turns\n",
" /ping Connectivity check\n",
"\n",
"Jobs:\n",
+85 -6
View File
@@ -63,7 +63,12 @@ impl Agent {
);
let system_prompt = if let Some(ws) = self.workspace() {
match ws
let scoped_workspace = if ws.user_id() == message.user_id {
Arc::clone(ws)
} else {
Arc::new(ws.scoped_to_user(&message.user_id))
};
match scoped_workspace
.system_prompt_for_context_tz(is_group_chat, user_tz)
.await
{
@@ -420,6 +425,19 @@ impl<'a> LoopDelegate for ChatDelegate<'a> {
content: Option<String>,
reason_ctx: &mut ReasoningContext,
) -> Result<Option<LoopOutcome>, Error> {
// Extract and sanitize the narrative before consuming `content`.
let narrative = content
.as_deref()
.filter(|c| !c.trim().is_empty())
.map(|c| {
let sanitized = self
.agent
.safety()
.sanitize_tool_output("agent_narrative", c);
sanitized.content
})
.filter(|c| !c.trim().is_empty());
// Add the assistant message with tool_calls to context.
// OpenAI protocol requires this before tool-result messages.
reason_ctx
@@ -440,6 +458,41 @@ impl<'a> LoopDelegate for ChatDelegate<'a> {
)
.await;
// Build per-tool decisions for the reasoning update.
// Sanitize each rationale through SafetyLayer (parity with JobDelegate).
let decisions: Vec<crate::channels::ToolDecision> = tool_calls
.iter()
.filter_map(|tc| {
tc.reasoning.as_ref().map(|r| {
let sanitized = self
.agent
.safety()
.sanitize_tool_output("tool_rationale", r)
.content;
crate::channels::ToolDecision {
tool_name: tc.name.clone(),
rationale: sanitized,
}
})
})
.collect();
// Emit reasoning update to channels.
if narrative.is_some() || !decisions.is_empty() {
let _ = self
.agent
.channels
.send_status(
&self.message.channel,
StatusUpdate::ReasoningUpdate {
narrative: narrative.clone().unwrap_or_default(),
decisions: decisions.clone(),
},
&self.message.metadata,
)
.await;
}
// Record tool calls in the thread with sensitive params redacted.
{
let mut redacted_args: Vec<serde_json::Value> = Vec::with_capacity(tool_calls.len());
@@ -455,8 +508,23 @@ impl<'a> LoopDelegate for ChatDelegate<'a> {
if let Some(thread) = sess.threads.get_mut(&self.thread_id)
&& let Some(turn) = thread.last_turn_mut()
{
// Set turn-level narrative.
if turn.narrative.is_none() {
turn.narrative = narrative;
}
for (tc, safe_args) in tool_calls.iter().zip(redacted_args) {
turn.record_tool_call(&tc.name, safe_args);
let sanitized_rationale = tc.reasoning.as_ref().map(|r| {
self.agent
.safety()
.sanitize_tool_output("tool_rationale", r)
.content
});
turn.record_tool_call_with_reasoning(
&tc.name,
safe_args,
sanitized_rationale,
Some(tc.id.clone()),
);
}
}
}
@@ -726,7 +794,7 @@ impl<'a> LoopDelegate for ChatDelegate<'a> {
if let Some(thread) = sess.threads.get_mut(&self.thread_id)
&& let Some(turn) = thread.last_turn_mut()
{
turn.record_tool_error(error_msg.clone());
turn.record_tool_error_for(&tc.id, error_msg.clone());
}
}
reason_ctx
@@ -852,16 +920,19 @@ impl<'a> LoopDelegate for ChatDelegate<'a> {
Err(e) => format!("Tool '{}' failed: {}", tc.name, e),
};
// Record sanitized result in thread
// Record sanitized result in thread (identity-based matching).
{
let mut sess = self.session.lock().await;
if let Some(thread) = sess.threads.get_mut(&self.thread_id)
&& let Some(turn) = thread.last_turn_mut()
{
if is_tool_error {
turn.record_tool_error(result_content.clone());
turn.record_tool_error_for(&tc.id, result_content.clone());
} else {
turn.record_tool_result(serde_json::json!(result_content));
turn.record_tool_result_for(
&tc.id,
serde_json::json!(result_content),
);
}
}
}
@@ -1462,11 +1533,13 @@ mod tests {
id: "call_2".to_string(),
name: "http".to_string(),
arguments: serde_json::json!({"url": "https://example.com"}),
reasoning: None,
},
ToolCall {
id: "call_3".to_string(),
name: "echo".to_string(),
arguments: serde_json::json!({"message": "done"}),
reasoning: None,
},
],
user_timezone: None,
@@ -1652,6 +1725,7 @@ mod tests {
id: "call_1".to_string(),
name: "echo".to_string(),
arguments: serde_json::json!({"message": "hi"}),
reasoning: None,
}],
),
ChatMessage::tool_result("call_1", "echo", "hi"),
@@ -1744,11 +1818,13 @@ mod tests {
id: "c1".to_string(),
name: "http".to_string(),
arguments: serde_json::json!({}),
reasoning: None,
},
ToolCall {
id: "c2".to_string(),
name: "echo".to_string(),
arguments: serde_json::json!({}),
reasoning: None,
},
],
),
@@ -1782,6 +1858,7 @@ mod tests {
id: "c1".to_string(),
name: "echo".to_string(),
arguments: serde_json::json!({}),
reasoning: None,
}],
),
ChatMessage::tool_result("c1", "echo", "done"),
@@ -1912,6 +1989,7 @@ mod tests {
id: crate::llm::generate_tool_call_id(0, 0),
name: "echo".to_string(),
arguments: serde_json::json!({"message": "looping"}),
reasoning: None,
}],
input_tokens: 0,
output_tokens: 5,
@@ -2065,6 +2143,7 @@ mod tests {
id: crate::llm::generate_tool_call_id(0, 0),
name: "nonexistent_tool".to_string(),
arguments: serde_json::json!({}),
reasoning: None,
}],
input_tokens: 0,
output_tokens: 5,
+60 -10
View File
@@ -18,6 +18,7 @@ use std::time::Duration;
use chrono::Utc;
use regex::Regex;
use tokio::sync::{RwLock, mpsc};
use tokio::task::JoinHandle;
use uuid::Uuid;
use crate::agent::Scheduler;
@@ -45,6 +46,11 @@ enum EventMatcher {
System { routine: Routine },
}
struct TriggeredRoutine {
routine: Routine,
detail: String,
}
/// Distinguishes why sandbox is unavailable so error messages are accurate.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SandboxReadiness {
@@ -202,6 +208,44 @@ impl RoutineEngine {
/// Check incoming message against event triggers. Returns number of routines fired.
pub async fn check_event_triggers(&self, message: &IncomingMessage, content: &str) -> usize {
let triggered = self.matching_event_triggers(message, content).await;
let fired = triggered.len();
for triggered in triggered {
std::mem::drop(self.spawn_fire(triggered.routine, "event", Some(triggered.detail)));
}
fired
}
/// Fire matching event-triggered routines and wait for them to complete.
///
/// Used by single-message REPL mode so the process does not exit before
/// background event-triggered routines finish.
pub async fn check_event_triggers_and_wait(
&self,
message: &IncomingMessage,
content: &str,
) -> usize {
let triggered = self.matching_event_triggers(message, content).await;
let fired = triggered.len();
let handles: Vec<JoinHandle<()>> = triggered
.into_iter()
.map(|triggered| self.spawn_fire(triggered.routine, "event", Some(triggered.detail)))
.collect();
for handle in handles {
if let Err(e) = handle.await {
tracing::warn!(error = %e, "Event-triggered routine task failed");
}
}
fired
}
async fn matching_event_triggers(
&self,
message: &IncomingMessage,
content: &str,
) -> Vec<TriggeredRoutine> {
let cache = self.event_cache.read().await;
// Early return if there are no message matchers at all.
@@ -209,10 +253,9 @@ impl RoutineEngine {
.iter()
.any(|m| matches!(m, EventMatcher::Message { .. }))
{
return 0;
return Vec::new();
}
let mut fired = 0;
let mut triggered = Vec::new();
// Collect routine IDs for batch query
let routine_ids: Vec<Uuid> = cache
@@ -224,13 +267,13 @@ impl RoutineEngine {
.collect();
if routine_ids.is_empty() {
return 0;
return Vec::new();
}
// Single batch query instead of N queries
let concurrent_counts = match self.batch_concurrent_counts(&routine_ids).await {
Some(counts) => counts,
None => return 0,
None => return Vec::new(),
};
for matcher in cache.iter() {
@@ -285,11 +328,13 @@ impl RoutineEngine {
}
let detail = truncate(content, 200);
self.spawn_fire(routine.clone(), "event", Some(detail));
fired += 1;
triggered.push(TriggeredRoutine {
routine: routine.clone(),
detail,
});
}
fired
triggered
}
/// Emit a structured event to system-event routines.
@@ -845,7 +890,12 @@ impl RoutineEngine {
}
/// Spawn a fire in a background task.
fn spawn_fire(&self, routine: Routine, trigger_type: &str, trigger_detail: Option<String>) {
fn spawn_fire(
&self,
routine: Routine,
trigger_type: &str,
trigger_detail: Option<String>,
) -> JoinHandle<()> {
let run = RoutineRun {
id: Uuid::new_v4(),
routine_id: routine.id,
@@ -882,7 +932,7 @@ impl RoutineEngine {
return;
}
execute_routine(engine, routine, run).await;
});
})
}
fn check_cooldown(&self, routine: &Routine) -> bool {
+192 -1
View File
@@ -449,6 +449,7 @@ impl Thread {
id: call_id.clone(),
name: tc.name.clone(),
arguments: tc.parameters.clone(),
reasoning: None,
})
.collect();
@@ -522,7 +523,12 @@ impl Thread {
&& let Some(ref tcs) = assistant_msg.tool_calls
{
for tc in tcs {
turn.record_tool_call(&tc.name, tc.arguments.clone());
turn.record_tool_call_with_reasoning(
&tc.name,
tc.arguments.clone(),
tc.reasoning.clone(),
Some(tc.id.clone()),
);
}
}
@@ -602,6 +608,10 @@ pub struct Turn {
pub completed_at: Option<DateTime<Utc>>,
/// Error message (if failed).
pub error: Option<String>,
/// Agent's reasoning narrative for this turn.
/// Cleaned via `clean_response` and sanitized through `SafetyLayer` before storage.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub narrative: Option<String>,
/// Transient image content parts for multimodal LLM input.
/// Not serialized — images are only needed for the current LLM call.
/// The text description in `user_input` persists for compaction/context.
@@ -621,6 +631,7 @@ impl Turn {
started_at: Utc::now(),
completed_at: None,
error: None,
narrative: None,
image_content_parts: Vec::new(),
}
}
@@ -656,6 +667,26 @@ impl Turn {
parameters: params,
result: None,
error: None,
rationale: None,
tool_call_id: None,
});
}
/// Record a tool call with reasoning context.
pub fn record_tool_call_with_reasoning(
&mut self,
name: impl Into<String>,
params: serde_json::Value,
rationale: Option<String>,
tool_call_id: Option<String>,
) {
self.tool_calls.push(TurnToolCall {
name: name.into(),
parameters: params,
result: None,
error: None,
rationale,
tool_call_id,
});
}
@@ -672,6 +703,60 @@ impl Turn {
call.error = Some(error.into());
}
}
/// Record a tool result by tool_call_id, with fallback to first pending call.
pub fn record_tool_result_for(&mut self, tool_call_id: &str, result: serde_json::Value) {
if let Some(call) = self
.tool_calls
.iter_mut()
.find(|c| c.tool_call_id.as_deref() == Some(tool_call_id))
{
call.result = Some(result);
} else if let Some(call) = self
.tool_calls
.iter_mut()
.find(|c| c.result.is_none() && c.error.is_none())
{
tracing::debug!(
tool_call_id = %tool_call_id,
fallback_tool = %call.name,
"tool_call_id not found, falling back to first pending call"
);
call.result = Some(result);
} else {
tracing::warn!(
tool_call_id = %tool_call_id,
"Tool result dropped: no matching or pending tool call"
);
}
}
/// Record a tool error by tool_call_id, with fallback to first pending call.
pub fn record_tool_error_for(&mut self, tool_call_id: &str, error: impl Into<String>) {
if let Some(call) = self
.tool_calls
.iter_mut()
.find(|c| c.tool_call_id.as_deref() == Some(tool_call_id))
{
call.error = Some(error.into());
} else if let Some(call) = self
.tool_calls
.iter_mut()
.find(|c| c.result.is_none() && c.error.is_none())
{
tracing::debug!(
tool_call_id = %tool_call_id,
fallback_tool = %call.name,
"tool_call_id not found, falling back to first pending call"
);
call.error = Some(error.into());
} else {
tracing::warn!(
tool_call_id = %tool_call_id,
"Tool error dropped: no matching or pending tool call"
);
}
}
}
/// Record of a tool call made during a turn.
@@ -685,6 +770,12 @@ pub struct TurnToolCall {
pub result: Option<serde_json::Value>,
/// Error from the tool (if failed).
pub error: Option<String>,
/// Agent's reasoning for choosing this tool.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub rationale: Option<String>,
/// The tool_call_id from the LLM, for identity-based result matching.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub tool_call_id: Option<String>,
}
#[cfg(test)]
@@ -1309,6 +1400,7 @@ mod tests {
id: "call_0".to_string(),
name: "search".to_string(),
arguments: serde_json::json!({"q": "test"}),
reasoning: None,
};
let messages = vec![
ChatMessage::user("Find test"),
@@ -1339,6 +1431,7 @@ mod tests {
id: "call_0".to_string(),
name: "http".to_string(),
arguments: serde_json::json!({}),
reasoning: None,
};
let messages = vec![
ChatMessage::user("Fetch URL"),
@@ -1404,11 +1497,13 @@ mod tests {
id: "call_a".to_string(),
name: "search".to_string(),
arguments: serde_json::json!({"q": "data"}),
reasoning: None,
};
let tc2 = ToolCall {
id: "call_b".to_string(),
name: "write".to_string(),
arguments: serde_json::json!({"path": "out.txt"}),
reasoning: None,
};
let messages = vec![
ChatMessage::user("Find and save"),
@@ -1620,4 +1715,100 @@ mod tests {
let merged = thread.drain_pending_messages().unwrap();
assert_eq!(merged, "failed batch\nnew msg");
}
#[test]
fn test_record_tool_result_for_by_id() {
let mut turn = Turn::new(0, "test");
turn.record_tool_call_with_reasoning(
"tool_a",
serde_json::json!({}),
None,
Some("id_a".into()),
);
turn.record_tool_call_with_reasoning(
"tool_b",
serde_json::json!({}),
None,
Some("id_b".into()),
);
// Record result for second tool by ID
turn.record_tool_result_for("id_b", serde_json::json!("result_b"));
assert!(turn.tool_calls[0].result.is_none());
assert_eq!(
turn.tool_calls[1].result.as_ref().unwrap(),
&serde_json::json!("result_b")
);
}
#[test]
fn test_record_tool_error_for_by_id() {
let mut turn = Turn::new(0, "test");
turn.record_tool_call_with_reasoning(
"tool_a",
serde_json::json!({}),
None,
Some("id_a".into()),
);
turn.record_tool_call_with_reasoning(
"tool_b",
serde_json::json!({}),
None,
Some("id_b".into()),
);
turn.record_tool_error_for("id_a", "failed");
assert_eq!(turn.tool_calls[0].error.as_deref(), Some("failed"));
assert!(turn.tool_calls[1].error.is_none());
}
#[test]
fn test_record_tool_result_for_fallback_to_pending() {
let mut turn = Turn::new(0, "test");
turn.record_tool_call_with_reasoning(
"tool_a",
serde_json::json!({}),
None,
Some("id_a".into()),
);
turn.record_tool_call_with_reasoning(
"tool_b",
serde_json::json!({}),
None,
Some("id_b".into()),
);
// First tool already has a result
turn.tool_calls[0].result = Some(serde_json::json!("done"));
// Unknown ID should fall back to first pending (tool_b)
turn.record_tool_result_for("unknown_id", serde_json::json!("fallback"));
assert_eq!(
turn.tool_calls[0].result.as_ref().unwrap(),
&serde_json::json!("done")
);
assert_eq!(
turn.tool_calls[1].result.as_ref().unwrap(),
&serde_json::json!("fallback")
);
}
#[test]
fn test_record_tool_result_for_no_pending_is_noop() {
let mut turn = Turn::new(0, "test");
turn.record_tool_call_with_reasoning(
"tool_a",
serde_json::json!({}),
None,
Some("id_a".into()),
);
turn.tool_calls[0].result = Some(serde_json::json!("done"));
// No pending calls, unknown ID — should be a no-op
turn.record_tool_result_for("unknown_id", serde_json::json!("lost"));
assert_eq!(
turn.tool_calls[0].result.as_ref().unwrap(),
&serde_json::json!("done")
);
}
}
+11
View File
@@ -92,6 +92,17 @@ impl SubmissionParser {
args: vec![],
};
}
if lower == "/reasoning" || lower.starts_with("/reasoning ") {
let args: Vec<String> = trimmed
.split_whitespace()
.skip(1)
.map(|s| s.to_string())
.collect();
return Submission::SystemCommand {
command: "reasoning".to_string(),
args,
};
}
if lower == "/restart" {
tracing::debug!("[SubmissionParser::parse] Recognized /restart command");
return Submission::SystemCommand {
+58 -11
View File
@@ -513,10 +513,10 @@ impl Agent {
};
thread.complete_turn(&response);
let (turn_number, tool_calls) = thread
let (turn_number, tool_calls, narrative) = thread
.turns
.last()
.map(|t| (t.turn_number, t.tool_calls.clone()))
.map(|t| (t.turn_number, t.tool_calls.clone(), t.narrative.clone()))
.unwrap_or_default();
let _ = self
.channels
@@ -534,6 +534,7 @@ impl Agent {
&message.user_id,
turn_number,
&tool_calls,
narrative.as_deref(),
)
.await;
self.persist_assistant_response(
@@ -725,7 +726,9 @@ impl Agent {
///
/// Stored between the user and assistant messages so that
/// `build_turns_from_db_messages` can reconstruct the tool call history.
/// Content is a JSON array of tool call summaries.
/// Content is a JSON object: `{ "calls": [...], "narrative": "..." }`.
/// The `calls` array contains tool call summaries with optional `rationale`
/// and `tool_call_id` fields. Legacy rows may be plain JSON arrays.
pub(super) async fn persist_tool_calls(
&self,
thread_id: Uuid,
@@ -733,6 +736,7 @@ impl Agent {
user_id: &str,
turn_number: usize,
tool_calls: &[crate::agent::session::TurnToolCall],
narrative: Option<&str>,
) {
if tool_calls.is_empty() {
return;
@@ -767,11 +771,30 @@ impl Agent {
if let Some(ref error) = tc.error {
obj["error"] = serde_json::Value::String(truncate_preview(error, 200));
}
if let Some(ref rationale) = tc.rationale {
obj["rationale"] = serde_json::Value::String(truncate_preview(rationale, 500));
}
if let Some(ref tool_call_id) = tc.tool_call_id {
obj["tool_call_id"] =
serde_json::Value::String(truncate_preview(tool_call_id, 128));
}
obj
})
.collect();
let content = match serde_json::to_string(&summaries) {
// Wrap in an object with optional narrative so it can be reconstructed.
// safety: no byte-index slicing here; comment describes JSON shape
let wrapper = if let Some(n) = narrative {
serde_json::json!({
"narrative": truncate_preview(n, 1000),
"calls": summaries,
})
} else {
serde_json::json!({
"calls": summaries,
})
};
let content = match serde_json::to_string(&wrapper) {
Ok(c) => c,
Err(e) => {
tracing::warn!("Failed to serialize tool calls: {}", e);
@@ -1104,9 +1127,12 @@ impl Agent {
&& let Some(turn) = thread.last_turn_mut()
{
if is_tool_error {
turn.record_tool_error(result_content.clone());
turn.record_tool_error_for(&pending.tool_call_id, result_content.clone());
} else {
turn.record_tool_result(serde_json::json!(result_content));
turn.record_tool_result_for(
&pending.tool_call_id,
serde_json::json!(result_content),
);
}
}
}
@@ -1358,9 +1384,12 @@ impl Agent {
&& let Some(turn) = thread.last_turn_mut()
{
if is_deferred_error {
turn.record_tool_error(deferred_content.clone());
turn.record_tool_error_for(&tc.id, deferred_content.clone());
} else {
turn.record_tool_result(serde_json::json!(deferred_content));
turn.record_tool_result_for(
&tc.id,
serde_json::json!(deferred_content),
);
}
}
}
@@ -1459,10 +1488,10 @@ impl Agent {
let (response, suggestions) =
crate::agent::dispatcher::extract_suggestions(&response);
thread.complete_turn(&response);
let (turn_number, tool_calls) = thread
let (turn_number, tool_calls, narrative) = thread
.turns
.last()
.map(|t| (t.turn_number, t.tool_calls.clone()))
.map(|t| (t.turn_number, t.tool_calls.clone(), t.narrative.clone()))
.unwrap_or_default();
// User message already persisted at turn start; save tool calls then assistant response
self.persist_tool_calls(
@@ -1471,6 +1500,7 @@ impl Agent {
&message.user_id,
turn_number,
&tool_calls,
narrative.as_deref(),
)
.await;
self.persist_assistant_response(
@@ -1816,7 +1846,20 @@ fn rebuild_chat_messages_from_db(
"assistant" => result.push(ChatMessage::assistant(&msg.content)),
"tool_calls" => {
// Try to parse the enriched JSON and rebuild tool messages.
if let Ok(calls) = serde_json::from_str::<Vec<serde_json::Value>>(&msg.content) {
// Supports two formats:
// - Old: plain JSON array of tool call summaries
// - New: wrapped object { "calls": [...], "narrative": "..." }
let calls: Vec<serde_json::Value> =
match serde_json::from_str::<serde_json::Value>(&msg.content) {
Ok(serde_json::Value::Array(arr)) => arr,
Ok(serde_json::Value::Object(obj)) => obj
.get("calls")
.and_then(|v| v.as_array())
.cloned()
.unwrap_or_default(),
_ => Vec::new(),
};
{
if calls.is_empty() {
continue;
}
@@ -1839,6 +1882,10 @@ fn rebuild_chat_messages_from_db(
.get("parameters")
.cloned()
.unwrap_or(serde_json::json!({})),
reasoning: c
.get("rationale")
.and_then(|v| v.as_str())
.map(String::from),
})
.collect();
+16
View File
@@ -265,6 +265,15 @@ impl OutgoingResponse {
}
}
/// A single tool decision within a reasoning update.
#[derive(Debug, Clone)]
pub struct ToolDecision {
/// Tool name.
pub tool_name: String,
/// Agent's reasoning for choosing this tool.
pub rationale: String,
}
/// Status update types for showing agent activity.
#[derive(Debug, Clone)]
pub enum StatusUpdate {
@@ -333,6 +342,13 @@ pub enum StatusUpdate {
},
/// Suggested follow-up messages for the user.
Suggestions { suggestions: Vec<String> },
/// Agent reasoning update (why it chose specific tools).
ReasoningUpdate {
/// Human-readable summary of the agent's decision.
narrative: String,
/// Per-tool decisions.
decisions: Vec<ToolDecision>,
},
/// Per-turn token usage and cost summary (shown as subtle metadata).
TurnCost {
input_tokens: u64,
+1 -1
View File
@@ -39,7 +39,7 @@ mod webhook_server;
pub use channel::{
AttachmentKind, Channel, ChannelSecretUpdater, IncomingAttachment, IncomingMessage,
MessageStream, OutgoingResponse, StatusUpdate, routing_target_from_metadata,
MessageStream, OutgoingResponse, StatusUpdate, ToolDecision, routing_target_from_metadata,
};
pub use http::{HttpChannel, HttpChannelState};
pub use manager::ChannelManager;
+65 -9
View File
@@ -75,6 +75,7 @@ const SLASH_COMMANDS: &[&str] = &[
"/suggest",
"/thread",
"/resume",
"/reasoning",
];
/// Rustyline helper for slash-command tab completion.
@@ -430,6 +431,18 @@ impl ReplChannel {
let _ = execute!(stderr, terminal::Clear(terminal::ClearType::FromCursorDown));
}
}
async fn finish_single_message_turn(&self) {
if self.single_message.is_none() {
return;
}
let tx = self.msg_tx.lock().ok().and_then(|mut guard| guard.take());
if let Some(tx) = tx {
let msg = IncomingMessage::new("repl", &self.user_id, "/quit");
let _ = tx.send(msg).await;
}
}
}
impl Default for ReplChannel {
@@ -479,7 +492,9 @@ impl Channel for ReplChannel {
async fn start(&self) -> Result<MessageStream, ChannelError> {
let (tx, rx) = mpsc::channel(32);
// Store tx so send_status can inject approval responses directly
// Approval prompts inject responses back through this sender.
// In single-message mode we keep it until the turn finishes, then
// drop it after enqueuing /quit so the receiver stream can close.
if let Ok(mut guard) = self.msg_tx.lock() {
*guard = Some(tx.clone());
}
@@ -495,11 +510,10 @@ impl Channel for ReplChannel {
// Single message mode: send it and return
if let Some(msg) = single_message {
let incoming = IncomingMessage::new("repl", &user_id, &msg).with_timezone(&sys_tz);
let incoming = IncomingMessage::new("repl", &user_id, &msg)
.with_metadata(serde_json::json!({ "single_message_mode": true }))
.with_timezone(&sys_tz);
let _ = tx.blocking_send(incoming);
// Ensure the agent exits after handling exactly one turn in -m mode,
// even when other channels (gateway/http) are enabled.
let _ = tx.blocking_send(IncomingMessage::new("repl", &user_id, "/quit"));
return;
}
@@ -662,6 +676,7 @@ impl Channel for ReplChannel {
println!();
println!();
self.stdin_locked.store(false, Ordering::Relaxed);
self.finish_single_message_turn().await;
return Ok(());
}
@@ -680,6 +695,7 @@ impl Channel for ReplChannel {
println!();
// Unlock stdin so readline can resume
self.stdin_locked.store(false, Ordering::Relaxed);
self.finish_single_message_turn().await;
Ok(())
}
@@ -779,6 +795,7 @@ impl Channel for ReplChannel {
let msg_tx = Arc::clone(&self.msg_tx);
let user_id = self.user_id.clone();
let lock_flag = Arc::clone(&self.stdin_locked);
let single_message_mode = self.single_message.is_some();
tokio::task::spawn_blocking(move || {
let action = run_approval_selector(allow_always).unwrap_or("n");
// Unlock stdin so readline can resume after approval
@@ -787,7 +804,12 @@ impl Channel for ReplChannel {
return;
};
if let Some(tx) = guard.as_ref() {
let msg = IncomingMessage::new("repl", &user_id, action);
let msg = if single_message_mode {
IncomingMessage::new("repl", &user_id, action)
.with_metadata(serde_json::json!({ "single_message_mode": true }))
} else {
IncomingMessage::new("repl", &user_id, action)
};
let _ = tx.blocking_send(msg);
}
});
@@ -841,6 +863,19 @@ impl Channel for ReplChannel {
StatusUpdate::Suggestions { .. } => {
// Suggestions are only rendered by the web gateway
}
StatusUpdate::ReasoningUpdate {
narrative,
decisions,
} => {
if !narrative.is_empty() {
let display = truncate_for_preview(&narrative, CLI_STATUS_MAX);
eprintln!(" \x1b[94m\u{25B6} {display}\x1b[0m");
}
for d in &decisions {
let display = truncate_for_preview(&d.rationale, CLI_STATUS_MAX);
eprintln!(" \x1b[90m\u{2192} {}: {display}\x1b[0m", d.tool_name);
}
}
StatusUpdate::TurnCost { .. } => {
// Cost display is handled by the TUI channel
}
@@ -875,6 +910,7 @@ impl Channel for ReplChannel {
#[cfg(test)]
mod tests {
use futures::StreamExt;
use tokio::time::{Duration, timeout};
use super::*;
@@ -883,16 +919,36 @@ mod tests {
let repl = ReplChannel::with_message("hi".to_string());
let mut stream = repl.start().await.expect("repl start should succeed");
let first = stream.next().await.expect("first message missing");
let first = timeout(Duration::from_secs(1), stream.next())
.await
.expect("timed out waiting for first message")
.expect("first message missing");
assert_eq!(first.channel, "repl");
assert_eq!(first.content, "hi");
let second = stream.next().await.expect("quit message missing");
assert!(
timeout(Duration::from_millis(100), stream.next())
.await
.is_err(),
"single-message mode should wait for the turn to finish before quitting"
);
repl.respond(&first, OutgoingResponse::text("done"))
.await
.expect("respond should succeed");
let second = timeout(Duration::from_secs(1), stream.next())
.await
.expect("timed out waiting for quit message")
.expect("quit message missing");
assert_eq!(second.channel, "repl");
assert_eq!(second.content, "/quit");
assert!(
stream.next().await.is_none(),
timeout(Duration::from_secs(1), stream.next())
.await
.expect("timed out waiting for stream to close")
.is_none(),
"stream should end after /quit"
);
}
+14
View File
@@ -3061,6 +3061,20 @@ fn status_to_wit(
},
// Suggestions and turn cost are web-gateway-only; skip for WASM channels
StatusUpdate::Suggestions { .. } | StatusUpdate::TurnCost { .. } => return None,
StatusUpdate::ReasoningUpdate {
narrative,
decisions,
} => {
let mut msg = narrative.clone();
for d in decisions {
msg.push_str(&format!("\n{}: {}", d.tool_name, d.rationale));
}
wit_channel::StatusUpdate {
status: wit_channel::StatusType::Status,
message: msg,
metadata_json,
}
}
})
}
+2
View File
@@ -398,8 +398,10 @@ pub async fn chat_history_handler(
truncate_preview(&s, 500)
}),
error: tc.error.clone(),
rationale: tc.rationale.clone(),
})
.collect(),
narrative: t.narrative.clone(),
})
.collect();
+14
View File
@@ -489,6 +489,20 @@ impl Channel for GatewayChannel {
},
StatusUpdate::Suggestions { suggestions } => AppEvent::Suggestions {
suggestions,
thread_id: thread_id.clone(),
},
StatusUpdate::ReasoningUpdate {
narrative,
decisions,
} => AppEvent::ReasoningUpdate {
narrative,
decisions: decisions
.into_iter()
.map(|d| crate::channels::web::types::ToolDecisionDto {
tool_name: d.tool_name,
rationale: d.rationale,
})
.collect(),
thread_id,
},
StatusUpdate::TurnCost {
+2
View File
@@ -231,6 +231,7 @@ pub fn convert_messages(messages: &[OpenAiMessage]) -> Result<Vec<ChatMessage>,
name: tc.function.name.clone(),
arguments: serde_json::from_str(&tc.function.arguments)
.unwrap_or(serde_json::Value::Object(Default::default())),
reasoning: None,
})
.collect();
Ok(ChatMessage::assistant_with_tool_calls(
@@ -954,6 +955,7 @@ mod tests {
id: "call_abc".to_string(),
name: "search".to_string(),
arguments: serde_json::json!({"query": "rust"}),
reasoning: None,
}];
let converted = convert_tool_calls_to_openai(&calls);
+2
View File
@@ -1725,8 +1725,10 @@ async fn chat_history_handler(
truncate_preview(&s, 500)
}),
error: tc.error.clone(),
rationale: tc.rationale.clone(),
})
.collect(),
narrative: t.narrative.clone(),
})
.collect();
+7 -1
View File
@@ -63,6 +63,9 @@ pub struct TurnInfo {
pub started_at: String,
pub completed_at: Option<String>,
pub tool_calls: Vec<ToolCallInfo>,
/// Agent's reasoning narrative for this turn.
#[serde(skip_serializing_if = "Option::is_none")]
pub narrative: Option<String>,
}
#[derive(Debug, Serialize)]
@@ -74,6 +77,9 @@ pub struct ToolCallInfo {
pub result_preview: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
/// Agent's reasoning for choosing this tool.
#[serde(skip_serializing_if = "Option::is_none")]
pub rationale: Option<String>,
}
#[derive(Debug, Serialize)]
@@ -116,7 +122,7 @@ pub struct ApprovalRequest {
// --- App Event (re-exported from ironclaw_common) ---
pub use ironclaw_common::AppEvent;
pub use ironclaw_common::{AppEvent, ToolDecisionDto};
// --- Memory ---
+87 -12
View File
@@ -4,6 +4,21 @@ use crate::channels::web::types::{ToolCallInfo, TurnInfo};
pub use ironclaw_common::truncate_preview;
/// Parse tool call summary JSON objects into `ToolCallInfo` structs.
fn parse_tool_call_infos(calls: &[serde_json::Value]) -> Vec<ToolCallInfo> {
calls
.iter()
.map(|c| ToolCallInfo {
name: c["name"].as_str().unwrap_or("unknown").to_string(),
has_result: c.get("result_preview").is_some_and(|v| !v.is_null()),
has_error: c.get("error").is_some_and(|v| !v.is_null()),
result_preview: c["result_preview"].as_str().map(String::from),
error: c["error"].as_str().map(String::from),
rationale: c["rationale"].as_str().map(String::from),
})
.collect()
}
/// Build TurnInfo pairs from flat DB messages (user/tool_calls/assistant triples).
///
/// Handles three message patterns:
@@ -27,6 +42,7 @@ pub fn build_turns_from_db_messages(
started_at: msg.created_at.to_rfc3339(),
completed_at: None,
tool_calls: Vec::new(),
narrative: None,
};
// Check if next message is a tool_calls record
@@ -34,18 +50,28 @@ pub fn build_turns_from_db_messages(
&& next.role == "tool_calls"
{
let tc_msg = iter.next().expect("peeked");
match serde_json::from_str::<Vec<serde_json::Value>>(&tc_msg.content) {
Ok(calls) => {
turn.tool_calls = calls
.iter()
.map(|c| ToolCallInfo {
name: c["name"].as_str().unwrap_or("unknown").to_string(),
has_result: c.get("result_preview").is_some(),
has_error: c.get("error").is_some(),
result_preview: c["result_preview"].as_str().map(String::from),
error: c["error"].as_str().map(String::from),
})
.collect();
// Parse tool_calls JSON — supports two formats:
// safety: no byte-index slicing; comment describes JSON shape
match serde_json::from_str::<serde_json::Value>(&tc_msg.content) {
Ok(serde_json::Value::Array(calls)) => {
// Old format: plain array
turn.tool_calls = parse_tool_call_infos(&calls);
}
Ok(serde_json::Value::Object(obj)) => {
// New wrapped format with narrative
turn.narrative = obj
.get("narrative")
.and_then(|v| v.as_str())
.map(String::from);
if let Some(serde_json::Value::Array(calls)) = obj.get("calls") {
turn.tool_calls = parse_tool_call_infos(calls);
}
}
Ok(_) => {
tracing::warn!(
message_id = %tc_msg.id,
"Unexpected tool_calls JSON shape in DB, skipping"
);
}
Err(e) => {
tracing::warn!(
@@ -83,6 +109,7 @@ pub fn build_turns_from_db_messages(
started_at: msg.created_at.to_rfc3339(),
completed_at: Some(msg.created_at.to_rfc3339()),
tool_calls: Vec::new(),
narrative: None,
});
turn_number += 1;
}
@@ -201,4 +228,52 @@ mod tests {
assert!(turns[0].tool_calls.is_empty());
assert_eq!(turns[0].state, "Completed");
}
#[test]
fn test_build_turns_with_wrapped_tool_calls_format() {
let tc_json = serde_json::json!({
"narrative": "Searching memory for context before proceeding.",
"calls": [
{"name": "memory_search", "result_preview": "found 3 items", "rationale": "consult prior context"},
{"name": "shell", "error": "permission denied"}
]
});
let messages = vec![
make_msg("user", "Find info", 0),
make_msg("tool_calls", &tc_json.to_string(), 500),
make_msg("assistant", "Here's what I found", 1000),
];
let turns = build_turns_from_db_messages(&messages);
assert_eq!(turns.len(), 1);
assert_eq!(
turns[0].narrative.as_deref(),
Some("Searching memory for context before proceeding.")
);
assert_eq!(turns[0].tool_calls.len(), 2);
assert_eq!(turns[0].tool_calls[0].name, "memory_search");
assert_eq!(
turns[0].tool_calls[0].rationale.as_deref(),
Some("consult prior context")
);
assert!(turns[0].tool_calls[0].has_result);
assert_eq!(turns[0].tool_calls[1].name, "shell");
assert!(turns[0].tool_calls[1].has_error);
assert_eq!(turns[0].response.as_deref(), Some("Here's what I found"));
}
#[test]
fn test_build_turns_wrapped_format_without_narrative() {
let tc_json = serde_json::json!({
"calls": [{"name": "echo", "result_preview": "hello"}]
});
let messages = vec![
make_msg("user", "Say hi", 0),
make_msg("tool_calls", &tc_json.to_string(), 500),
make_msg("assistant", "Done", 1000),
];
let turns = build_turns_from_db_messages(&messages);
assert_eq!(turns.len(), 1);
assert!(turns[0].narrative.is_none());
assert_eq!(turns[0].tool_calls.len(), 1);
}
}
+2
View File
@@ -575,6 +575,7 @@ fn extract_response_content(response: &AnthropicResponse) -> (Option<String>, Ve
id: id.clone(),
name: name.clone(),
arguments: input.clone(),
reasoning: None,
});
}
}
@@ -623,6 +624,7 @@ mod tests {
id: "call_1".to_string(),
name: "search".to_string(),
arguments: serde_json::json!({"q": "test"}),
reasoning: None,
}];
let messages = vec![
ChatMessage::user("Search for test"),
+7
View File
@@ -522,6 +522,7 @@ fn extract_content_blocks(
id: tu.tool_use_id().to_string(),
name: tu.name().to_string(),
arguments: document_to_json(tu.input()),
reasoning: None,
});
}
// Ignore reasoning, citations, images, etc.
@@ -759,11 +760,13 @@ mod tests {
id: "call_1".to_string(),
name: "echo".to_string(),
arguments: serde_json::json!({"text": "hi"}),
reasoning: None,
};
let tc2 = crate::llm::provider::ToolCall {
id: "call_2".to_string(),
name: "time".to_string(),
arguments: serde_json::json!({}),
reasoning: None,
};
let messages = vec![
@@ -802,6 +805,7 @@ mod tests {
id: "call_1".to_string(),
name: "search".to_string(),
arguments: serde_json::json!({"query": "test"}),
reasoning: None,
};
let messages = vec![
@@ -825,6 +829,7 @@ mod tests {
id: "call_1".to_string(),
name: "echo".to_string(),
arguments: serde_json::json!({}),
reasoning: None,
};
let messages = vec![
@@ -989,11 +994,13 @@ mod tests {
id: "call_abc".to_string(),
name: "get_weather".to_string(),
arguments: serde_json::json!({"city": "NYC"}),
reasoning: None,
};
let tc2 = crate::llm::provider::ToolCall {
id: "call_def".to_string(),
name: "get_time".to_string(),
arguments: serde_json::json!({"tz": "EST"}),
reasoning: None,
};
let messages = vec![
+2
View File
@@ -732,6 +732,7 @@ impl LlmProvider for CodexChatGptProvider {
id: tc.call_id,
name: tc.name,
arguments: args,
reasoning: None,
}
})
.collect();
@@ -825,6 +826,7 @@ mod tests {
id: "call_1".to_string(),
name: "search".to_string(),
arguments: json!({"query": "rust"}),
reasoning: None,
};
let msg = ChatMessage::assistant_with_tool_calls(Some("thinking...".into()), vec![tc]);
let items = CodexChatGptProvider::message_to_input_items(&msg);
+1
View File
@@ -1898,6 +1898,7 @@ impl GeminiOauthProvider {
id,
name,
arguments: args,
reasoning: None,
});
}
}
+2
View File
@@ -596,6 +596,7 @@ fn extract_choice_content(choice: &OpenAiChoice) -> (Option<String>, Vec<ToolCal
name: tc.function.name.clone(),
arguments: serde_json::from_str(&tc.function.arguments)
.unwrap_or(serde_json::Value::Object(serde_json::Map::new())),
reasoning: None,
})
.collect()
})
@@ -628,6 +629,7 @@ mod tests {
id: "call_1".to_string(),
name: "search".to_string(),
arguments: serde_json::json!({"q": "test"}),
reasoning: None,
}];
let messages = vec![
ChatMessage::user("Search"),
+7
View File
@@ -587,6 +587,7 @@ impl LlmProvider for NearAiChatProvider {
id: tc.id,
name: tc.function.name,
arguments,
reasoning: None,
}
})
.collect();
@@ -1180,11 +1181,13 @@ mod tests {
id: "call_1".to_string(),
name: "list_issues".to_string(),
arguments: serde_json::json!({"owner": "foo", "repo": "bar"}),
reasoning: None,
},
ToolCall {
id: "call_2".to_string(),
name: "search".to_string(),
arguments: serde_json::json!({"query": "test"}),
reasoning: None,
},
];
@@ -1217,6 +1220,7 @@ mod tests {
id: "call_1".to_string(),
name: "test".to_string(),
arguments: serde_json::json!({"key": "value"}),
reasoning: None,
};
let msg = ChatMessage::assistant_with_tool_calls(None, vec![tc]);
let chat_msg: ChatCompletionMessage = msg.into();
@@ -1460,6 +1464,7 @@ mod tests {
id: tc.id,
name: tc.function.name,
arguments,
reasoning: None,
}
})
.collect();
@@ -1509,6 +1514,7 @@ mod tests {
id: tc.id,
name: tc.function.name,
arguments,
reasoning: None,
}
})
.collect();
@@ -2131,6 +2137,7 @@ mod tests {
id: "call_1".to_string(),
name: "test".to_string(),
arguments: serde_json::json!({}),
reasoning: None,
}],
);
let chat_msg: ChatCompletionMessage = msg.into();
+5
View File
@@ -625,6 +625,7 @@ fn parse_sse_response(body: &str) -> Result<ParsedResponse, LlmError> {
id: state.call_id,
name: state.name,
arguments,
reasoning: None,
});
} else {
// Fallback: extract directly from the item
@@ -650,6 +651,7 @@ fn parse_sse_response(body: &str) -> Result<ParsedResponse, LlmError> {
id: call_id,
name,
arguments,
reasoning: None,
});
}
}
@@ -727,6 +729,7 @@ fn parse_sse_response(body: &str) -> Result<ParsedResponse, LlmError> {
id: state.call_id,
name: state.name,
arguments,
reasoning: None,
});
}
}
@@ -822,11 +825,13 @@ mod tests {
id: "call_1".to_string(),
name: "search".to_string(),
arguments: serde_json::json!({"query": "test"}),
reasoning: None,
},
ToolCall {
id: "call_2".to_string(),
name: "read".to_string(),
arguments: serde_json::json!({"path": "/tmp"}),
reasoning: None,
},
];
let msg =
+8
View File
@@ -231,6 +231,10 @@ pub struct ToolCall {
pub id: String,
pub name: String,
pub arguments: serde_json::Value,
/// Optional reasoning for why this tool was chosen — supplied by the provider
/// or derived from the shared response content as a fallback.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub reasoning: Option<String>,
}
/// Generate a tool-call ID that satisfies all providers.
@@ -637,6 +641,7 @@ mod tests {
id: "call_1".to_string(),
name: "echo".to_string(),
arguments: serde_json::json!({}),
reasoning: None,
};
let mut messages = vec![
ChatMessage::user("hello"),
@@ -680,6 +685,7 @@ mod tests {
id: "call_1".to_string(),
name: "echo".to_string(),
arguments: serde_json::json!({}),
reasoning: None,
};
let mut messages = vec![
ChatMessage::user("test"),
@@ -705,11 +711,13 @@ mod tests {
id: "call_sel_1".to_string(),
name: "search".to_string(),
arguments: serde_json::json!({"q": "test"}),
reasoning: None,
};
let tc2 = ToolCall {
id: "call_sel_2".to_string(),
name: "http".to_string(),
arguments: serde_json::json!({"url": "https://example.com"}),
reasoning: None,
};
let mut messages = vec![
ChatMessage::system("You are a helpful assistant."),
+85 -12
View File
@@ -525,17 +525,35 @@ impl Reasoning {
let response = self.llm.complete_with_tools(request).await?;
let reasoning = response.content.unwrap_or_default();
let shared_reasoning = response
.content
.map(|c| {
let pre_truncated = truncate_at_tool_tags(&c);
clean_response(&pre_truncated)
})
.unwrap_or_default();
let selections: Vec<ToolSelection> = response
.tool_calls
.into_iter()
.map(|tool_call| ToolSelection {
tool_name: tool_call.name,
parameters: tool_call.arguments,
reasoning: reasoning.clone(),
alternatives: vec![],
tool_call_id: tool_call.id,
.map(|tool_call| {
// Prefer per-tool reasoning if the provider supplied it,
// otherwise fall back to the shared response content.
let rationale = tool_call
.reasoning
.map(|r| {
let pre_truncated = truncate_at_tool_tags(&r);
clean_response(&pre_truncated)
})
.filter(|r| !r.trim().is_empty())
.unwrap_or_else(|| shared_reasoning.clone());
ToolSelection {
tool_name: tool_call.name,
parameters: tool_call.arguments,
reasoning: rationale,
alternatives: vec![],
tool_call_id: tool_call.id,
}
})
.collect();
@@ -664,13 +682,36 @@ Respond in JSON format:
// If there were tool calls, return them for execution
if !response.tool_calls.is_empty() {
let narrative = response.content.map(|c| {
let pre_truncated = truncate_at_tool_tags(&c);
clean_response(&pre_truncated)
});
// Populate per-tool reasoning from the shared narrative when the
// provider did not supply per-tool rationale.
let tool_calls: Vec<ToolCall> = response
.tool_calls
.into_iter()
.map(|mut tc| {
if tc.reasoning.as_ref().is_none_or(|r| r.trim().is_empty()) {
tc.reasoning = narrative.as_ref().filter(|n| !n.is_empty()).cloned();
} else {
// Clean provider-supplied per-tool reasoning the same way
// we clean the shared narrative (strip thinking/tool tags).
tc.reasoning = tc
.reasoning
.map(|r| {
let pre_truncated = truncate_at_tool_tags(&r);
clean_response(&pre_truncated)
})
.filter(|r| !r.trim().is_empty());
}
tc
})
.collect();
return Ok(RespondOutput {
result: RespondResult::ToolCalls {
tool_calls: response.tool_calls,
content: response.content.map(|c| {
let pre_truncated = truncate_at_tool_tags(&c);
clean_response(&pre_truncated)
}),
tool_calls,
content: narrative,
},
usage,
});
@@ -1350,6 +1391,7 @@ fn recover_tool_calls_from_content(
),
name: name.to_string(),
arguments,
reasoning: None,
});
continue;
}
@@ -1364,6 +1406,7 @@ fn recover_tool_calls_from_content(
),
name: name.to_string(),
arguments: serde_json::Value::Object(Default::default()),
reasoning: None,
});
}
}
@@ -1401,6 +1444,7 @@ fn recover_tool_calls_from_content(
),
name: name.to_string(),
arguments,
reasoning: None,
});
remaining = &args_start[bracket_end + 1..];
continue;
@@ -1412,6 +1456,7 @@ fn recover_tool_calls_from_content(
id: super::provider::generate_tool_call_id(calls.len(), RECOVERED_TOOL_CALL_SEED),
name: name.to_string(),
arguments: serde_json::Value::Object(Default::default()),
reasoning: None,
});
remaining = after_name;
}
@@ -3145,4 +3190,32 @@ That's my plan."#;
"Text <function_call>{}</function_call> middle "
);
}
/// Verify that reasoning normalization strips thinking tags and tool tags
/// from per-tool reasoning, matching the cleaning applied to shared reasoning.
#[test]
fn test_reasoning_normalization_strips_thinking_tags() {
let raw = "<thinking>Let me consider...</thinking>Search memory for prior context";
let pre_truncated = truncate_at_tool_tags(raw);
let cleaned = clean_response(&pre_truncated);
assert!(!cleaned.contains("<thinking>"));
assert!(cleaned.contains("Search memory"));
}
#[test]
fn test_reasoning_normalization_strips_tool_tags() {
let raw = "Calling search <tool_call>{\"name\": \"search\"}";
let pre_truncated = truncate_at_tool_tags(raw);
let cleaned = clean_response(&pre_truncated);
assert!(!cleaned.contains("<tool_call>"));
assert!(cleaned.contains("Calling search"));
}
#[test]
fn test_reasoning_normalization_empty_after_cleaning() {
let raw = "<thinking>internal only</thinking>";
let pre_truncated = truncate_at_tool_tags(raw);
let cleaned = clean_response(&pre_truncated);
assert!(cleaned.trim().is_empty());
}
}
+7
View File
@@ -490,6 +490,7 @@ fn extract_response(
id: tc.id.clone(),
name: tc.function.name.clone(),
arguments: tc.function.arguments.clone(),
reasoning: None,
});
}
// Reasoning and Image variants are not mapped to IronClaw types
@@ -880,6 +881,7 @@ mod tests {
id: "Xt7mK9pQ2".to_string(),
name: "search".to_string(),
arguments: serde_json::json!({"query": "test"}),
reasoning: None,
};
let msg = ChatMessage::assistant_with_tool_calls(Some("thinking".to_string()), vec![tc]);
let messages = vec![msg];
@@ -997,6 +999,7 @@ mod tests {
id: "".to_string(),
name: "search".to_string(),
arguments: serde_json::json!({"query": "test"}),
reasoning: None,
};
let messages = vec![ChatMessage::assistant_with_tool_calls(None, vec![tc])];
let (_preamble, history) = convert_messages(&messages);
@@ -1028,6 +1031,7 @@ mod tests {
id: " ".to_string(),
name: "search".to_string(),
arguments: serde_json::json!({"query": "test"}),
reasoning: None,
};
let messages = vec![ChatMessage::assistant_with_tool_calls(None, vec![tc])];
let (_preamble, history) = convert_messages(&messages);
@@ -1061,6 +1065,7 @@ mod tests {
id: "".to_string(),
name: "search".to_string(),
arguments: serde_json::json!({"query": "test"}),
reasoning: None,
};
let assistant_msg = ChatMessage::assistant_with_tool_calls(None, vec![tc]);
let tool_result_msg = ChatMessage {
@@ -1380,11 +1385,13 @@ mod tests {
id: "call_a".to_string(),
name: "search".to_string(),
arguments: serde_json::json!({"q": "rust"}),
reasoning: None,
};
let tc2 = IronToolCall {
id: "call_b".to_string(),
name: "fetch".to_string(),
arguments: serde_json::json!({"url": "https://example.com"}),
reasoning: None,
};
let assistant = ChatMessage::assistant_with_tool_calls(None, vec![tc1, tc2]);
let result_a = ChatMessage::tool_result("call_a", "search", "search results");
+15
View File
@@ -14,6 +14,7 @@ use serde::{Deserialize, Serialize};
use tokio::sync::{Mutex, broadcast};
use uuid::Uuid;
use crate::channels::web::types::ToolDecisionDto;
use crate::db::Database;
use crate::llm::{CompletionRequest, LlmProvider, ToolCompletionRequest};
use crate::orchestrator::auth::{TokenStore, worker_auth_middleware};
@@ -344,6 +345,20 @@ async fn job_event_handler(
// gain context/memory tracking capabilities.
fallback_deliverable: payload.data.get("fallback_deliverable").cloned(),
},
"reasoning" => {
let narrative = payload
.data
.get("narrative")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
let decisions = ToolDecisionDto::from_json_array(&payload.data["decisions"]);
AppEvent::JobReasoning {
job_id: job_id_str,
narrative,
decisions,
}
}
_ => AppEvent::JobStatus {
job_id: job_id_str,
message: payload
+15 -1
View File
@@ -915,7 +915,7 @@ fn parse_routine_create_request(
fn build_routine_trigger(trigger: &NormalizedTriggerRequest) -> Trigger {
match trigger {
NormalizedTriggerRequest::Cron { schedule, timezone } => Trigger::Cron {
schedule: schedule.clone(),
schedule: normalize_cron_expression(schedule),
timezone: timezone.clone(),
},
NormalizedTriggerRequest::Manual => Trigger::Manual,
@@ -1836,6 +1836,20 @@ mod tests {
assert_eq!(parsed.cooldown_secs, 30);
}
#[test]
fn build_routine_trigger_normalizes_cron_schedule() {
let trigger = build_routine_trigger(&NormalizedTriggerRequest::Cron {
schedule: "0 0 9 * * MON-FRI".to_string(),
timezone: Some("UTC".to_string()),
});
assert!(matches!(
trigger,
Trigger::Cron { schedule, timezone }
if schedule == "0 0 9 * * MON-FRI *" && timezone.as_deref() == Some("UTC")
));
}
#[test]
fn parses_grouped_message_event_with_tools() {
let params = serde_json::json!({
+67 -1
View File
@@ -18,6 +18,7 @@ use crate::agent::agentic_loop::{
};
use crate::agent::scheduler::WorkerMessage;
use crate::agent::task::TaskOutput;
use crate::channels::web::types::ToolDecisionDto;
use crate::context::{ContextManager, JobState};
use crate::db::Database;
use crate::error::Error;
@@ -200,6 +201,19 @@ impl Worker {
.map(|s| s.to_string()),
fallback_deliverable: data.get("fallback_deliverable").cloned(),
}),
"reasoning" => {
let narrative = data
.get("narrative")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
let decisions = ToolDecisionDto::from_json_array(&data["decisions"]);
Some(AppEvent::JobReasoning {
job_id: job_id_str,
narrative,
decisions,
})
}
_ => None,
};
if let Some(event) = event {
@@ -897,6 +911,11 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
id: selection.tool_call_id.clone(),
name: selection.tool_name.clone(),
arguments: selection.parameters.clone(),
reasoning: if action.reasoning.is_empty() {
None
} else {
Some(action.reasoning.clone())
},
}],
));
@@ -1357,6 +1376,48 @@ impl<'a> LoopDelegate for JobDelegate<'a> {
);
}
// Emit reasoning event if any tool calls carry reasoning.
// Sanitize narrative and per-tool rationale through SafetyLayer
// (parity with ChatDelegate in dispatcher.rs).
let sanitized_narrative = content
.as_deref()
.filter(|c| !c.trim().is_empty())
.map(|c| {
self.worker
.deps
.safety
.sanitize_tool_output("job_narrative", c)
.content
})
.filter(|c| !c.trim().is_empty())
.unwrap_or_default();
let decisions: Vec<serde_json::Value> = tool_calls
.iter()
.filter_map(|tc| {
tc.reasoning.as_ref().map(|r| {
let sanitized = self
.worker
.deps
.safety
.sanitize_tool_output("tool_rationale", r)
.content;
serde_json::json!({
"tool_name": tc.name,
"rationale": sanitized,
})
})
})
.collect();
if !decisions.is_empty() {
self.worker.log_event(
"reasoning",
serde_json::json!({
"narrative": sanitized_narrative,
"decisions": decisions,
}),
);
}
// Add assistant message with tool_calls (OpenAI protocol)
reason_ctx
.messages
@@ -1371,7 +1432,7 @@ impl<'a> LoopDelegate for JobDelegate<'a> {
.map(|tc| ToolSelection {
tool_name: tc.name.clone(),
parameters: tc.arguments.clone(),
reasoning: String::new(),
reasoning: tc.reasoning.clone().unwrap_or_default(),
alternatives: vec![],
tool_call_id: tc.id.clone(),
})
@@ -1424,6 +1485,11 @@ fn selections_to_tool_calls(selections: &[ToolSelection]) -> Vec<ToolCall> {
id: s.tool_call_id.clone(),
name: s.tool_name.clone(),
arguments: s.parameters.clone(),
reasoning: if s.reasoning.is_empty() {
None
} else {
Some(s.reasoning.clone())
},
})
.collect()
}
+55
View File
@@ -149,6 +149,7 @@ fn reject_if_injected(path: &str, content: &str) -> Result<(), WorkspaceError> {
///
/// Allows Workspace to work with either a PostgreSQL `Repository` (the original
/// path) or any `Database` trait implementation (e.g. libSQL backend).
#[derive(Clone)]
enum WorkspaceStorage {
/// PostgreSQL-backed repository (uses connection pool directly).
#[cfg(feature = "postgres")]
@@ -576,6 +577,60 @@ impl Workspace {
self
}
/// Clone the workspace configuration for a different primary user scope.
///
/// This preserves search config, embeddings, shared read scopes, memory
/// layers, and privacy classifier while switching the primary read/write
/// scope to `user_id`.
pub fn scoped_to_user(&self, user_id: impl Into<String>) -> Self {
let user_id = user_id.into();
let mut memory_layers = self.memory_layers.clone();
for layer in &mut memory_layers {
if layer.sensitivity == crate::workspace::layer::LayerSensitivity::Private
&& layer.scope == self.user_id
{
layer.scope = user_id.clone();
}
}
let mut read_user_ids = vec![user_id.clone()];
for scope in &self.read_user_ids {
if scope != &self.user_id && !read_user_ids.contains(scope) {
read_user_ids.push(scope.clone());
}
}
for scope in crate::workspace::layer::MemoryLayer::read_scopes(&memory_layers) {
if !read_user_ids.contains(&scope) {
read_user_ids.push(scope);
}
}
let preserve_flags = user_id == self.user_id;
Self {
user_id,
read_user_ids,
agent_id: self.agent_id,
storage: self.storage.clone(),
embeddings: self.embeddings.clone(),
bootstrap_pending: std::sync::atomic::AtomicBool::new(if preserve_flags {
self.bootstrap_pending
.load(std::sync::atomic::Ordering::Acquire)
} else {
false
}),
bootstrap_completed: std::sync::atomic::AtomicBool::new(if preserve_flags {
self.bootstrap_completed
.load(std::sync::atomic::Ordering::Acquire)
} else {
false
}),
search_defaults: self.search_defaults.clone(),
memory_layers,
privacy_classifier: self.privacy_classifier.clone(),
}
}
/// Get the user ID (primary scope for writes).
pub fn user_id(&self) -> &str {
&self.user_id
+1
View File
@@ -15,6 +15,7 @@ use crate::workspace::document::{MemoryChunk, MemoryDocument, WorkspaceEntry};
use crate::workspace::search::{RankedResult, SearchConfig, SearchResult, fuse_results};
/// Database repository for workspace operations.
#[derive(Clone)]
pub struct Repository {
pool: Pool,
}
@@ -253,6 +253,6 @@ async def test_telegram_hot_activation_transitions_installed_to_active(page):
assert await card.locator(SEL["ext_pairing_label"]).count() == 0
assert captured_setup_payloads == [
{"secrets": {"telegram_bot_token": "123456789:ABCdefGhI"}},
{"secrets": {}},
{"secrets": {"telegram_bot_token": "123456789:ABCdefGhI"}, "fields": {}},
{"secrets": {}, "fields": {}},
]
+3 -2
View File
@@ -587,6 +587,7 @@ mod advanced {
async fn mcp_extension_lifecycle() {
use crate::support::mock_mcp_server::{MockToolResponse, start_mock_mcp_server};
use ironclaw::extensions::{AuthHint, ExtensionKind, ExtensionSource, RegistryEntry};
const TEST_USER_ID: &str = "test-user";
// 1. Start mock MCP server with pre-configured tool responses.
let mock_server = start_mock_mcp_server(vec![
@@ -654,14 +655,14 @@ mod advanced {
ext_mgr
.secrets()
.create(
"default",
TEST_USER_ID,
ironclaw::secrets::CreateSecretParams::new(secret_name, "mock-access-token")
.with_provider("mcp:mock-notion".to_string()),
)
.await
.expect("failed to inject test token");
let activate_result = ext_mgr.activate("mock-notion", "default").await;
let activate_result = ext_mgr.activate("mock-notion", TEST_USER_ID).await;
assert!(
activate_result.is_ok(),
"activation failed: {:?}",
+1 -1
View File
@@ -439,7 +439,7 @@ mod tests {
match &routine.trigger {
Trigger::Cron { schedule, timezone } => {
assert_eq!(schedule, "0 0 9 * * MON-FRI");
assert_eq!(schedule, "0 0 9 * * MON-FRI *");
assert_eq!(timezone.as_deref(), Some("UTC"));
}
other => panic!("expected cron trigger, got {other:?}"),
+3 -1
View File
@@ -12,6 +12,7 @@ mod tests {
use crate::support::test_rig::TestRigBuilder;
use crate::support::trace_llm::LlmTrace;
use ironclaw::workspace::Workspace;
// -----------------------------------------------------------------------
// Test 1: write_chunk_search
@@ -268,6 +269,7 @@ mod tests {
#[tokio::test]
async fn identity_in_system_prompt() {
const TEST_USER_ID: &str = "test-user";
let trace = LlmTrace::from_file(concat!(
env!("CARGO_MANIFEST_DIR"),
"/tests/fixtures/llm_traces/workspace/identity_prompt.json"
@@ -280,7 +282,7 @@ mod tests {
.await;
// Seed an IDENTITY.md so the system prompt has real content to inject.
let ws = rig.workspace().expect("workspace must be available");
let ws = Workspace::new_with_db(TEST_USER_ID, rig.database().clone());
ws.write(
"IDENTITY.md",
"I am TestBot, a helpful testing assistant created for E2E verification.",
+7 -7
View File
@@ -1,10 +1,10 @@
//! Tests proving that multi-tenant system prompts are broken.
//! Regression tests for multi-tenant system prompts.
//!
//! Bug: In multi-tenant mode, the agent loop uses `self.workspace()` which
//! returns a single shared workspace (user_id="default"). Identity files
//! (IDENTITY.md, SOUL.md, USER.md) seeded under per-user IDs ("alice",
//! "bob") are invisible to this workspace, so the system prompt is
//! empty/wrong.
//! The agent must build the conversational system prompt from a workspace
//! scoped to the incoming message's user, not from the shared owner-scope
//! workspace created at startup. Otherwise per-user identity files
//! (IDENTITY.md, SOUL.md, USER.md) become invisible and different users can
//! see the same owner-scoped prompt.
//!
//! These tests:
//! 1. Seed identity files for two users (alice, bob) in the database
@@ -13,7 +13,7 @@
//! correct user's identity
//! 4. Verify user A's identity doesn't leak into user B's prompt
//!
//! All tests are expected to FAIL until the bug is fixed.
//! These tests ensure each user's identity is isolated correctly.
#[cfg(feature = "libsql")]
mod support;
+1
View File
@@ -94,6 +94,7 @@ impl LlmProvider for MockLlmProvider {
id: "call_mock_001".to_string(),
name: tool.name.clone(),
arguments: serde_json::json!({"test": true}),
reasoning: None,
}],
input_tokens: 15,
output_tokens: 8,
+1
View File
@@ -566,6 +566,7 @@ impl LlmProvider for TraceLlm {
id: tc.id,
name: tc.name,
arguments: tc.arguments,
reasoning: None,
})
.collect();
Ok(ToolCompletionResponse {