Compare commits

..
Author SHA1 Message Date
ZakiandClaude Opus 4.6 51cec37514 fix(libsql): standardize timestamp storage to RFC 3339 with UTC offset (#663)
Replace all `datetime('now')` defaults in libsql_migrations.rs with
`strftime('%Y-%m-%dT%H:%M:%fZ', 'now')` so new rows get proper RFC 3339
timestamps (e.g. `2024-01-15T10:30:00.123Z`) instead of naive datetimes
(e.g. `2024-01-15 10:30:00`).

Add tracing::warn! to parse_timestamp() naive fallback paths so legacy
timestamps are still accepted but produce a visible deprecation signal.

Backward compatible: no data migration needed; existing naive timestamps
continue to parse correctly via the multi-format fallback.

Co-Authored-By: Claude Opus 4.6 <[email protected]>
2026-03-07 17:32:18 -08:00
69 changed files with 892 additions and 6330 deletions
-7
View File
@@ -2,16 +2,9 @@
DATABASE_URL=postgres://localhost/ironclaw DATABASE_URL=postgres://localhost/ironclaw
DATABASE_POOL_SIZE=10 DATABASE_POOL_SIZE=10
# Vector store for workspace memory (optional)
# When set to "lancedb", uses LanceDB for semantic search instead of pgvector/libsql
# VECTOR_BACKEND=builtin # default: use database's built-in index (pgvector or libsql_vector_idx); "pgvector" is also accepted as an alias for "builtin"
# VECTOR_BACKEND=lancedb
# LANCEDB_PATH=~/.ironclaw/lancedb # path for LanceDB when VECTOR_BACKEND=lancedb
# LLM Provider # LLM Provider
# LLM_BACKEND=nearai # default # LLM_BACKEND=nearai # default
# Possible values: nearai, ollama, openai_compatible, openai, anthropic, tinfoil # Possible values: nearai, ollama, openai_compatible, openai, anthropic, tinfoil
# LLM_REQUEST_TIMEOUT_SECS=120 # Increase for local LLMs (Ollama, vLLM, LM Studio)
# === Anthropic Direct === # === Anthropic Direct ===
# Two auth modes: # Two auth modes:
-10
View File
@@ -36,11 +36,6 @@ jobs:
uses: dtolnay/rust-toolchain@stable uses: dtolnay/rust-toolchain@stable
with: with:
components: clippy components: clippy
- name: Install protoc (for lancedb)
if: matrix.name == 'all-features'
uses: arduino/setup-protoc@v3
with:
repo-token: ${{ secrets.GITHUB_TOKEN }}
- uses: Swatinem/rust-cache@v2 - uses: Swatinem/rust-cache@v2
with: with:
key: clippy-${{ matrix.name }} key: clippy-${{ matrix.name }}
@@ -67,11 +62,6 @@ jobs:
uses: dtolnay/rust-toolchain@stable uses: dtolnay/rust-toolchain@stable
with: with:
components: clippy components: clippy
- name: Install protoc (for lancedb)
if: matrix.name == 'all-features'
uses: arduino/setup-protoc@v3
with:
repo-token: ${{ secrets.GITHUB_TOKEN }}
- uses: Swatinem/rust-cache@v2 - uses: Swatinem/rust-cache@v2
with: with:
key: clippy-windows-${{ matrix.name }} key: clippy-windows-${{ matrix.name }}
+1 -15
View File
@@ -14,7 +14,7 @@ jobs:
matrix: matrix:
include: include:
- name: all-features - name: all-features
flags: "--features postgres,libsql,lancedb,html-to-markdown" flags: "--features postgres,libsql,html-to-markdown"
- name: default - name: default
flags: "" flags: ""
- name: libsql-only - name: libsql-only
@@ -26,11 +26,6 @@ jobs:
uses: dtolnay/rust-toolchain@stable uses: dtolnay/rust-toolchain@stable
with: with:
targets: wasm32-wasip2 targets: wasm32-wasip2
- name: Install protoc (for lancedb)
if: matrix.name == 'all-features'
uses: arduino/setup-protoc@v3
with:
repo-token: ${{ secrets.GITHUB_TOKEN }}
- uses: Swatinem/rust-cache@v2 - uses: Swatinem/rust-cache@v2
with: with:
key: ${{ matrix.name }} key: ${{ matrix.name }}
@@ -71,11 +66,6 @@ jobs:
uses: actions/checkout@v6 uses: actions/checkout@v6
- name: Install Rust - name: Install Rust
uses: dtolnay/rust-toolchain@stable uses: dtolnay/rust-toolchain@stable
- name: Install protoc (for lancedb)
if: matrix.name == 'all-features'
uses: arduino/setup-protoc@v3
with:
repo-token: ${{ secrets.GITHUB_TOKEN }}
- uses: Swatinem/rust-cache@v2 - uses: Swatinem/rust-cache@v2
with: with:
key: windows-${{ matrix.name }} key: windows-${{ matrix.name }}
@@ -92,10 +82,6 @@ jobs:
uses: dtolnay/rust-toolchain@stable uses: dtolnay/rust-toolchain@stable
with: with:
targets: wasm32-wasip2 targets: wasm32-wasip2
- name: Install protoc (for lancedb)
uses: arduino/setup-protoc@v3
with:
repo-token: ${{ secrets.GITHUB_TOKEN }}
- uses: Swatinem/rust-cache@v2 - uses: Swatinem/rust-cache@v2
with: with:
key: wasm-extensions key: wasm-extensions
+1 -2
View File
@@ -4,9 +4,8 @@
.env.* .env.*
!.env.example !.env.example
# Claude Code worktrees and lock files # Claude Code worktrees
.claude/worktrees/ .claude/worktrees/
.claude/scheduled_tasks.lock
# Sidecar tool data # Sidecar tool data
.sidecar/ .sidecar/
Generated
+36 -3044
View File
File diff suppressed because it is too large Load Diff
-8
View File
@@ -73,8 +73,6 @@ toml = "0.8"
# Core types # Core types
uuid = { version = "1", features = ["v4", "v5", "serde"] } uuid = { version = "1", features = ["v4", "v5", "serde"] }
chrono = { version = "0.4", features = ["serde"] } chrono = { version = "0.4", features = ["serde"] }
chrono-tz = "0.10"
iana-time-zone = "0.1"
rust_decimal = { version = "1", features = ["serde", "serde-with-str", "maths"] } rust_decimal = { version = "1", features = ["serde", "serde-with-str", "maths"] }
rust_decimal_macros = "1" rust_decimal_macros = "1"
@@ -125,11 +123,6 @@ open = "5"
# The postgres feature provides ToSql/FromSql for postgres-types (shared by tokio-postgres) # The postgres feature provides ToSql/FromSql for postgres-types (shared by tokio-postgres)
pgvector = { version = "0.4", features = ["postgres"], optional = true } pgvector = { version = "0.4", features = ["postgres"], optional = true }
# LanceDB vector store (optional alternative to pgvector/libsql for workspace search)
lancedb = { version = "0.26", optional = true }
arrow-array = { version = "57", optional = true }
arrow-schema = { version = "57", optional = true }
# WASM sandbox for untrusted tool execution # WASM sandbox for untrusted tool execution
wasmtime = { version = "28", features = ["component-model"] } wasmtime = { version = "28", features = ["component-model"] }
wasmtime-wasi = "28" # WASI support for component model wasmtime-wasi = "28" # WASI support for component model
@@ -194,7 +187,6 @@ insta = "1.46.3"
[features] [features]
default = ["postgres", "libsql", "html-to-markdown"] default = ["postgres", "libsql", "html-to-markdown"]
lancedb = ["dep:lancedb", "dep:arrow-array", "dep:arrow-schema"]
postgres = [ postgres = [
"dep:deadpool-postgres", "dep:deadpool-postgres",
"dep:tokio-postgres", "dep:tokio-postgres",
+2 -2
View File
@@ -39,7 +39,7 @@ This document tracks feature parity between IronClaw (Rust implementation) and O
| Network modes (loopback/LAN/remote) | ✅ | 🚧 | HTTP only | | Network modes (loopback/LAN/remote) | ✅ | 🚧 | HTTP only |
| OpenAI-compatible HTTP API | ✅ | ✅ | /v1/chat/completions, per-request `model` override | | OpenAI-compatible HTTP API | ✅ | ✅ | /v1/chat/completions, per-request `model` override |
| Canvas hosting | ✅ | ❌ | Agent-driven UI | | Canvas hosting | ✅ | ❌ | Agent-driven UI |
| Gateway lock (PID-based) | ✅ | | `fs4` flock-based, acquired in `main.rs` before agent startup | | Gateway lock (PID-based) | ✅ | | |
| launchd/systemd integration | ✅ | ❌ | | | launchd/systemd integration | ✅ | ❌ | |
| Bonjour/mDNS discovery | ✅ | ❌ | | | Bonjour/mDNS discovery | ✅ | ❌ | |
| Tailscale integration | ✅ | ❌ | | | Tailscale integration | ✅ | ❌ | |
@@ -340,7 +340,7 @@ This document tracks feature parity between IronClaw (Rust implementation) and O
| Gemini embeddings | ✅ | ❌ | | | Gemini embeddings | ✅ | ❌ | |
| Local embeddings | ✅ | ❌ | | | Local embeddings | ✅ | ❌ | |
| SQLite-vec backend | ✅ | ❌ | IronClaw uses PostgreSQL | | SQLite-vec backend | ✅ | ❌ | IronClaw uses PostgreSQL |
| LanceDB backend | ✅ | | VectorStore trait + LanceDbVectorStore (configured via VECTOR_BACKEND=lancedb) | | LanceDB backend | ✅ | | Configurable auto-capture max length |
| QMD backend | ✅ | ❌ | | | QMD backend | ✅ | ❌ | |
| Atomic reindexing | ✅ | ✅ | | | Atomic reindexing | ✅ | ✅ | |
| Embeddings batching | ✅ | ✅ | `embed_batch` on EmbeddingProvider trait | | Embeddings batching | ✅ | ✅ | `embed_batch` on EmbeddingProvider trait |
+1 -1
View File
@@ -3,7 +3,7 @@ services:
postgres: postgres:
image: pgvector/pgvector:pg16 image: pgvector/pgvector:pg16
ports: ports:
- "127.0.0.1:5432:5432" - "5432:5432"
environment: environment:
POSTGRES_DB: ironclaw POSTGRES_DB: ironclaw
POSTGRES_USER: ironclaw POSTGRES_USER: ironclaw
+1 -6
View File
@@ -356,12 +356,6 @@ impl Agent {
if let Some(workspace) = self.workspace() { if let Some(workspace) = self.workspace() {
let mut config = AgentHeartbeatConfig::default() let mut config = AgentHeartbeatConfig::default()
.with_interval(std::time::Duration::from_secs(hb_config.interval_secs)); .with_interval(std::time::Duration::from_secs(hb_config.interval_secs));
config.quiet_hours_start = hb_config.quiet_hours_start;
config.quiet_hours_end = hb_config.quiet_hours_end;
config.timezone = hb_config
.timezone
.clone()
.or_else(|| Some(self.config.default_timezone.clone()));
if let (Some(user), Some(channel)) = if let (Some(user), Some(channel)) =
(&hb_config.notify_user, &hb_config.notify_channel) (&hb_config.notify_user, &hb_config.notify_channel)
{ {
@@ -417,6 +411,7 @@ impl Agent {
hygiene, hygiene,
workspace.clone(), workspace.clone(),
self.cheap_llm().clone(), self.cheap_llm().clone(),
self.safety().clone(),
Some(notify_tx), Some(notify_tx),
self.store().map(Arc::clone), self.store().map(Arc::clone),
)) ))
+7 -48
View File
@@ -345,6 +345,7 @@ impl Agent {
crate::workspace::hygiene::HygieneConfig::default(), crate::workspace::hygiene::HygieneConfig::default(),
workspace.clone(), workspace.clone(),
self.llm().clone(), self.llm().clone(),
self.safety().clone(),
); );
match runner.check_heartbeat().await { match runner.check_heartbeat().await {
@@ -405,7 +406,7 @@ impl Agent {
.with_max_tokens(512) .with_max_tokens(512)
.with_temperature(0.3); .with_temperature(0.3);
let reasoning = Reasoning::new(self.llm().clone()); let reasoning = Reasoning::new(self.llm().clone(), self.safety().clone());
match reasoning.complete(request).await { match reasoning.complete(request).await {
Ok((text, _usage)) => Ok(SubmissionResult::response(format!( Ok((text, _usage)) => Ok(SubmissionResult::response(format!(
"Thread Summary:\n\n{}", "Thread Summary:\n\n{}",
@@ -453,7 +454,7 @@ impl Agent {
.with_max_tokens(512) .with_max_tokens(512)
.with_temperature(0.5); .with_temperature(0.5);
let reasoning = Reasoning::new(self.llm().clone()); let reasoning = Reasoning::new(self.llm().clone(), self.safety().clone());
match reasoning.complete(request).await { match reasoning.complete(request).await {
Ok((text, _usage)) => Ok(SubmissionResult::response(format!( Ok((text, _usage)) => Ok(SubmissionResult::response(format!(
"Suggested Next Steps:\n\n{}", "Suggested Next Steps:\n\n{}",
@@ -662,14 +663,10 @@ impl Agent {
} }
match self.llm().set_model(requested) { match self.llm().set_model(requested) {
Ok(()) => { Ok(()) => Ok(SubmissionResult::response(format!(
// Persist the model choice so it survives restarts. "Switched model to: {}",
self.persist_selected_model(requested).await; requested
Ok(SubmissionResult::response(format!( ))),
"Switched model to: {}",
requested
)))
}
Err(e) => Ok(SubmissionResult::error(format!( Err(e) => Ok(SubmissionResult::error(format!(
"Failed to switch model: {}", "Failed to switch model: {}",
e e
@@ -825,42 +822,4 @@ impl Agent {
_ => Ok(None), _ => Ok(None),
} }
} }
/// Persist the selected model to the settings store (DB and/or TOML config).
///
/// Best-effort: logs warnings on failure but does not propagate errors,
/// since the in-memory model switch already succeeded.
async fn persist_selected_model(&self, model: &str) {
// 1. Persist to DB if available.
if let Some(store) = self.store() {
let value = serde_json::Value::String(model.to_string());
if let Err(e) = store.set_setting("default", "selected_model", &value).await {
tracing::warn!("Failed to persist model to DB: {}", e);
}
}
// 2. Update TOML config file if it exists (sync I/O in spawn_blocking).
let model_owned = model.to_string();
if let Err(e) = tokio::task::spawn_blocking(move || {
let toml_path = crate::settings::Settings::default_toml_path();
match crate::settings::Settings::load_toml(&toml_path) {
Ok(Some(mut settings)) => {
settings.selected_model = Some(model_owned);
if let Err(e) = settings.save_toml(&toml_path) {
tracing::warn!("Failed to persist model to config.toml: {}", e);
}
}
Ok(None) => {
// No config file on disk; nothing to update.
}
Err(e) => {
tracing::warn!("Failed to load config.toml for model persistence: {}", e);
}
}
})
.await
{
tracing::warn!("Model TOML persistence task failed: {}", e);
}
}
} }
+12 -4
View File
@@ -13,6 +13,7 @@ use crate::agent::context_monitor::{CompactionStrategy, ContextBreakdown};
use crate::agent::session::Thread; use crate::agent::session::Thread;
use crate::error::Error; use crate::error::Error;
use crate::llm::{ChatMessage, CompletionRequest, LlmProvider, Reasoning}; use crate::llm::{ChatMessage, CompletionRequest, LlmProvider, Reasoning};
use crate::safety::SafetyLayer;
use crate::workspace::Workspace; use crate::workspace::Workspace;
/// Result of a compaction operation. /// Result of a compaction operation.
@@ -33,12 +34,13 @@ pub struct CompactionResult {
/// Compacts conversation context to stay within limits. /// Compacts conversation context to stay within limits.
pub struct ContextCompactor { pub struct ContextCompactor {
llm: Arc<dyn LlmProvider>, llm: Arc<dyn LlmProvider>,
safety: Arc<SafetyLayer>,
} }
impl ContextCompactor { impl ContextCompactor {
/// Create a new context compactor. /// Create a new context compactor.
pub fn new(llm: Arc<dyn LlmProvider>) -> Self { pub fn new(llm: Arc<dyn LlmProvider>, safety: Arc<SafetyLayer>) -> Self {
Self { llm } Self { llm, safety }
} }
/// Compact a thread's context using the given strategy. /// Compact a thread's context using the given strategy.
@@ -231,7 +233,7 @@ Be brief but capture all important details. Use bullet points."#,
.with_max_tokens(1024) .with_max_tokens(1024)
.with_temperature(0.3); .with_temperature(0.3);
let reasoning = Reasoning::new(self.llm.clone()); let reasoning = Reasoning::new(self.llm.clone(), self.safety.clone());
let (text, _) = reasoning.complete(request).await?; let (text, _) = reasoning.complete(request).await?;
Ok(text) Ok(text)
} }
@@ -344,11 +346,17 @@ mod tests {
// === QA Plan - Compaction strategy tests === // === QA Plan - Compaction strategy tests ===
use crate::agent::context_monitor::CompactionStrategy; use crate::agent::context_monitor::CompactionStrategy;
use crate::config::SafetyConfig;
use crate::safety::SafetyLayer;
use crate::testing::StubLlm; use crate::testing::StubLlm;
/// Helper: build a `ContextCompactor` with the given `StubLlm`. /// Helper: build a `ContextCompactor` with the given `StubLlm`.
fn make_compactor(llm: Arc<StubLlm>) -> ContextCompactor { fn make_compactor(llm: Arc<StubLlm>) -> ContextCompactor {
ContextCompactor::new(llm) let safety = Arc::new(SafetyLayer::new(&SafetyConfig {
max_output_length: 100_000,
injection_check_enabled: false,
}));
ContextCompactor::new(llm, safety)
} }
/// Helper: build a thread with `n` completed turns. /// Helper: build a thread with `n` completed turns.
+12 -20
View File
@@ -50,18 +50,8 @@ impl Agent {
// Load workspace system prompt (identity files: AGENTS.md, SOUL.md, etc.) // Load workspace system prompt (identity files: AGENTS.md, SOUL.md, etc.)
// In group chats, MEMORY.md is excluded to prevent leaking personal context. // In group chats, MEMORY.md is excluded to prevent leaking personal context.
// Resolve the user's timezone
let user_tz = crate::timezone::resolve_timezone(
message.timezone.as_deref(),
None, // user setting lookup can be added later
&self.config.default_timezone,
);
let system_prompt = if let Some(ws) = self.workspace() { let system_prompt = if let Some(ws) = self.workspace() {
match ws match ws.system_prompt_for_context(is_group_chat).await {
.system_prompt_for_context_tz(is_group_chat, user_tz)
.await
{
Ok(prompt) if !prompt.is_empty() => Some(prompt), Ok(prompt) if !prompt.is_empty() => Some(prompt),
Ok(_) => None, Ok(_) => None,
Err(e) => { Err(e) => {
@@ -113,7 +103,7 @@ impl Agent {
None None
}; };
let mut reasoning = Reasoning::new(self.llm().clone()) let mut reasoning = Reasoning::new(self.llm().clone(), self.safety().clone())
.with_channel(message.channel.clone()) .with_channel(message.channel.clone())
.with_model_name(self.llm().active_model_name()) .with_model_name(self.llm().active_model_name())
.with_group_chat(is_group_chat); .with_group_chat(is_group_chat);
@@ -140,7 +130,6 @@ impl Agent {
let mut job_ctx = let mut job_ctx =
JobContext::with_user(&message.user_id, "chat", "Interactive chat session"); JobContext::with_user(&message.user_id, "chat", "Interactive chat session");
job_ctx.http_interceptor = self.deps.http_interceptor.clone(); job_ctx.http_interceptor = self.deps.http_interceptor.clone();
job_ctx.user_timezone = user_tz.name().to_string();
// Build system prompts once for this turn. Two variants: with tools // Build system prompts once for this turn. Two variants: with tools
// (normal iterations) and without (force_text final iteration). // (normal iterations) and without (force_text final iteration).
@@ -796,7 +785,6 @@ impl Agent {
tool_call_id: tc.id.clone(), tool_call_id: tc.id.clone(),
context_messages: context_messages.clone(), context_messages: context_messages.clone(),
deferred_tool_calls: tool_calls[approval_idx + 1..].to_vec(), deferred_tool_calls: tool_calls[approval_idx + 1..].to_vec(),
user_timezone: Some(user_tz.name().to_string()),
}; };
return Ok(AgenticLoopResult::NeedApproval { pending }); return Ok(AgenticLoopResult::NeedApproval { pending });
@@ -1158,7 +1146,6 @@ mod tests {
max_actions_per_hour: None, max_actions_per_hour: None,
max_tool_iterations: 50, max_tool_iterations: 50,
auto_approve_tools: false, auto_approve_tools: false,
default_timezone: "UTC".to_string(),
}, },
deps, deps,
Arc::new(ChannelManager::new()), Arc::new(ChannelManager::new()),
@@ -1261,7 +1248,6 @@ mod tests {
arguments: serde_json::json!({"message": "done"}), arguments: serde_json::json!({"message": "done"}),
}, },
], ],
user_timezone: None,
}; };
let json = serde_json::to_string(&pending).expect("serialize"); let json = serde_json::to_string(&pending).expect("serialize");
@@ -1609,8 +1595,12 @@ mod tests {
use crate::testing::StubLlm; use crate::testing::StubLlm;
let stub = Arc::new(StubLlm::failing_non_transient("ctx-bomb")); let stub = Arc::new(StubLlm::failing_non_transient("ctx-bomb"));
let safety = Arc::new(SafetyLayer::new(&SafetyConfig {
max_output_length: 100_000,
injection_check_enabled: false,
}));
let reasoning = Reasoning::new(stub.clone()); let reasoning = Reasoning::new(stub.clone(), safety);
// Build a fat context with lots of history. // Build a fat context with lots of history.
let messages = vec![ let messages = vec![
@@ -1720,7 +1710,11 @@ mod tests {
use crate::llm::{Reasoning, ReasoningContext, RespondResult, ToolDefinition}; use crate::llm::{Reasoning, ReasoningContext, RespondResult, ToolDefinition};
let provider = Arc::new(AlwaysToolCallProvider); let provider = Arc::new(AlwaysToolCallProvider);
let reasoning = Reasoning::new(provider); let safety = Arc::new(SafetyLayer::new(&SafetyConfig {
max_output_length: 100_000,
injection_check_enabled: false,
}));
let reasoning = Reasoning::new(provider, safety);
let tool_def = ToolDefinition { let tool_def = ToolDefinition {
name: "echo".to_string(), name: "echo".to_string(),
@@ -1906,7 +1900,6 @@ mod tests {
max_actions_per_hour: None, max_actions_per_hour: None,
max_tool_iterations, max_tool_iterations,
auto_approve_tools: true, auto_approve_tools: true,
default_timezone: "UTC".to_string(),
}, },
deps, deps,
Arc::new(ChannelManager::new()), Arc::new(ChannelManager::new()),
@@ -2022,7 +2015,6 @@ mod tests {
max_actions_per_hour: None, max_actions_per_hour: None,
max_tool_iterations: max_iter, max_tool_iterations: max_iter,
auto_approve_tools: true, auto_approve_tools: true,
default_timezone: "UTC".to_string(),
}, },
deps, deps,
Arc::new(ChannelManager::new()), Arc::new(ChannelManager::new()),
+8 -114
View File
@@ -31,6 +31,7 @@ use tokio::sync::mpsc;
use crate::channels::OutgoingResponse; use crate::channels::OutgoingResponse;
use crate::db::Database; use crate::db::Database;
use crate::llm::{ChatMessage, CompletionRequest, LlmProvider, Reasoning}; use crate::llm::{ChatMessage, CompletionRequest, LlmProvider, Reasoning};
use crate::safety::SafetyLayer;
use crate::workspace::Workspace; use crate::workspace::Workspace;
use crate::workspace::hygiene::HygieneConfig; use crate::workspace::hygiene::HygieneConfig;
@@ -47,12 +48,6 @@ pub struct HeartbeatConfig {
pub notify_user_id: Option<String>, pub notify_user_id: Option<String>,
/// Channel to notify on heartbeat findings. /// Channel to notify on heartbeat findings.
pub notify_channel: Option<String>, pub notify_channel: Option<String>,
/// Hour (0-23) when quiet hours start.
pub quiet_hours_start: Option<u32>,
/// Hour (0-23) when quiet hours end.
pub quiet_hours_end: Option<u32>,
/// Timezone for quiet hours evaluation (IANA name).
pub timezone: Option<String>,
} }
impl Default for HeartbeatConfig { impl Default for HeartbeatConfig {
@@ -63,9 +58,6 @@ impl Default for HeartbeatConfig {
max_failures: 3, max_failures: 3,
notify_user_id: None, notify_user_id: None,
notify_channel: None, notify_channel: None,
quiet_hours_start: None,
quiet_hours_end: None,
timezone: None,
} }
} }
} }
@@ -83,26 +75,6 @@ impl HeartbeatConfig {
self self
} }
/// Check whether the current time falls within configured quiet hours.
pub fn is_quiet_hours(&self) -> bool {
use chrono::Timelike;
let (Some(start), Some(end)) = (self.quiet_hours_start, self.quiet_hours_end) else {
return false;
};
let tz = self
.timezone
.as_deref()
.and_then(crate::timezone::parse_timezone)
.unwrap_or(chrono_tz::UTC);
let now_hour = crate::timezone::now_in_tz(tz).hour();
if start <= end {
now_hour >= start && now_hour < end
} else {
// Wraps midnight, e.g. 22..06
now_hour >= start || now_hour < end
}
}
/// Set the notification target. /// Set the notification target.
pub fn with_notify(mut self, user_id: impl Into<String>, channel: impl Into<String>) -> Self { pub fn with_notify(mut self, user_id: impl Into<String>, channel: impl Into<String>) -> Self {
self.notify_user_id = Some(user_id.into()); self.notify_user_id = Some(user_id.into());
@@ -130,6 +102,7 @@ pub struct HeartbeatRunner {
hygiene_config: HygieneConfig, hygiene_config: HygieneConfig,
workspace: Arc<Workspace>, workspace: Arc<Workspace>,
llm: Arc<dyn LlmProvider>, llm: Arc<dyn LlmProvider>,
safety: Arc<SafetyLayer>,
response_tx: Option<mpsc::Sender<OutgoingResponse>>, response_tx: Option<mpsc::Sender<OutgoingResponse>>,
store: Option<Arc<dyn Database>>, store: Option<Arc<dyn Database>>,
consecutive_failures: u32, consecutive_failures: u32,
@@ -142,12 +115,14 @@ impl HeartbeatRunner {
hygiene_config: HygieneConfig, hygiene_config: HygieneConfig,
workspace: Arc<Workspace>, workspace: Arc<Workspace>,
llm: Arc<dyn LlmProvider>, llm: Arc<dyn LlmProvider>,
safety: Arc<SafetyLayer>,
) -> Self { ) -> Self {
Self { Self {
config, config,
hygiene_config, hygiene_config,
workspace, workspace,
llm, llm,
safety,
response_tx: None, response_tx: None,
store: None, store: None,
consecutive_failures: 0, consecutive_failures: 0,
@@ -187,12 +162,6 @@ impl HeartbeatRunner {
loop { loop {
interval.tick().await; interval.tick().await;
// Skip during quiet hours
if self.config.is_quiet_hours() {
tracing::debug!("Heartbeat skipped: quiet hours");
continue;
}
// Run memory hygiene in the background so it never delays the // Run memory hygiene in the background so it never delays the
// heartbeat checklist. Failures are logged inside run_if_due. // heartbeat checklist. Failures are logged inside run_if_due.
let hygiene_workspace = Arc::clone(&self.workspace); let hygiene_workspace = Arc::clone(&self.workspace);
@@ -303,7 +272,7 @@ impl HeartbeatRunner {
.with_max_tokens(max_tokens) .with_max_tokens(max_tokens)
.with_temperature(0.3); .with_temperature(0.3);
let reasoning = Reasoning::new(self.llm.clone()); let reasoning = Reasoning::new(self.llm.clone(), self.safety.clone());
let (content, _usage) = match reasoning.complete(request).await { let (content, _usage) = match reasoning.complete(request).await {
Ok(r) => r, Ok(r) => r,
Err(e) => return HeartbeatResult::Failed(format!("LLM call failed: {}", e)), Err(e) => return HeartbeatResult::Failed(format!("LLM call failed: {}", e)),
@@ -417,10 +386,11 @@ pub fn spawn_heartbeat(
hygiene_config: HygieneConfig, hygiene_config: HygieneConfig,
workspace: Arc<Workspace>, workspace: Arc<Workspace>,
llm: Arc<dyn LlmProvider>, llm: Arc<dyn LlmProvider>,
safety: Arc<SafetyLayer>,
response_tx: Option<mpsc::Sender<OutgoingResponse>>, response_tx: Option<mpsc::Sender<OutgoingResponse>>,
store: Option<Arc<dyn Database>>, store: Option<Arc<dyn Database>>,
) -> tokio::task::JoinHandle<()> { ) -> tokio::task::JoinHandle<()> {
let mut runner = HeartbeatRunner::new(config, hygiene_config, workspace, llm); let mut runner = HeartbeatRunner::new(config, hygiene_config, workspace, llm, safety);
if let Some(tx) = response_tx { if let Some(tx) = response_tx {
runner = runner.with_response_channel(tx); runner = runner.with_response_channel(tx);
} }
@@ -562,83 +532,6 @@ mod tests {
assert!(!is_effectively_empty(content)); assert!(!is_effectively_empty(content));
} }
// ==================== quiet hours ====================
#[test]
fn test_quiet_hours_inside() {
use chrono::{Timelike, Utc};
let now_utc = Utc::now();
let hour = now_utc.hour();
let start = hour;
let end = (hour + 1) % 24;
let config = HeartbeatConfig {
quiet_hours_start: Some(start),
quiet_hours_end: Some(end),
timezone: Some("UTC".to_string()),
..HeartbeatConfig::default()
};
// Current UTC hour is inside [start, end) by construction
assert!(config.is_quiet_hours());
}
#[test]
fn test_quiet_hours_outside() {
use chrono::{Timelike, Utc};
let now_utc = Utc::now();
let hour = now_utc.hour();
let start = (hour + 1) % 24;
let end = (hour + 2) % 24;
let config = HeartbeatConfig {
quiet_hours_start: Some(start),
quiet_hours_end: Some(end),
timezone: Some("UTC".to_string()),
..HeartbeatConfig::default()
};
// Current UTC hour is outside [start, end) by construction
assert!(!config.is_quiet_hours());
}
#[test]
fn test_quiet_hours_wraparound_excludes_now() {
use chrono::{Timelike, Utc};
let now_utc = Utc::now();
let hour = now_utc.hour();
// Window covers all hours except the current one
let start = (hour + 1) % 24;
let end = hour;
let config = HeartbeatConfig {
quiet_hours_start: Some(start),
quiet_hours_end: Some(end),
timezone: Some("UTC".to_string()),
..HeartbeatConfig::default()
};
assert!(!config.is_quiet_hours());
}
#[test]
fn test_quiet_hours_none_configured() {
let config = HeartbeatConfig::default();
assert!(!config.is_quiet_hours());
}
#[test]
fn test_quiet_hours_same_start_end() {
let config = HeartbeatConfig {
quiet_hours_start: Some(10),
quiet_hours_end: Some(10),
timezone: Some("UTC".to_string()),
..HeartbeatConfig::default()
};
// start == end means zero-width window, should be false
assert!(!config.is_quiet_hours());
}
#[test] #[test]
fn test_spawn_heartbeat_accepts_store_param() { fn test_spawn_heartbeat_accepts_store_param() {
// Regression: spawn_heartbeat must accept an optional Database store // Regression: spawn_heartbeat must accept an optional Database store
@@ -650,6 +543,7 @@ mod tests {
HygieneConfig, HygieneConfig,
Arc<crate::workspace::Workspace>, Arc<crate::workspace::Workspace>,
Arc<dyn crate::llm::LlmProvider>, Arc<dyn crate::llm::LlmProvider>,
Arc<crate::safety::SafetyLayer>,
Option<tokio::sync::mpsc::Sender<crate::channels::OutgoingResponse>>, Option<tokio::sync::mpsc::Sender<crate::channels::OutgoingResponse>>,
Option<Arc<dyn crate::db::Database>>, Option<Arc<dyn crate::db::Database>>,
) -> tokio::task::JoinHandle<()> = spawn_heartbeat; ) -> tokio::task::JoinHandle<()> = spawn_heartbeat;
+9 -87
View File
@@ -57,11 +57,7 @@ pub struct Routine {
#[serde(tag = "type", rename_all = "snake_case")] #[serde(tag = "type", rename_all = "snake_case")]
pub enum Trigger { pub enum Trigger {
/// Fire on a cron schedule (e.g. "0 9 * * MON-FRI" or "every 2h"). /// Fire on a cron schedule (e.g. "0 9 * * MON-FRI" or "every 2h").
Cron { Cron { schedule: String },
schedule: String,
#[serde(default)]
timezone: Option<String>,
},
/// Fire when a channel message matches a pattern. /// Fire when a channel message matches a pattern.
Event { Event {
/// Optional channel filter (e.g. "telegram", "slack"). /// Optional channel filter (e.g. "telegram", "slack").
@@ -103,21 +99,7 @@ impl Trigger {
field: "schedule".into(), field: "schedule".into(),
})? })?
.to_string(); .to_string();
let timezone = config Ok(Trigger::Cron { schedule })
.get("timezone")
.and_then(|v| v.as_str())
.and_then(|tz| {
if crate::timezone::parse_timezone(tz).is_some() {
Some(tz.to_string())
} else {
tracing::warn!(
"Ignoring invalid timezone '{}' from DB for cron trigger",
tz
);
None
}
});
Ok(Trigger::Cron { schedule, timezone })
} }
"event" => { "event" => {
let pattern = config let pattern = config
@@ -155,10 +137,7 @@ impl Trigger {
/// Serialize trigger-specific config to JSON for DB storage. /// Serialize trigger-specific config to JSON for DB storage.
pub fn to_config_json(&self) -> serde_json::Value { pub fn to_config_json(&self) -> serde_json::Value {
match self { match self {
Trigger::Cron { schedule, timezone } => serde_json::json!({ Trigger::Cron { schedule } => serde_json::json!({ "schedule": schedule }),
"schedule": schedule,
"timezone": timezone,
}),
Trigger::Event { channel, pattern } => serde_json::json!({ Trigger::Event { channel, pattern } => serde_json::json!({
"pattern": pattern, "pattern": pattern,
"channel": channel, "channel": channel,
@@ -436,25 +415,12 @@ pub fn content_hash(content: &str) -> u64 {
} }
/// Parse a cron expression and compute the next fire time from now. /// Parse a cron expression and compute the next fire time from now.
/// pub fn next_cron_fire(schedule: &str) -> Result<Option<DateTime<Utc>>, RoutineError> {
/// When `timezone` is provided and valid, the schedule is evaluated in that
/// timezone and the result is converted back to UTC. Otherwise UTC is used.
pub fn next_cron_fire(
schedule: &str,
timezone: Option<&str>,
) -> Result<Option<DateTime<Utc>>, RoutineError> {
let cron_schedule = let cron_schedule =
cron::Schedule::from_str(schedule).map_err(|e| RoutineError::InvalidCron { cron::Schedule::from_str(schedule).map_err(|e| RoutineError::InvalidCron {
reason: e.to_string(), reason: e.to_string(),
})?; })?;
if let Some(tz) = timezone.and_then(crate::timezone::parse_timezone) { Ok(cron_schedule.upcoming(Utc).next())
Ok(cron_schedule
.upcoming(tz)
.next()
.map(|dt| dt.with_timezone(&Utc)))
} else {
Ok(cron_schedule.upcoming(Utc).next())
}
} }
#[cfg(test)] #[cfg(test)]
@@ -467,11 +433,10 @@ mod tests {
fn test_trigger_roundtrip() { fn test_trigger_roundtrip() {
let trigger = Trigger::Cron { let trigger = Trigger::Cron {
schedule: "0 9 * * MON-FRI".to_string(), schedule: "0 9 * * MON-FRI".to_string(),
timezone: None,
}; };
let json = trigger.to_config_json(); let json = trigger.to_config_json();
let parsed = Trigger::from_db("cron", json).expect("parse cron"); let parsed = Trigger::from_db("cron", json).expect("parse cron");
assert!(matches!(parsed, Trigger::Cron { schedule, .. } if schedule == "0 9 * * MON-FRI")); assert!(matches!(parsed, Trigger::Cron { schedule } if schedule == "0 9 * * MON-FRI"));
} }
#[test] #[test]
@@ -544,58 +509,16 @@ mod tests {
#[test] #[test]
fn test_next_cron_fire_valid() { fn test_next_cron_fire_valid() {
// Every minute should always have a next fire // Every minute should always have a next fire
let next = next_cron_fire("* * * * * *", None).expect("valid cron"); let next = next_cron_fire("* * * * * *").expect("valid cron");
assert!(next.is_some()); assert!(next.is_some());
} }
#[test] #[test]
fn test_next_cron_fire_invalid() { fn test_next_cron_fire_invalid() {
let result = next_cron_fire("not a cron", None); let result = next_cron_fire("not a cron");
assert!(result.is_err()); assert!(result.is_err());
} }
#[test]
fn test_trigger_cron_timezone_roundtrip() {
let trigger = Trigger::Cron {
schedule: "0 9 * * MON-FRI".to_string(),
timezone: Some("America/New_York".to_string()),
};
let json = trigger.to_config_json();
let parsed = Trigger::from_db("cron", json).expect("parse cron");
assert!(matches!(parsed, Trigger::Cron { schedule, timezone }
if schedule == "0 9 * * MON-FRI"
&& timezone.as_deref() == Some("America/New_York")));
}
#[test]
fn test_trigger_cron_no_timezone_backward_compat() {
let json = serde_json::json!({"schedule": "0 9 * * *"});
let parsed = Trigger::from_db("cron", json).expect("parse cron");
assert!(matches!(parsed, Trigger::Cron { timezone, .. } if timezone.is_none()));
}
#[test]
fn test_trigger_cron_invalid_timezone_coerced_to_none() {
let json = serde_json::json!({"schedule": "0 9 * * *", "timezone": "Fake/Zone"});
let parsed = Trigger::from_db("cron", json).expect("parse cron");
assert!(
matches!(parsed, Trigger::Cron { timezone, .. } if timezone.is_none()),
"invalid timezone should be coerced to None"
);
}
#[test]
fn test_next_cron_fire_with_timezone() {
let next_utc = next_cron_fire("0 0 9 * * * *", None)
.expect("valid cron")
.expect("has next");
let next_est = next_cron_fire("0 0 9 * * * *", Some("America/New_York"))
.expect("valid cron")
.expect("has next");
// EST is UTC-5 (or EDT UTC-4), so the UTC result should differ
assert_ne!(next_utc, next_est, "timezone should shift the fire time");
}
#[test] #[test]
fn test_guardrails_default() { fn test_guardrails_default() {
let g = RoutineGuardrails::default(); let g = RoutineGuardrails::default();
@@ -608,8 +531,7 @@ mod tests {
fn test_trigger_type_tag() { fn test_trigger_type_tag() {
assert_eq!( assert_eq!(
Trigger::Cron { Trigger::Cron {
schedule: String::new(), schedule: String::new()
timezone: None,
} }
.type_tag(), .type_tag(),
"cron" "cron"
+13 -12
View File
@@ -170,7 +170,7 @@ impl RoutineEngine {
continue; continue;
} }
let detail = if let Trigger::Cron { ref schedule, .. } = routine.trigger { let detail = if let Trigger::Cron { ref schedule } = routine.trigger {
Some(schedule.clone()) Some(schedule.clone())
} else { } else {
None None
@@ -380,12 +380,8 @@ async fn execute_routine(ctx: EngineContext, routine: Routine, run: RoutineRun)
// Update routine runtime state // Update routine runtime state
let now = Utc::now(); let now = Utc::now();
let next_fire = if let Trigger::Cron { let next_fire = if let Trigger::Cron { ref schedule } = routine.trigger {
ref schedule, next_cron_fire(schedule).unwrap_or(None)
ref timezone,
} = routine.trigger
{
next_cron_fire(schedule, timezone.as_deref()).unwrap_or(None)
} else { } else {
None None
}; };
@@ -492,13 +488,18 @@ async fn execute_full_job(
reason: "scheduler not available".to_string(), reason: "scheduler not available".to_string(),
})?; })?;
let mut metadata = serde_json::json!({ "max_iterations": max_iterations }); // Set the message tool's default channel/target from the routine's notify config
// Carry the routine's notify config in job metadata so the message tool // so the LLM can send results without triggering cross-channel approval.
// can resolve channel/target per-job without global state mutation. // TODO: This mutates shared global state and can race with concurrent jobs.
// Move notify config into JobContext metadata and apply per-job instead.
if let Some(channel) = &routine.notify.channel { if let Some(channel) = &routine.notify.channel {
metadata["notify_channel"] = serde_json::json!(channel); scheduler
.tools()
.set_message_tool_context(Some(channel.clone()), Some(routine.notify.user.clone()))
.await;
} }
metadata["notify_user"] = serde_json::json!(&routine.notify.user);
let metadata = serde_json::json!({ "max_iterations": max_iterations });
// Build approval context: UnlessAutoApproved tools are auto-approved for routines; // Build approval context: UnlessAutoApproved tools are auto-approved for routines;
// Always tools require explicit listing in tool_permissions. // Always tools require explicit listing in tool_permissions.
-6
View File
@@ -164,10 +164,6 @@ pub struct PendingApproval {
/// executed yet when approval was requested. /// executed yet when approval was requested.
#[serde(default)] #[serde(default)]
pub deferred_tool_calls: Vec<ToolCall>, pub deferred_tool_calls: Vec<ToolCall>,
/// User timezone at the time the approval was requested, so it persists
/// through the approval flow even if the approval message lacks timezone.
#[serde(default)]
pub user_timezone: Option<String>,
} }
/// A conversation thread within a session. /// A conversation thread within a session.
@@ -980,7 +976,6 @@ mod tests {
tool_call_id: "call_123".to_string(), tool_call_id: "call_123".to_string(),
context_messages: vec![ChatMessage::user("do it")], context_messages: vec![ChatMessage::user("do it")],
deferred_tool_calls: vec![], deferred_tool_calls: vec![],
user_timezone: None,
}; };
thread.await_approval(approval); thread.await_approval(approval);
@@ -1006,7 +1001,6 @@ mod tests {
tool_call_id: "call_456".to_string(), tool_call_id: "call_456".to_string(),
context_messages: vec![], context_messages: vec![],
deferred_tool_calls: vec![], deferred_tool_calls: vec![],
user_timezone: None,
}; };
thread.await_approval(approval); thread.await_approval(approval);
+2 -14
View File
@@ -230,7 +230,7 @@ impl Agent {
) )
.await; .await;
let compactor = ContextCompactor::new(self.llm().clone()); let compactor = ContextCompactor::new(self.llm().clone(), self.safety().clone());
if let Err(e) = compactor if let Err(e) = compactor
.compact(thread, strategy, self.workspace().map(|w| w.as_ref())) .compact(thread, strategy, self.workspace().map(|w| w.as_ref()))
.await .await
@@ -627,7 +627,7 @@ impl Agent {
crate::agent::context_monitor::CompactionStrategy::Summarize { keep_recent: 5 }, crate::agent::context_monitor::CompactionStrategy::Summarize { keep_recent: 5 },
); );
let compactor = ContextCompactor::new(self.llm().clone()); let compactor = ContextCompactor::new(self.llm().clone(), self.safety().clone());
match compactor match compactor
.compact(thread, strategy, self.workspace().map(|w| w.as_ref())) .compact(thread, strategy, self.workspace().map(|w| w.as_ref()))
.await .await
@@ -746,16 +746,6 @@ impl Agent {
let mut job_ctx = let mut job_ctx =
JobContext::with_user(&message.user_id, "chat", "Interactive chat session"); JobContext::with_user(&message.user_id, "chat", "Interactive chat session");
job_ctx.http_interceptor = self.deps.http_interceptor.clone(); job_ctx.http_interceptor = self.deps.http_interceptor.clone();
// Prefer a valid timezone from the approval message, fall back to the
// resolved timezone stored when the approval was originally requested.
let tz_candidate = message
.timezone
.as_deref()
.filter(|tz| crate::timezone::parse_timezone(tz).is_some())
.or(pending.user_timezone.as_deref());
if let Some(tz) = tz_candidate {
job_ctx.user_timezone = tz.to_string();
}
let _ = self let _ = self
.channels .channels
@@ -1121,8 +1111,6 @@ impl Agent {
tool_call_id: tc.id.clone(), tool_call_id: tc.id.clone(),
context_messages: context_messages.clone(), context_messages: context_messages.clone(),
deferred_tool_calls: deferred_tool_calls[approval_idx + 1..].to_vec(), deferred_tool_calls: deferred_tool_calls[approval_idx + 1..].to_vec(),
// Carry forward the resolved timezone from the original pending approval
user_timezone: pending.user_timezone.clone(),
}; };
let request_id = new_pending.request_id; let request_id = new_pending.request_id;
+1 -1
View File
@@ -212,7 +212,7 @@ impl Worker {
let job_ctx = self.context_manager().get_context(self.job_id).await?; let job_ctx = self.context_manager().get_context(self.job_id).await?;
// Create reasoning engine // Create reasoning engine
let reasoning = Reasoning::new(self.llm().clone()); let reasoning = Reasoning::new(self.llm().clone(), self.safety().clone());
// Build initial reasoning context (tool definitions refreshed each iteration in execution_loop) // Build initial reasoning context (tool definitions refreshed each iteration in execution_loop)
let mut reason_ctx = ReasoningContext::new().with_job(&job_ctx.description); let mut reason_ctx = ReasoningContext::new().with_job(&job_ctx.description);
+26 -64
View File
@@ -255,18 +255,15 @@ impl AppBuilder {
self.libsql_db.take(); self.libsql_db.take();
} }
// Re-resolve only the LLM config with OS credentials. // Re-resolve config with OS credentials
let store: Option<&(dyn crate::db::SettingsStore + Sync)> = if let Some(ref db) = self.db {
self.db.as_ref().map(|db| db.as_ref() as _); let toml_path = self.toml_path.as_deref();
let toml_path = self.toml_path.as_deref(); if let Ok(refreshed) =
if let Err(e) = self Config::from_db_with_toml(db.as_ref(), "default", toml_path).await
.config {
.re_resolve_llm(store, "default", toml_path) self.config = refreshed;
.await tracing::debug!("LlmConfig re-resolved after OS credential injection");
{ }
tracing::warn!(
"Failed to re-resolve LLM config after OS credential injection: {e}"
);
} }
return Ok(()); return Ok(());
@@ -311,16 +308,18 @@ impl AppBuilder {
// Inject LLM API keys from encrypted storage // Inject LLM API keys from encrypted storage
crate::config::inject_llm_keys_from_secrets(secrets.as_ref(), "default").await; crate::config::inject_llm_keys_from_secrets(secrets.as_ref(), "default").await;
// Re-resolve only the LLM config with newly available keys. // Re-resolve config with newly available keys
let store: Option<&(dyn crate::db::SettingsStore + Sync)> = if let Some(ref db) = self.db {
self.db.as_ref().map(|db| db.as_ref() as _); let toml_path = self.toml_path.as_deref();
let toml_path = self.toml_path.as_deref(); match Config::from_db_with_toml(db.as_ref(), "default", toml_path).await {
if let Err(e) = self Ok(refreshed) => {
.config self.config = refreshed;
.re_resolve_llm(store, "default", toml_path) tracing::debug!("LlmConfig re-resolved after secret injection");
.await }
{ Err(e) => {
tracing::warn!("Failed to re-resolve LLM config after secret injection: {e}"); tracing::warn!("Failed to re-resolve config after secret injection: {}", e);
}
}
} }
} }
@@ -386,53 +385,12 @@ impl AppBuilder {
.embeddings .embeddings
.create_provider(&self.config.llm.nearai.base_url, self.session.clone()); .create_provider(&self.config.llm.nearai.base_url, self.session.clone());
// Create optional external vector store for workspace semantic search
let vector_store: Option<Arc<dyn crate::workspace::VectorStore>> = {
#[cfg(feature = "lancedb")]
{
if self.config.database.vector_backend == crate::config::VectorBackend::LanceDb {
let path = self
.config
.database
.lancedb_path
.clone()
.unwrap_or_else(crate::config::default_lancedb_path);
let dim = embeddings.as_ref().map(|p| p.dimension());
match crate::workspace::LanceDbVectorStore::new(path, dim).await {
Ok(store) => {
tracing::info!("LanceDB vector store connected for workspace search");
Some(Arc::new(store) as Arc<dyn crate::workspace::VectorStore>)
}
Err(e) => {
tracing::warn!("Failed to initialize LanceDB: {}", e);
None
}
}
} else {
None
}
}
#[cfg(not(feature = "lancedb"))]
{
if self.config.database.vector_backend == crate::config::VectorBackend::LanceDb {
tracing::warn!(
"VECTOR_BACKEND=lancedb but 'lancedb' feature not enabled; \
falling back to built-in vector search"
);
}
None
}
};
// Register memory tools if database is available // Register memory tools if database is available
let workspace = if let Some(ref db) = self.db { let workspace = if let Some(ref db) = self.db {
let mut ws = Workspace::new_with_db("default", db.clone()); let mut ws = Workspace::new_with_db("default", db.clone());
if let Some(ref emb) = embeddings { if let Some(ref emb) = embeddings {
ws = ws.with_embeddings(emb.clone()); ws = ws.with_embeddings(emb.clone());
} }
if let Some(ref vs) = vector_store {
ws = ws.with_vector_store(vs.clone());
}
let ws = Arc::new(ws); let ws = Arc::new(ws);
tools.register_memory_tools(Arc::clone(&ws)); tools.register_memory_tools(Arc::clone(&ws));
Some(ws) Some(ws)
@@ -445,7 +403,11 @@ impl AppBuilder {
&& (self.config.agent.allow_local_tools || !self.config.sandbox.enabled) && (self.config.agent.allow_local_tools || !self.config.sandbox.enabled)
{ {
tools tools
.register_builder_tool(llm.clone(), Some(self.config.builder.to_builder_config())) .register_builder_tool(
llm.clone(),
safety.clone(),
Some(self.config.builder.to_builder_config()),
)
.await; .await;
tracing::info!("Builder mode enabled"); tracing::info!("Builder mode enabled");
} }
-251
View File
@@ -414,103 +414,10 @@ pub enum MigrationError {
Io(String), Io(String),
} }
// ── PID Lock ──────────────────────────────────────────────────────────────
/// Path to the PID lock file: `~/.ironclaw/ironclaw.pid`.
pub fn pid_lock_path() -> PathBuf {
ironclaw_base_dir().join("ironclaw.pid")
}
/// A PID-based lock that prevents multiple IronClaw instances from running
/// simultaneously.
///
/// Uses `fs4::try_lock_exclusive()` for atomic locking (no TOCTOU race),
/// then writes the current PID into the locked file for diagnostics.
/// The OS-level lock is held for the lifetime of this struct and
/// automatically released on drop (along with the PID file cleanup).
#[derive(Debug)]
pub struct PidLock {
path: PathBuf,
/// Held open to maintain the OS-level exclusive lock.
_file: std::fs::File,
}
/// Errors from PID lock acquisition.
#[derive(Debug, thiserror::Error)]
pub enum PidLockError {
#[error("Another IronClaw instance is already running (PID {pid})")]
AlreadyRunning { pid: u32 },
#[error("Failed to acquire PID lock: {0}")]
Io(#[from] std::io::Error),
}
impl PidLock {
/// Try to acquire the PID lock.
///
/// Uses an exclusive file lock (`flock`/`LockFileEx`) so that two
/// concurrent processes cannot both acquire the lock — no TOCTOU race.
/// If the lock file exists but the holding process is gone (stale),
/// the lock is reclaimed automatically by the OS.
pub fn acquire() -> Result<Self, PidLockError> {
Self::acquire_at(pid_lock_path())
}
/// Acquire at a specific path (for testing).
fn acquire_at(path: PathBuf) -> Result<Self, PidLockError> {
use fs4::FileExt;
use std::fs::OpenOptions;
use std::io::Write;
// Ensure parent directory exists
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
// Open (or create) the lock file
let mut file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(&path)?;
// Try non-blocking exclusive lock — if another process holds it,
// this fails immediately instead of blocking.
if let Err(e) = file.try_lock_exclusive() {
if e.kind() == std::io::ErrorKind::WouldBlock {
// Lock held by another process — read its PID for the error message
let pid = std::fs::read_to_string(&path)
.ok()
.and_then(|s| s.trim().parse::<u32>().ok())
.unwrap_or(0);
return Err(PidLockError::AlreadyRunning { pid });
}
// Other errors (permissions, unsupported filesystem, etc.)
return Err(PidLockError::Io(e));
}
// We hold the exclusive lock — write our PID
file.set_len(0)?; // truncate
write!(file, "{}", std::process::id())?;
Ok(PidLock { path, _file: file })
}
}
impl Drop for PidLock {
fn drop(&mut self) {
// Remove the PID file; the OS-level lock is released when _file is dropped.
let _ = std::fs::remove_file(&self.path);
}
}
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
use std::process::Command;
use std::sync::Mutex; use std::sync::Mutex;
use std::thread;
use std::time::{Duration, Instant};
use tempfile::tempdir; use tempfile::tempdir;
static ENV_MUTEX: Mutex<()> = Mutex::new(()); static ENV_MUTEX: Mutex<()> = Mutex::new(());
@@ -1079,162 +986,4 @@ INJECTED="pwned"#;
unsafe { std::env::remove_var("IRONCLAW_BASE_DIR") }; unsafe { std::env::remove_var("IRONCLAW_BASE_DIR") };
} }
} }
// ── PID Lock tests ───────────────────────────────────────────────
#[test]
fn test_pid_lock_acquire_and_drop() {
let dir = tempdir().unwrap();
let pid_path = dir.path().join("ironclaw.pid");
// Acquire lock
let lock = PidLock::acquire_at(pid_path.clone()).unwrap();
assert!(pid_path.exists());
// PID file should contain our PID
let contents = std::fs::read_to_string(&pid_path).unwrap();
assert_eq!(contents.trim().parse::<u32>().unwrap(), std::process::id());
// Drop should remove the file
drop(lock);
assert!(!pid_path.exists());
}
#[test]
fn test_pid_lock_rejects_second_acquire() {
let dir = tempdir().unwrap();
let pid_path = dir.path().join("ironclaw.pid");
// First lock succeeds
let _lock1 = PidLock::acquire_at(pid_path.clone()).unwrap();
// Second acquire on same file must fail (exclusive flock held)
let result = PidLock::acquire_at(pid_path.clone());
assert!(result.is_err());
match result.unwrap_err() {
PidLockError::AlreadyRunning { pid } => {
assert_eq!(pid, std::process::id());
}
other => panic!("expected AlreadyRunning, got: {}", other),
}
}
#[test]
fn test_pid_lock_reclaims_after_drop() {
let dir = tempdir().unwrap();
let pid_path = dir.path().join("ironclaw.pid");
// Acquire and release
let lock = PidLock::acquire_at(pid_path.clone()).unwrap();
drop(lock);
// Should succeed — OS lock was released on drop
let lock2 = PidLock::acquire_at(pid_path).unwrap();
drop(lock2);
}
#[test]
fn test_pid_lock_reclaims_stale_file_without_flock() {
let dir = tempdir().unwrap();
let pid_path = dir.path().join("ironclaw.pid");
// Write a stale PID file manually (no flock held)
std::fs::write(&pid_path, "4294967294").unwrap();
// Should succeed because no OS lock is held on the file
let lock = PidLock::acquire_at(pid_path.clone()).unwrap();
let contents = std::fs::read_to_string(&pid_path).unwrap();
assert_eq!(contents.trim().parse::<u32>().unwrap(), std::process::id());
drop(lock);
}
#[test]
fn test_pid_lock_handles_corrupt_pid_file() {
let dir = tempdir().unwrap();
let pid_path = dir.path().join("ironclaw.pid");
// Write garbage (no flock held)
std::fs::write(&pid_path, "not-a-number").unwrap();
// Should succeed — no OS lock held, file is reclaimed
let lock = PidLock::acquire_at(pid_path).unwrap();
drop(lock);
}
#[test]
fn test_pid_lock_creates_parent_dirs() {
let dir = tempdir().unwrap();
let pid_path = dir.path().join("nested").join("deep").join("ironclaw.pid");
let lock = PidLock::acquire_at(pid_path.clone()).unwrap();
assert!(pid_path.exists());
drop(lock);
}
#[test]
fn test_pid_lock_child_helper_holds_lock() {
if std::env::var("IRONCLAW_PID_LOCK_CHILD").ok().as_deref() != Some("1") {
return;
}
let pid_path = PathBuf::from(
std::env::var("IRONCLAW_PID_LOCK_PATH").expect("IRONCLAW_PID_LOCK_PATH missing"),
);
let hold_ms = std::env::var("IRONCLAW_PID_LOCK_HOLD_MS")
.ok()
.and_then(|s| s.parse::<u64>().ok())
.unwrap_or(3000);
let _lock = PidLock::acquire_at(pid_path).expect("child failed to acquire pid lock");
thread::sleep(Duration::from_millis(hold_ms));
}
#[test]
fn test_pid_lock_rejects_lock_held_by_other_process() {
let dir = tempdir().unwrap();
let pid_path = dir.path().join("ironclaw.pid");
let current_exe = std::env::current_exe().unwrap();
let mut child = Command::new(current_exe)
.args([
"--exact",
"bootstrap::tests::test_pid_lock_child_helper_holds_lock",
"--nocapture",
"--test-threads=1",
])
.env("IRONCLAW_PID_LOCK_CHILD", "1")
.env("IRONCLAW_PID_LOCK_PATH", pid_path.display().to_string())
.env("IRONCLAW_PID_LOCK_HOLD_MS", "3000")
.spawn()
.unwrap();
let started = Instant::now();
while started.elapsed() < Duration::from_secs(2) {
if pid_path.exists() {
break;
}
if let Some(status) = child.try_wait().unwrap() {
panic!("child exited before acquiring lock: {}", status);
}
thread::sleep(Duration::from_millis(20));
}
assert!(
pid_path.exists(),
"child did not create lock file in time: {}",
pid_path.display()
);
let result = PidLock::acquire_at(pid_path.clone());
match result.unwrap_err() {
PidLockError::AlreadyRunning { .. } => {}
other => panic!("expected AlreadyRunning, got: {}", other),
}
let status = child.wait().unwrap();
assert!(status.success(), "child process failed: {}", status);
// After the child exits, lock should be released and reacquirable.
let lock = PidLock::acquire_at(pid_path).unwrap();
drop(lock);
}
} }
-15
View File
@@ -79,8 +79,6 @@ pub struct IncomingMessage {
pub received_at: DateTime<Utc>, pub received_at: DateTime<Utc>,
/// Channel-specific metadata. /// Channel-specific metadata.
pub metadata: serde_json::Value, pub metadata: serde_json::Value,
/// IANA timezone string from the client (e.g. "America/New_York").
pub timezone: Option<String>,
/// File or media attachments on this message. /// File or media attachments on this message.
pub attachments: Vec<IncomingAttachment>, pub attachments: Vec<IncomingAttachment>,
} }
@@ -101,7 +99,6 @@ impl IncomingMessage {
thread_id: None, thread_id: None,
received_at: Utc::now(), received_at: Utc::now(),
metadata: serde_json::Value::Null, metadata: serde_json::Value::Null,
timezone: None,
attachments: Vec::new(), attachments: Vec::new(),
} }
} }
@@ -124,12 +121,6 @@ impl IncomingMessage {
self self
} }
/// Set the client timezone.
pub fn with_timezone(mut self, tz: impl Into<String>) -> Self {
self.timezone = Some(tz.into());
self
}
/// Set attachments. /// Set attachments.
pub fn with_attachments(mut self, attachments: Vec<IncomingAttachment>) -> Self { pub fn with_attachments(mut self, attachments: Vec<IncomingAttachment>) -> Self {
self.attachments = attachments; self.attachments = attachments;
@@ -463,10 +454,4 @@ mod tests {
panic!("expected ToolCompleted variant"); panic!("expected ToolCompleted variant");
} }
} }
#[test]
fn test_incoming_message_with_timezone() {
let msg = IncomingMessage::new("test", "user1", "hello").with_timezone("America/New_York");
assert_eq!(msg.timezone.as_deref(), Some("America/New_York"));
}
} }
+9 -50
View File
@@ -18,7 +18,7 @@
//! - `Esc` - Interrupt current operation //! - `Esc` - Interrupt current operation
use std::borrow::Cow; use std::borrow::Cow;
use std::io::{self, IsTerminal, Write}; use std::io::{self, Write};
use std::sync::Arc; use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::atomic::{AtomicBool, Ordering};
@@ -297,15 +297,10 @@ impl Channel for ReplChannel {
let esc_interrupt_triggered_for_thread = Arc::new(AtomicBool::new(false)); let esc_interrupt_triggered_for_thread = Arc::new(AtomicBool::new(false));
std::thread::spawn(move || { std::thread::spawn(move || {
let sys_tz = crate::timezone::detect_system_timezone().name().to_string();
// Single message mode: send it and return // Single message mode: send it and return
if let Some(msg) = single_message { if let Some(msg) = single_message {
let incoming = IncomingMessage::new("repl", "default", &msg).with_timezone(&sys_tz); let incoming = IncomingMessage::new("repl", "default", &msg);
let _ = tx.blocking_send(incoming); 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", "default", "/quit"));
return; return;
} }
@@ -366,8 +361,7 @@ impl Channel for ReplChannel {
"/quit" | "/exit" => { "/quit" | "/exit" => {
// Forward shutdown command so the agent loop exits even // Forward shutdown command so the agent loop exits even
// when other channels (e.g. web gateway) are still active. // when other channels (e.g. web gateway) are still active.
let msg = IncomingMessage::new("repl", "default", "/quit") let msg = IncomingMessage::new("repl", "default", "/quit");
.with_timezone(&sys_tz);
let _ = tx.blocking_send(msg); let _ = tx.blocking_send(msg);
break; break;
} }
@@ -388,8 +382,7 @@ impl Channel for ReplChannel {
_ => {} _ => {}
} }
let msg = let msg = IncomingMessage::new("repl", "default", line);
IncomingMessage::new("repl", "default", line).with_timezone(&sys_tz);
if tx.blocking_send(msg).is_err() { if tx.blocking_send(msg).is_err() {
break; break;
} }
@@ -397,29 +390,21 @@ impl Channel for ReplChannel {
Err(ReadlineError::Interrupted) => { Err(ReadlineError::Interrupted) => {
if esc_interrupt_triggered_for_thread.swap(false, Ordering::Relaxed) { if esc_interrupt_triggered_for_thread.swap(false, Ordering::Relaxed) {
// Esc: interrupt current operation and keep REPL open. // Esc: interrupt current operation and keep REPL open.
let msg = IncomingMessage::new("repl", "default", "/interrupt") let msg = IncomingMessage::new("repl", "default", "/interrupt");
.with_timezone(&sys_tz);
if tx.blocking_send(msg).is_err() { if tx.blocking_send(msg).is_err() {
break; break;
} }
} else { } else {
// Ctrl+C (VINTR): request graceful shutdown. // Ctrl+C (VINTR): request graceful shutdown.
let msg = IncomingMessage::new("repl", "default", "/quit") let msg = IncomingMessage::new("repl", "default", "/quit");
.with_timezone(&sys_tz);
let _ = tx.blocking_send(msg); let _ = tx.blocking_send(msg);
break; break;
} }
} }
Err(ReadlineError::Eof) => { Err(ReadlineError::Eof) => {
// Ctrl+D in interactive mode: graceful shutdown. // Ctrl+D: send /quit so the agent loop runs graceful shutdown
// In daemon mode (stdin = /dev/null, no TTY), EOF arrives let msg = IncomingMessage::new("repl", "default", "/quit");
// immediately — just drop the REPL thread silently so other let _ = tx.blocking_send(msg);
// channels (gateway, telegram, …) keep running.
if std::io::stdin().is_terminal() {
let msg = IncomingMessage::new("repl", "default", "/quit")
.with_timezone(&sys_tz);
let _ = tx.blocking_send(msg);
}
break; break;
} }
Err(e) => { Err(e) => {
@@ -629,29 +614,3 @@ impl Channel for ReplChannel {
Ok(()) Ok(())
} }
} }
#[cfg(test)]
mod tests {
use futures::StreamExt;
use super::*;
#[tokio::test]
async fn single_message_mode_sends_message_then_quit() {
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");
assert_eq!(first.channel, "repl");
assert_eq!(first.content, "hi");
let second = stream.next().await.expect("quit message missing");
assert_eq!(second.channel, "repl");
assert_eq!(second.content, "/quit");
assert!(
stream.next().await.is_none(),
"stream should end after /quit"
);
}
}
+1 -1
View File
@@ -264,7 +264,7 @@ pub async fn routines_runs_handler(
/// Convert a Routine to the trimmed RoutineInfo for list display. /// Convert a Routine to the trimmed RoutineInfo for list display.
fn routine_to_info(r: &crate::agent::routine::Routine) -> RoutineInfo { fn routine_to_info(r: &crate::agent::routine::Routine) -> RoutineInfo {
let (trigger_type, trigger_summary) = match &r.trigger { let (trigger_type, trigger_summary) = match &r.trigger {
crate::agent::routine::Trigger::Cron { schedule, .. } => { crate::agent::routine::Trigger::Cron { schedule } => {
("cron".to_string(), format!("cron: {}", schedule)) ("cron".to_string(), format!("cron: {}", schedule))
} }
crate::agent::routine::Trigger::Event { crate::agent::routine::Trigger::Event {
+1 -10
View File
@@ -610,7 +610,6 @@ async fn oauth_callback_handler(
async fn chat_send_handler( async fn chat_send_handler(
State(state): State<Arc<GatewayState>>, State(state): State<Arc<GatewayState>>,
headers: axum::http::HeaderMap,
Json(req): Json<SendMessageRequest>, Json(req): Json<SendMessageRequest>,
) -> Result<(StatusCode, Json<SendMessageResponse>), (StatusCode, String)> { ) -> Result<(StatusCode, Json<SendMessageResponse>), (StatusCode, String)> {
tracing::debug!( tracing::debug!(
@@ -627,14 +626,6 @@ async fn chat_send_handler(
} }
let mut msg = IncomingMessage::new("gateway", &state.user_id, &req.content); let mut msg = IncomingMessage::new("gateway", &state.user_id, &req.content);
// Prefer timezone from JSON body, fall back to X-Timezone header
let tz = req
.timezone
.as_deref()
.or_else(|| headers.get("X-Timezone").and_then(|v| v.to_str().ok()));
if let Some(tz) = tz {
msg = msg.with_timezone(tz);
}
if let Some(ref thread_id) = req.thread_id { if let Some(ref thread_id) = req.thread_id {
msg = msg.with_thread(thread_id); msg = msg.with_thread(thread_id);
@@ -2124,7 +2115,7 @@ async fn routines_runs_handler(
/// Convert a Routine to the trimmed RoutineInfo for list display. /// Convert a Routine to the trimmed RoutineInfo for list display.
fn routine_to_info(r: &crate::agent::routine::Routine) -> RoutineInfo { fn routine_to_info(r: &crate::agent::routine::Routine) -> RoutineInfo {
let (trigger_type, trigger_summary) = match &r.trigger { let (trigger_type, trigger_summary) = match &r.trigger {
crate::agent::routine::Trigger::Cron { schedule, .. } => { crate::agent::routine::Trigger::Cron { schedule } => {
("cron".to_string(), format!("cron: {}", schedule)) ("cron".to_string(), format!("cron: {}", schedule))
} }
crate::agent::routine::Trigger::Event { crate::agent::routine::Trigger::Event {
+2 -10
View File
@@ -181,7 +181,6 @@ function confirmRestart() {
body: { body: {
content: '/restart', content: '/restart',
thread_id: currentThreadId, thread_id: currentThreadId,
timezone: Intl.DateTimeFormat().resolvedOptions().timeZone,
}, },
}) })
.then((response) => { .then((response) => {
@@ -455,7 +454,7 @@ function sendMessage() {
apiFetch('/api/chat/send', { apiFetch('/api/chat/send', {
method: 'POST', method: 'POST',
body: { content, thread_id: currentThreadId || undefined, timezone: Intl.DateTimeFormat().resolvedOptions().timeZone }, body: { content, thread_id: currentThreadId || undefined },
}).catch((err) => { }).catch((err) => {
addMessage('system', 'Failed to send: ' + err.message); addMessage('system', 'Failed to send: ' + err.message);
}); });
@@ -564,13 +563,6 @@ function sendApprovalAction(requestId, action) {
function renderMarkdown(text) { function renderMarkdown(text) {
if (typeof marked !== 'undefined') { if (typeof marked !== 'undefined') {
// Escape raw HTML error pages instead of rendering them as markup.
// Only triggers when the text *starts with* a doctype or <html> tag
// (after optional whitespace), so normal messages that mention HTML
// tags in prose or code fences are not affected. See #263.
if (/^\s*<!doctype\s/i.test(text) || /^\s*<html[\s>]/i.test(text)) {
return escapeHtml(text);
}
let html = marked.parse(text); let html = marked.parse(text);
// Sanitize HTML output to prevent XSS from tool output or LLM responses. // Sanitize HTML output to prevent XSS from tool output or LLM responses.
html = sanitizeRenderedHtml(html); html = sanitizeRenderedHtml(html);
@@ -1481,7 +1473,7 @@ chatInput.addEventListener('keydown', (e) => {
} }
} }
if (e.key === 'Enter' && !e.shiftKey && !e.isComposing) { if (e.key === 'Enter' && !e.shiftKey) {
e.preventDefault(); e.preventDefault();
hideSlashAutocomplete(); hideSlashAutocomplete();
sendMessage(); sendMessage();
+2 -8
View File
@@ -9,7 +9,6 @@ use uuid::Uuid;
pub struct SendMessageRequest { pub struct SendMessageRequest {
pub content: String, pub content: String,
pub thread_id: Option<String>, pub thread_id: Option<String>,
pub timezone: Option<String>,
} }
#[derive(Debug, Serialize)] #[derive(Debug, Serialize)]
@@ -614,7 +613,6 @@ pub enum WsClientMessage {
Message { Message {
content: String, content: String,
thread_id: Option<String>, thread_id: Option<String>,
timezone: Option<String>,
}, },
/// Approve or deny a pending tool execution. /// Approve or deny a pending tool execution.
#[serde(rename = "approval")] #[serde(rename = "approval")]
@@ -800,9 +798,7 @@ mod tests {
let json = r#"{"type":"message","content":"hello","thread_id":"t1"}"#; let json = r#"{"type":"message","content":"hello","thread_id":"t1"}"#;
let msg: WsClientMessage = serde_json::from_str(json).unwrap(); let msg: WsClientMessage = serde_json::from_str(json).unwrap();
match msg { match msg {
WsClientMessage::Message { WsClientMessage::Message { content, thread_id } => {
content, thread_id, ..
} => {
assert_eq!(content, "hello"); assert_eq!(content, "hello");
assert_eq!(thread_id.as_deref(), Some("t1")); assert_eq!(thread_id.as_deref(), Some("t1"));
} }
@@ -815,9 +811,7 @@ mod tests {
let json = r#"{"type":"message","content":"hi"}"#; let json = r#"{"type":"message","content":"hi"}"#;
let msg: WsClientMessage = serde_json::from_str(json).unwrap(); let msg: WsClientMessage = serde_json::from_str(json).unwrap();
match msg { match msg {
WsClientMessage::Message { WsClientMessage::Message { content, thread_id } => {
content, thread_id, ..
} => {
assert_eq!(content, "hi"); assert_eq!(content, "hi");
assert!(thread_id.is_none()); assert!(thread_id.is_none());
} }
+1 -10
View File
@@ -156,15 +156,8 @@ async fn handle_client_message(
direct_tx: &mpsc::Sender<WsServerMessage>, direct_tx: &mpsc::Sender<WsServerMessage>,
) { ) {
match msg { match msg {
WsClientMessage::Message { WsClientMessage::Message { content, thread_id } => {
content,
thread_id,
timezone,
} => {
let mut incoming = IncomingMessage::new("gateway", user_id, &content); let mut incoming = IncomingMessage::new("gateway", user_id, &content);
if let Some(ref tz) = timezone {
incoming = incoming.with_timezone(tz);
}
if let Some(ref tid) = thread_id { if let Some(ref tid) = thread_id {
incoming = incoming.with_thread(tid); incoming = incoming.with_thread(tid);
} }
@@ -356,7 +349,6 @@ mod tests {
WsClientMessage::Message { WsClientMessage::Message {
content: "hello agent".to_string(), content: "hello agent".to_string(),
thread_id: Some("t1".to_string()), thread_id: Some("t1".to_string()),
timezone: None,
}, },
&state, &state,
"user1", "user1",
@@ -381,7 +373,6 @@ mod tests {
WsClientMessage::Message { WsClientMessage::Message {
content: "hello".to_string(), content: "hello".to_string(),
thread_id: None, thread_id: None,
timezone: None,
}, },
&state, &state,
"user1", "user1",
+1 -123
View File
@@ -8,35 +8,9 @@ use std::path::PathBuf;
use crate::bootstrap::ironclaw_base_dir; use crate::bootstrap::ironclaw_base_dir;
use crate::settings::Settings; use crate::settings::Settings;
/// Load settings from JSON and TOML config files, matching the runtime
/// priority: TOML overlay > settings.json > defaults.
///
/// This mirrors the loading chain in `Config::from_env_with_toml()` but
/// without resolving the full `Config` (which requires async + secrets).
fn load_settings() -> Settings {
load_settings_from(&Settings::default_path(), &Settings::default_toml_path())
}
/// Inner implementation with injectable paths (testable).
fn load_settings_from(json_path: &std::path::Path, toml_path: &std::path::Path) -> Settings {
let mut settings = Settings::load_from(json_path);
match Settings::load_toml(toml_path) {
Ok(Some(toml_settings)) => {
settings.merge_from(&toml_settings);
}
Ok(None) => {} // File not found — fine for default path
Err(e) => {
eprintln!("Warning: failed to parse {}: {}", toml_path.display(), e);
}
}
settings
}
/// Run the status command, printing system health info. /// Run the status command, printing system health info.
pub async fn run_status_command() -> anyhow::Result<()> { pub async fn run_status_command() -> anyhow::Result<()> {
let settings = load_settings(); let settings = Settings::default();
println!("IronClaw Status"); println!("IronClaw Status");
println!("===============\n"); println!("===============\n");
@@ -235,99 +209,3 @@ fn default_tools_dir() -> PathBuf {
fn default_channels_dir() -> PathBuf { fn default_channels_dir() -> PathBuf {
ironclaw_base_dir().join("channels") ironclaw_base_dir().join("channels")
} }
#[cfg(test)]
mod tests {
use super::load_settings_from;
/// Regression test for #354: load_settings_from must read config.toml.
#[test]
fn reads_toml_heartbeat_enabled() {
let dir = tempfile::tempdir().expect("tempdir");
let json_path = dir.path().join("settings.json");
let toml_path = dir.path().join("config.toml");
// No JSON file — only TOML
std::fs::write(
&toml_path,
"[heartbeat]\nenabled = true\ninterval_secs = 600",
)
.expect("write toml");
let settings = load_settings_from(&json_path, &toml_path);
assert!(settings.heartbeat.enabled);
assert_eq!(settings.heartbeat.interval_secs, 600);
}
/// Without any config files, defaults are returned.
#[test]
fn defaults_without_config_files() {
let dir = tempfile::tempdir().expect("tempdir");
let settings = load_settings_from(
&dir.path().join("nonexistent.json"),
&dir.path().join("nonexistent.toml"),
);
assert!(!settings.heartbeat.enabled);
}
/// settings.json is respected.
#[test]
fn reads_json_heartbeat_enabled() {
let dir = tempfile::tempdir().expect("tempdir");
let json_path = dir.path().join("settings.json");
let toml_path = dir.path().join("nonexistent.toml");
std::fs::write(
&json_path,
r#"{"heartbeat":{"enabled":true,"interval_secs":900}}"#,
)
.expect("write json");
let settings = load_settings_from(&json_path, &toml_path);
assert!(settings.heartbeat.enabled);
assert_eq!(settings.heartbeat.interval_secs, 900);
}
/// TOML overlay wins over JSON settings.
#[test]
fn toml_overlay_wins_over_json() {
let dir = tempfile::tempdir().expect("tempdir");
let json_path = dir.path().join("settings.json");
let toml_path = dir.path().join("config.toml");
std::fs::write(
&json_path,
r#"{"heartbeat":{"enabled":false,"interval_secs":100}}"#,
)
.expect("write json");
std::fs::write(
&toml_path,
"[heartbeat]\nenabled = true\ninterval_secs = 200",
)
.expect("write toml");
let settings = load_settings_from(&json_path, &toml_path);
assert!(settings.heartbeat.enabled);
assert_eq!(settings.heartbeat.interval_secs, 200);
}
/// Invalid TOML is warned but doesn't crash; falls back to JSON/defaults.
#[test]
fn invalid_toml_falls_back_gracefully() {
let dir = tempfile::tempdir().expect("tempdir");
let json_path = dir.path().join("settings.json");
let toml_path = dir.path().join("config.toml");
std::fs::write(
&json_path,
r#"{"heartbeat":{"enabled":true,"interval_secs":500}}"#,
)
.expect("write json");
std::fs::write(&toml_path, "this is not valid toml [[[").expect("write bad toml");
let settings = load_settings_from(&json_path, &toml_path);
// Should fall back to JSON values, not crash
assert!(settings.heartbeat.enabled);
assert_eq!(settings.heartbeat.interval_secs, 500);
}
}
-37
View File
@@ -27,8 +27,6 @@ pub struct AgentConfig {
pub max_tool_iterations: usize, pub max_tool_iterations: usize,
/// When true, skip tool approval checks entirely. For benchmarks/CI. /// When true, skip tool approval checks entirely. For benchmarks/CI.
pub auto_approve_tools: bool, pub auto_approve_tools: bool,
/// Default timezone for new sessions (IANA name, e.g. "America/New_York").
pub default_timezone: String,
} }
impl AgentConfig { impl AgentConfig {
@@ -49,7 +47,6 @@ impl AgentConfig {
max_actions_per_hour: None, max_actions_per_hour: None,
max_tool_iterations: 10, max_tool_iterations: 10,
auto_approve_tools: true, auto_approve_tools: true,
default_timezone: "UTC".to_string(),
} }
} }
@@ -92,40 +89,6 @@ impl AgentConfig {
"AGENT_AUTO_APPROVE_TOOLS", "AGENT_AUTO_APPROVE_TOOLS",
settings.agent.auto_approve_tools, settings.agent.auto_approve_tools,
)?, )?,
default_timezone: {
let tz: String = parse_optional_env(
"DEFAULT_TIMEZONE",
settings.agent.default_timezone.clone(),
)?;
if crate::timezone::parse_timezone(&tz).is_none() {
return Err(ConfigError::InvalidValue {
key: "DEFAULT_TIMEZONE".into(),
message: format!("invalid IANA timezone: '{tz}'"),
});
}
tz
},
}) })
} }
} }
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_default_timezone_rejects_invalid() {
let mut settings = Settings::default();
settings.agent.default_timezone = "Fake/Zone".to_string();
let result = AgentConfig::resolve(&settings);
assert!(result.is_err(), "invalid IANA timezone should be rejected");
}
#[test]
fn test_default_timezone_accepts_valid() {
let settings = Settings::default(); // default is "UTC"
let config = AgentConfig::resolve(&settings).expect("resolve");
assert_eq!(config.default_timezone, "UTC");
}
}
-92
View File
@@ -82,31 +82,6 @@ impl std::str::FromStr for SslMode {
} }
} }
/// Which vector store backend to use for workspace semantic search.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum VectorBackend {
/// Use the database's built-in vector support (pgvector or libsql_vector_idx).
#[default]
Builtin,
/// Use LanceDB as an external vector store (requires `lancedb` feature).
LanceDb,
}
impl std::str::FromStr for VectorBackend {
type Err = String;
fn from_str(s: &str) -> Result<Self, Self::Err> {
match s.to_lowercase().as_str() {
"" | "builtin" | "pgvector" | "libsql" => Ok(Self::Builtin),
"lancedb" | "lance" => Ok(Self::LanceDb),
_ => Err(format!(
"invalid vector backend '{}', expected 'builtin' or 'lancedb'",
s
)),
}
}
}
/// Database configuration. /// Database configuration.
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct DatabaseConfig { pub struct DatabaseConfig {
@@ -126,12 +101,6 @@ pub struct DatabaseConfig {
pub libsql_url: Option<String>, pub libsql_url: Option<String>,
/// Turso auth token (required when libsql_url is set). /// Turso auth token (required when libsql_url is set).
pub libsql_auth_token: Option<SecretString>, pub libsql_auth_token: Option<SecretString>,
// -- Vector store fields --
/// Which vector store to use for workspace semantic search (default: Builtin).
pub vector_backend: VectorBackend,
/// Path to LanceDB directory (default: ~/.ironclaw/lancedb when vector_backend is LanceDb).
pub lancedb_path: Option<PathBuf>,
} }
impl DatabaseConfig { impl DatabaseConfig {
@@ -190,25 +159,6 @@ impl DatabaseConfig {
}); });
} }
let vector_backend: VectorBackend = if let Some(s) = optional_env("VECTOR_BACKEND")? {
s.parse().map_err(|e| ConfigError::InvalidValue {
key: "VECTOR_BACKEND".to_string(),
message: e,
})?
} else {
VectorBackend::default()
};
let lancedb_path = optional_env("LANCEDB_PATH")?
.map(PathBuf::from)
.or_else(|| {
if vector_backend == VectorBackend::LanceDb {
Some(default_lancedb_path())
} else {
None
}
});
Ok(Self { Ok(Self {
backend, backend,
url: SecretString::from(url), url: SecretString::from(url),
@@ -217,8 +167,6 @@ impl DatabaseConfig {
libsql_path, libsql_path,
libsql_url, libsql_url,
libsql_auth_token, libsql_auth_token,
vector_backend,
lancedb_path,
}) })
} }
@@ -247,11 +195,6 @@ pub fn default_libsql_path() -> PathBuf {
ironclaw_base_dir().join("ironclaw.db") ironclaw_base_dir().join("ironclaw.db")
} }
/// Default LanceDB directory (~/.ironclaw/lancedb).
pub fn default_lancedb_path() -> PathBuf {
ironclaw_base_dir().join("lancedb")
}
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
@@ -281,39 +224,4 @@ mod tests {
fn ssl_mode_parse_invalid() { fn ssl_mode_parse_invalid() {
assert!("invalid".parse::<SslMode>().is_err()); assert!("invalid".parse::<SslMode>().is_err());
} }
#[test]
fn vector_backend_parse() {
assert_eq!(
"builtin".parse::<VectorBackend>().unwrap(),
VectorBackend::Builtin
);
assert_eq!(
"pgvector".parse::<VectorBackend>().unwrap(),
VectorBackend::Builtin
);
assert_eq!(
"libsql".parse::<VectorBackend>().unwrap(),
VectorBackend::Builtin
);
assert_eq!("".parse::<VectorBackend>().unwrap(), VectorBackend::Builtin);
assert_eq!(
"lancedb".parse::<VectorBackend>().unwrap(),
VectorBackend::LanceDb
);
assert_eq!(
"lance".parse::<VectorBackend>().unwrap(),
VectorBackend::LanceDb
);
assert!("invalid".parse::<VectorBackend>().is_err());
}
#[test]
fn default_lancedb_path_under_ironclaw() {
let path = super::default_lancedb_path();
assert!(path.to_string_lossy().contains("ironclaw"));
assert!(path.to_string_lossy().ends_with("lancedb"));
}
} }
+1 -102
View File
@@ -1,4 +1,4 @@
use crate::config::helpers::{optional_env, parse_bool_env, parse_option_env, parse_optional_env}; use crate::config::helpers::{optional_env, parse_bool_env, parse_optional_env};
use crate::error::ConfigError; use crate::error::ConfigError;
use crate::settings::Settings; use crate::settings::Settings;
@@ -13,12 +13,6 @@ pub struct HeartbeatConfig {
pub notify_channel: Option<String>, pub notify_channel: Option<String>,
/// User ID to notify on heartbeat findings. /// User ID to notify on heartbeat findings.
pub notify_user: Option<String>, pub notify_user: Option<String>,
/// Hour (0-23) when quiet hours start.
pub quiet_hours_start: Option<u32>,
/// Hour (0-23) when quiet hours end.
pub quiet_hours_end: Option<u32>,
/// Timezone for quiet hours evaluation (IANA name).
pub timezone: Option<String>,
} }
impl Default for HeartbeatConfig { impl Default for HeartbeatConfig {
@@ -28,9 +22,6 @@ impl Default for HeartbeatConfig {
interval_secs: 1800, // 30 minutes interval_secs: 1800, // 30 minutes
notify_channel: None, notify_channel: None,
notify_user: None, notify_user: None,
quiet_hours_start: None,
quiet_hours_end: None,
timezone: None,
} }
} }
} }
@@ -47,98 +38,6 @@ impl HeartbeatConfig {
.or_else(|| settings.heartbeat.notify_channel.clone()), .or_else(|| settings.heartbeat.notify_channel.clone()),
notify_user: optional_env("HEARTBEAT_NOTIFY_USER")? notify_user: optional_env("HEARTBEAT_NOTIFY_USER")?
.or_else(|| settings.heartbeat.notify_user.clone()), .or_else(|| settings.heartbeat.notify_user.clone()),
quiet_hours_start: parse_option_env::<u32>("HEARTBEAT_QUIET_START")?
.or(settings.heartbeat.quiet_hours_start)
.map(|h| {
if h > 23 {
return Err(ConfigError::InvalidValue {
key: "HEARTBEAT_QUIET_START".into(),
message: "must be 0-23".into(),
});
}
Ok(h)
})
.transpose()?,
quiet_hours_end: parse_option_env::<u32>("HEARTBEAT_QUIET_END")?
.or(settings.heartbeat.quiet_hours_end)
.map(|h| {
if h > 23 {
return Err(ConfigError::InvalidValue {
key: "HEARTBEAT_QUIET_END".into(),
message: "must be 0-23".into(),
});
}
Ok(h)
})
.transpose()?,
timezone: {
let tz = optional_env("HEARTBEAT_TIMEZONE")?
.or_else(|| settings.heartbeat.timezone.clone());
if let Some(ref tz_str) = tz
&& crate::timezone::parse_timezone(tz_str).is_none()
{
return Err(ConfigError::InvalidValue {
key: "HEARTBEAT_TIMEZONE".into(),
message: format!("invalid IANA timezone: '{tz_str}'"),
});
}
tz
},
}) })
} }
} }
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_quiet_hours_settings_fallback() {
// When env vars are not set, settings values should be used
let mut settings = Settings::default();
settings.heartbeat.quiet_hours_start = Some(22);
settings.heartbeat.quiet_hours_end = Some(6);
let config = HeartbeatConfig::resolve(&settings).expect("resolve");
assert_eq!(config.quiet_hours_start, Some(22));
assert_eq!(config.quiet_hours_end, Some(6));
}
#[test]
fn test_quiet_hours_rejects_invalid_hour() {
let mut settings = Settings::default();
settings.heartbeat.quiet_hours_start = Some(24);
let result = HeartbeatConfig::resolve(&settings);
assert!(result.is_err());
}
#[test]
fn test_quiet_hours_accepts_boundary_values() {
let mut settings = Settings::default();
settings.heartbeat.quiet_hours_start = Some(0);
settings.heartbeat.quiet_hours_end = Some(23);
let config = HeartbeatConfig::resolve(&settings).expect("resolve");
assert_eq!(config.quiet_hours_start, Some(0));
assert_eq!(config.quiet_hours_end, Some(23));
}
#[test]
fn test_heartbeat_timezone_rejects_invalid() {
let mut settings = Settings::default();
settings.heartbeat.timezone = Some("Fake/Zone".to_string());
let result = HeartbeatConfig::resolve(&settings);
assert!(result.is_err(), "invalid IANA timezone should be rejected");
}
#[test]
fn test_heartbeat_timezone_accepts_valid() {
let mut settings = Settings::default();
settings.heartbeat.timezone = Some("America/New_York".to_string());
let config = HeartbeatConfig::resolve(&settings).expect("resolve");
assert_eq!(config.timezone.as_deref(), Some("America/New_York"));
}
}
-34
View File
@@ -103,10 +103,6 @@ pub struct LlmConfig {
/// Resolved provider config for registry-based providers. /// Resolved provider config for registry-based providers.
/// `None` when backend is "nearai". /// `None` when backend is "nearai".
pub provider: Option<RegistryProviderConfig>, pub provider: Option<RegistryProviderConfig>,
/// HTTP request timeout in seconds for LLM API calls.
/// Default: 120. Increase for local LLMs (Ollama, vLLM, LM Studio) that
/// need more time for prompt evaluation on consumer hardware.
pub request_timeout_secs: u64,
} }
/// NEAR AI configuration. /// NEAR AI configuration.
@@ -169,7 +165,6 @@ impl LlmConfig {
smart_routing_cascade: false, smart_routing_cascade: false,
}, },
provider: None, provider: None,
request_timeout_secs: 120,
} }
} }
@@ -259,8 +254,6 @@ impl LlmConfig {
)?) )?)
}; };
let request_timeout_secs = parse_optional_env("LLM_REQUEST_TIMEOUT_SECS", 120)?;
Ok(Self { Ok(Self {
backend: if is_nearai { backend: if is_nearai {
"nearai".to_string() "nearai".to_string()
@@ -272,7 +265,6 @@ impl LlmConfig {
session, session,
nearai, nearai,
provider, provider,
request_timeout_secs,
}) })
} }
@@ -1024,30 +1016,4 @@ mod tests {
assert_eq!(parsed, variant, "round-trip failed for {s}"); assert_eq!(parsed, variant, "round-trip failed for {s}");
} }
} }
#[test]
fn test_request_timeout_defaults_to_120() {
let _guard = ENV_MUTEX.lock().expect("env mutex poisoned");
// SAFETY: Under ENV_MUTEX.
unsafe {
std::env::remove_var("LLM_REQUEST_TIMEOUT_SECS");
}
let config = LlmConfig::resolve(&Settings::default()).expect("resolve");
assert_eq!(config.request_timeout_secs, 120);
}
#[test]
fn test_request_timeout_configurable() {
let _guard = ENV_MUTEX.lock().expect("env mutex poisoned");
// SAFETY: Under ENV_MUTEX.
unsafe {
std::env::set_var("LLM_REQUEST_TIMEOUT_SECS", "300");
}
let config = LlmConfig::resolve(&Settings::default()).expect("resolve");
assert_eq!(config.request_timeout_secs, 300);
// SAFETY: Cleanup
unsafe {
std::env::remove_var("LLM_REQUEST_TIMEOUT_SECS");
}
}
} }
+1 -32
View File
@@ -33,10 +33,7 @@ use crate::settings::Settings;
pub use self::agent::AgentConfig; pub use self::agent::AgentConfig;
pub use self::builder::BuilderModeConfig; pub use self::builder::BuilderModeConfig;
pub use self::channels::{ChannelsConfig, CliConfig, GatewayConfig, HttpConfig, SignalConfig}; pub use self::channels::{ChannelsConfig, CliConfig, GatewayConfig, HttpConfig, SignalConfig};
pub use self::database::{ pub use self::database::{DatabaseBackend, DatabaseConfig, SslMode, default_libsql_path};
DatabaseBackend, DatabaseConfig, SslMode, VectorBackend, default_lancedb_path,
default_libsql_path,
};
pub use self::embeddings::EmbeddingsConfig; pub use self::embeddings::EmbeddingsConfig;
pub use self::heartbeat::HeartbeatConfig; pub use self::heartbeat::HeartbeatConfig;
pub use self::hygiene::HygieneConfig; pub use self::hygiene::HygieneConfig;
@@ -110,8 +107,6 @@ impl Config {
libsql_path: Some(libsql_path), libsql_path: Some(libsql_path),
libsql_url: None, libsql_url: None,
libsql_auth_token: None, libsql_auth_token: None,
vector_backend: VectorBackend::default(),
lancedb_path: None,
}, },
llm: LlmConfig::for_testing(), llm: LlmConfig::for_testing(),
embeddings: EmbeddingsConfig::default(), embeddings: EmbeddingsConfig::default(),
@@ -262,32 +257,6 @@ impl Config {
Ok(()) Ok(())
} }
/// Re-resolve only the LLM config after credential injection.
///
/// Called by `AppBuilder::init_secrets()` after injecting API keys into
/// the env overlay. Only rebuilds `self.llm` — all other config fields
/// are unaffected, preserving values from the initial config load (or
/// from `Config::for_testing()` in test mode).
pub async fn re_resolve_llm(
&mut self,
store: Option<&(dyn crate::db::SettingsStore + Sync)>,
user_id: &str,
toml_path: Option<&std::path::Path>,
) -> Result<(), ConfigError> {
let settings = if let Some(store) = store {
let mut s = match store.get_all_settings(user_id).await {
Ok(map) => Settings::from_db_map(&map),
Err(_) => Settings::default(),
};
Self::apply_toml_overlay(&mut s, toml_path)?;
s
} else {
Settings::default()
};
self.llm = LlmConfig::resolve(&settings)?;
Ok(())
}
/// Build config from settings (shared by from_env and from_db). /// Build config from settings (shared by from_env and from_db).
async fn build(settings: &Settings) -> Result<Self, ConfigError> { async fn build(settings: &Settings) -> Result<Self, ConfigError> {
Ok(Self { Ok(Self {
-9
View File
@@ -164,8 +164,6 @@ pub struct JobContext {
/// previous results by ID via `$tool_call_id` parameter syntax. /// previous results by ID via `$tool_call_id` parameter syntax.
#[serde(skip)] #[serde(skip)]
pub tool_output_stash: Arc<tokio::sync::RwLock<HashMap<String, String>>>, pub tool_output_stash: Arc<tokio::sync::RwLock<HashMap<String, String>>>,
/// User's preferred timezone (IANA name, e.g. "America/New_York"). Defaults to "UTC".
pub user_timezone: String,
} }
impl JobContext { impl JobContext {
@@ -205,16 +203,9 @@ impl JobContext {
http_interceptor: None, http_interceptor: None,
metadata: serde_json::Value::Null, metadata: serde_json::Value::Null,
tool_output_stash: Arc::new(tokio::sync::RwLock::new(HashMap::new())), tool_output_stash: Arc::new(tokio::sync::RwLock::new(HashMap::new())),
user_timezone: "UTC".to_string(),
} }
} }
/// Set the user timezone on this context.
pub fn with_timezone(mut self, tz: impl Into<String>) -> Self {
self.user_timezone = tz.into();
self
}
/// Transition to a new state. /// Transition to a new state.
pub fn transition_to( pub fn transition_to(
&mut self, &mut self,
-3
View File
@@ -121,9 +121,6 @@ impl JobStore for LibSqlBackend {
tool_output_stash: std::sync::Arc::new(tokio::sync::RwLock::new( tool_output_stash: std::sync::Arc::new(tokio::sync::RwLock::new(
std::collections::HashMap::new(), std::collections::HashMap::new(),
)), )),
// TODO(#661): persist user_timezone in agent_jobs table so
// background/routine jobs retain the session's timezone context.
user_timezone: "UTC".to_string(),
})) }))
} }
None => Ok(None), None => Ok(None),
+102
View File
@@ -169,10 +169,18 @@ pub(crate) fn parse_timestamp(s: &str) -> Result<DateTime<Utc>, String> {
} }
// Naive with fractional seconds (legacy or SQLite datetime() output) // Naive with fractional seconds (legacy or SQLite datetime() output)
if let Ok(ndt) = NaiveDateTime::parse_from_str(s, "%Y-%m-%d %H:%M:%S%.f") { if let Ok(ndt) = NaiveDateTime::parse_from_str(s, "%Y-%m-%d %H:%M:%S%.f") {
tracing::warn!(
timestamp = s,
"parsing naive timestamp without timezone; assuming UTC — consider re-running migrations"
);
return Ok(ndt.and_utc()); return Ok(ndt.and_utc());
} }
// Naive without fractional seconds (legacy format) // Naive without fractional seconds (legacy format)
if let Ok(ndt) = NaiveDateTime::parse_from_str(s, "%Y-%m-%d %H:%M:%S") { if let Ok(ndt) = NaiveDateTime::parse_from_str(s, "%Y-%m-%d %H:%M:%S") {
tracing::warn!(
timestamp = s,
"parsing naive timestamp without timezone; assuming UTC — consider re-running migrations"
);
return Ok(ndt.and_utc()); return Ok(ndt.and_utc());
} }
Err(format!("unparseable timestamp: {:?}", s)) Err(format!("unparseable timestamp: {:?}", s))
@@ -510,4 +518,98 @@ mod tests {
); );
} }
} }
#[test]
fn test_parse_timestamp_rfc3339() {
use super::parse_timestamp;
// Standard RFC 3339 with Z suffix
let dt = parse_timestamp("2024-01-15T10:30:00.123Z").unwrap();
assert_eq!(
dt.to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
"2024-01-15T10:30:00.123Z"
);
// RFC 3339 with +00:00 offset
let dt = parse_timestamp("2024-01-15T10:30:00.000+00:00").unwrap();
assert_eq!(
dt.to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
"2024-01-15T10:30:00.000Z"
);
}
#[test]
fn test_parse_timestamp_naive_fallback() {
use super::parse_timestamp;
// Naive with fractional seconds (legacy datetime('now') output)
let dt = parse_timestamp("2024-01-15 10:30:00.123").unwrap();
assert_eq!(
dt.to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
"2024-01-15T10:30:00.123Z"
);
// Naive without fractional seconds
let dt = parse_timestamp("2024-01-15 10:30:00").unwrap();
assert_eq!(
dt.to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
"2024-01-15T10:30:00.000Z"
);
}
#[test]
fn test_parse_timestamp_invalid() {
use super::parse_timestamp;
assert!(parse_timestamp("not-a-timestamp").is_err());
assert!(parse_timestamp("").is_err());
}
#[tokio::test]
async fn test_default_timestamps_are_rfc3339() {
// Verify that DEFAULT column values produce RFC 3339 timestamps
// after the migration change from datetime('now') to strftime.
// Use file-based DB because in-memory doesn't share schema across connections.
let dir = tempfile::tempdir().unwrap();
let db_path = dir.path().join("test_ts.db");
let backend = LibSqlBackend::new_local(&db_path).await.unwrap();
backend.run_migrations().await.unwrap();
let conn = backend.connect().await.unwrap();
let id = uuid::Uuid::new_v4().to_string();
conn.execute(
"INSERT INTO conversations (id, channel, user_id) VALUES (?1, ?2, ?3)",
libsql::params![id.clone(), "test", "user1"],
)
.await
.unwrap();
let mut rows = conn
.query(
"SELECT started_at, last_activity FROM conversations WHERE id = ?1",
libsql::params![id],
)
.await
.unwrap();
let row = rows.next().await.unwrap().unwrap();
let started_at: String = row.get(0).unwrap();
let last_activity: String = row.get(1).unwrap();
// Must end with 'Z' (RFC 3339 UTC) and contain 'T' separator
assert!(
started_at.ends_with('Z') && started_at.contains('T'),
"started_at should be RFC 3339, got: {started_at}"
);
assert!(
last_activity.ends_with('Z') && last_activity.contains('T'),
"last_activity should be RFC 3339, got: {last_activity}"
);
// Must be parseable by the RFC 3339 parser directly (not just naive fallback)
use chrono::DateTime;
assert!(
DateTime::parse_from_rfc3339(&started_at).is_ok(),
"started_at not valid RFC 3339: {started_at}"
);
}
} }
+55 -55
View File
@@ -26,7 +26,7 @@ pub const SCHEMA: &str = r#"
CREATE TABLE IF NOT EXISTS _migrations ( CREATE TABLE IF NOT EXISTS _migrations (
version INTEGER PRIMARY KEY, version INTEGER PRIMARY KEY,
name TEXT NOT NULL, name TEXT NOT NULL,
applied_at TEXT NOT NULL DEFAULT (datetime('now')) applied_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
); );
-- ==================== Conversations ==================== -- ==================== Conversations ====================
@@ -36,8 +36,8 @@ CREATE TABLE IF NOT EXISTS conversations (
channel TEXT NOT NULL, channel TEXT NOT NULL,
user_id TEXT NOT NULL, user_id TEXT NOT NULL,
thread_id TEXT, thread_id TEXT,
started_at TEXT NOT NULL DEFAULT (datetime('now')), started_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
last_activity TEXT NOT NULL DEFAULT (datetime('now')), last_activity TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
metadata TEXT NOT NULL DEFAULT '{}' metadata TEXT NOT NULL DEFAULT '{}'
); );
@@ -59,7 +59,7 @@ CREATE TABLE IF NOT EXISTS conversation_messages (
conversation_id TEXT NOT NULL REFERENCES conversations(id) ON DELETE CASCADE, conversation_id TEXT NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
role TEXT NOT NULL, role TEXT NOT NULL,
content TEXT NOT NULL, content TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT (datetime('now')) created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
); );
CREATE INDEX IF NOT EXISTS idx_conversation_messages_conversation CREATE INDEX IF NOT EXISTS idx_conversation_messages_conversation
@@ -91,7 +91,7 @@ CREATE TABLE IF NOT EXISTS agent_jobs (
failure_reason TEXT, failure_reason TEXT,
stuck_since TEXT, stuck_since TEXT,
repair_attempts INTEGER NOT NULL DEFAULT 0, repair_attempts INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL DEFAULT (datetime('now')), created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
started_at TEXT, started_at TEXT,
completed_at TEXT completed_at TEXT
); );
@@ -116,7 +116,7 @@ CREATE TABLE IF NOT EXISTS job_actions (
duration_ms INTEGER, duration_ms INTEGER,
success INTEGER NOT NULL, success INTEGER NOT NULL,
error_message TEXT, error_message TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now')), created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
UNIQUE(job_id, sequence_num) UNIQUE(job_id, sequence_num)
); );
@@ -137,8 +137,8 @@ CREATE TABLE IF NOT EXISTS dynamic_tools (
failure_count INTEGER NOT NULL DEFAULT 0, failure_count INTEGER NOT NULL DEFAULT 0,
last_error TEXT, last_error TEXT,
status TEXT NOT NULL DEFAULT 'active', status TEXT NOT NULL DEFAULT 'active',
created_at TEXT NOT NULL DEFAULT (datetime('now')), created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now')) updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
); );
CREATE INDEX IF NOT EXISTS idx_dynamic_tools_status ON dynamic_tools(status); CREATE INDEX IF NOT EXISTS idx_dynamic_tools_status ON dynamic_tools(status);
@@ -156,7 +156,7 @@ CREATE TABLE IF NOT EXISTS llm_calls (
output_tokens INTEGER NOT NULL, output_tokens INTEGER NOT NULL,
cost TEXT NOT NULL, cost TEXT NOT NULL,
purpose TEXT, purpose TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now')) created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
); );
CREATE INDEX IF NOT EXISTS idx_llm_calls_job ON llm_calls(job_id); CREATE INDEX IF NOT EXISTS idx_llm_calls_job ON llm_calls(job_id);
@@ -176,7 +176,7 @@ CREATE TABLE IF NOT EXISTS estimation_snapshots (
actual_time_secs INTEGER, actual_time_secs INTEGER,
estimated_value TEXT NOT NULL, estimated_value TEXT NOT NULL,
actual_value TEXT, actual_value TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now')) created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
); );
CREATE INDEX IF NOT EXISTS idx_estimation_category ON estimation_snapshots(category); CREATE INDEX IF NOT EXISTS idx_estimation_category ON estimation_snapshots(category);
@@ -192,7 +192,7 @@ CREATE TABLE IF NOT EXISTS repair_attempts (
action_taken TEXT NOT NULL, action_taken TEXT NOT NULL,
success INTEGER NOT NULL, success INTEGER NOT NULL,
error_message TEXT, error_message TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now')) created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
); );
CREATE INDEX IF NOT EXISTS idx_repair_attempts_target ON repair_attempts(target_type, target_id); CREATE INDEX IF NOT EXISTS idx_repair_attempts_target ON repair_attempts(target_type, target_id);
@@ -206,8 +206,8 @@ CREATE TABLE IF NOT EXISTS memory_documents (
agent_id TEXT, agent_id TEXT,
path TEXT NOT NULL, path TEXT NOT NULL,
content TEXT NOT NULL, content TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT (datetime('now')), created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now')), updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
metadata TEXT NOT NULL DEFAULT '{}', metadata TEXT NOT NULL DEFAULT '{}',
UNIQUE (user_id, agent_id, path) UNIQUE (user_id, agent_id, path)
); );
@@ -222,7 +222,7 @@ CREATE TRIGGER IF NOT EXISTS update_memory_documents_updated_at
FOR EACH ROW FOR EACH ROW
WHEN NEW.updated_at = OLD.updated_at WHEN NEW.updated_at = OLD.updated_at
BEGIN BEGIN
UPDATE memory_documents SET updated_at = datetime('now') WHERE id = NEW.id; UPDATE memory_documents SET updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = NEW.id;
END; END;
-- ==================== Workspace: Memory Chunks ==================== -- ==================== Workspace: Memory Chunks ====================
@@ -234,7 +234,7 @@ CREATE TABLE IF NOT EXISTS memory_chunks (
chunk_index INTEGER NOT NULL, chunk_index INTEGER NOT NULL,
content TEXT NOT NULL, content TEXT NOT NULL,
embedding BLOB, embedding BLOB,
created_at TEXT NOT NULL DEFAULT (datetime('now')), created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
UNIQUE (document_id, chunk_index) UNIQUE (document_id, chunk_index)
); );
@@ -296,8 +296,8 @@ CREATE TABLE IF NOT EXISTS secrets (
expires_at TEXT, expires_at TEXT,
last_used_at TEXT, last_used_at TEXT,
usage_count INTEGER NOT NULL DEFAULT 0, usage_count INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL DEFAULT (datetime('now')), created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now')), updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
UNIQUE (user_id, name) UNIQUE (user_id, name)
); );
@@ -318,8 +318,8 @@ CREATE TABLE IF NOT EXISTS wasm_tools (
source_url TEXT, source_url TEXT,
trust_level TEXT NOT NULL DEFAULT 'user', trust_level TEXT NOT NULL DEFAULT 'user',
status TEXT NOT NULL DEFAULT 'active', status TEXT NOT NULL DEFAULT 'active',
created_at TEXT NOT NULL DEFAULT (datetime('now')), created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now')), updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
UNIQUE (user_id, name, version) UNIQUE (user_id, name, version)
); );
@@ -340,8 +340,8 @@ CREATE TABLE IF NOT EXISTS wasm_channels (
binary_hash BLOB NOT NULL, binary_hash BLOB NOT NULL,
capabilities_json TEXT NOT NULL DEFAULT '{}', capabilities_json TEXT NOT NULL DEFAULT '{}',
status TEXT NOT NULL DEFAULT 'active', status TEXT NOT NULL DEFAULT 'active',
created_at TEXT NOT NULL DEFAULT (datetime('now')), created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now')), updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
UNIQUE (user_id, name) UNIQUE (user_id, name)
); );
@@ -359,8 +359,8 @@ CREATE TABLE IF NOT EXISTS tool_capabilities (
max_response_body_bytes INTEGER NOT NULL DEFAULT 10485760, max_response_body_bytes INTEGER NOT NULL DEFAULT 10485760,
workspace_read_prefixes TEXT NOT NULL DEFAULT '[]', workspace_read_prefixes TEXT NOT NULL DEFAULT '[]',
http_timeout_secs INTEGER NOT NULL DEFAULT 30, http_timeout_secs INTEGER NOT NULL DEFAULT 30,
created_at TEXT NOT NULL DEFAULT (datetime('now')), created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now')), updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
UNIQUE (wasm_tool_id) UNIQUE (wasm_tool_id)
); );
@@ -373,7 +373,7 @@ CREATE TABLE IF NOT EXISTS leak_detection_patterns (
severity TEXT NOT NULL DEFAULT 'high', severity TEXT NOT NULL DEFAULT 'high',
action TEXT NOT NULL DEFAULT 'block', action TEXT NOT NULL DEFAULT 'block',
enabled INTEGER NOT NULL DEFAULT 1, enabled INTEGER NOT NULL DEFAULT 1,
created_at TEXT NOT NULL DEFAULT (datetime('now')) created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
); );
-- ==================== Rate Limit State ==================== -- ==================== Rate Limit State ====================
@@ -382,9 +382,9 @@ CREATE TABLE IF NOT EXISTS tool_rate_limit_state (
id TEXT PRIMARY KEY, id TEXT PRIMARY KEY,
wasm_tool_id TEXT NOT NULL REFERENCES wasm_tools(id) ON DELETE CASCADE, wasm_tool_id TEXT NOT NULL REFERENCES wasm_tools(id) ON DELETE CASCADE,
user_id TEXT NOT NULL, user_id TEXT NOT NULL,
minute_window_start TEXT NOT NULL DEFAULT (datetime('now')), minute_window_start TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
minute_count INTEGER NOT NULL DEFAULT 0, minute_count INTEGER NOT NULL DEFAULT 0,
hour_window_start TEXT NOT NULL DEFAULT (datetime('now')), hour_window_start TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
hour_count INTEGER NOT NULL DEFAULT 0, hour_count INTEGER NOT NULL DEFAULT 0,
UNIQUE (wasm_tool_id, user_id) UNIQUE (wasm_tool_id, user_id)
); );
@@ -400,7 +400,7 @@ CREATE TABLE IF NOT EXISTS secret_usage_log (
target_path TEXT, target_path TEXT,
success INTEGER NOT NULL, success INTEGER NOT NULL,
error_message TEXT, error_message TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now')) created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
); );
CREATE INDEX IF NOT EXISTS idx_secret_usage_user ON secret_usage_log(user_id); CREATE INDEX IF NOT EXISTS idx_secret_usage_user ON secret_usage_log(user_id);
@@ -415,7 +415,7 @@ CREATE TABLE IF NOT EXISTS leak_detection_events (
source TEXT NOT NULL, source TEXT NOT NULL,
action_taken TEXT NOT NULL, action_taken TEXT NOT NULL,
context_preview TEXT, context_preview TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now')) created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
); );
-- ==================== Tool Failures ==================== -- ==================== Tool Failures ====================
@@ -425,8 +425,8 @@ CREATE TABLE IF NOT EXISTS tool_failures (
tool_name TEXT NOT NULL UNIQUE, tool_name TEXT NOT NULL UNIQUE,
error_message TEXT, error_message TEXT,
error_count INTEGER DEFAULT 1, error_count INTEGER DEFAULT 1,
first_failure TEXT DEFAULT (datetime('now')), first_failure TEXT DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
last_failure TEXT DEFAULT (datetime('now')), last_failure TEXT DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
last_build_result TEXT, last_build_result TEXT,
repaired_at TEXT, repaired_at TEXT,
repair_attempts INTEGER DEFAULT 0 repair_attempts INTEGER DEFAULT 0
@@ -441,7 +441,7 @@ CREATE TABLE IF NOT EXISTS job_events (
job_id TEXT NOT NULL REFERENCES agent_jobs(id), job_id TEXT NOT NULL REFERENCES agent_jobs(id),
event_type TEXT NOT NULL, event_type TEXT NOT NULL,
data TEXT NOT NULL, data TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT (datetime('now')) created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
); );
CREATE INDEX IF NOT EXISTS idx_job_events_job ON job_events(job_id, id); CREATE INDEX IF NOT EXISTS idx_job_events_job ON job_events(job_id, id);
@@ -471,8 +471,8 @@ CREATE TABLE IF NOT EXISTS routines (
next_fire_at TEXT, next_fire_at TEXT,
run_count INTEGER NOT NULL DEFAULT 0, run_count INTEGER NOT NULL DEFAULT 0,
consecutive_failures INTEGER NOT NULL DEFAULT 0, consecutive_failures INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL DEFAULT (datetime('now')), created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now')), updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
UNIQUE (user_id, name) UNIQUE (user_id, name)
); );
@@ -485,13 +485,13 @@ CREATE TABLE IF NOT EXISTS routine_runs (
routine_id TEXT NOT NULL REFERENCES routines(id) ON DELETE CASCADE, routine_id TEXT NOT NULL REFERENCES routines(id) ON DELETE CASCADE,
trigger_type TEXT NOT NULL, trigger_type TEXT NOT NULL,
trigger_detail TEXT, trigger_detail TEXT,
started_at TEXT NOT NULL DEFAULT (datetime('now')), started_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
completed_at TEXT, completed_at TEXT,
status TEXT NOT NULL DEFAULT 'running', status TEXT NOT NULL DEFAULT 'running',
result_summary TEXT, result_summary TEXT,
tokens_used INTEGER, tokens_used INTEGER,
job_id TEXT REFERENCES agent_jobs(id), job_id TEXT REFERENCES agent_jobs(id),
created_at TEXT NOT NULL DEFAULT (datetime('now')) created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
); );
CREATE INDEX IF NOT EXISTS idx_routine_runs_routine ON routine_runs(routine_id); CREATE INDEX IF NOT EXISTS idx_routine_runs_routine ON routine_runs(routine_id);
@@ -502,7 +502,7 @@ CREATE TABLE IF NOT EXISTS settings (
user_id TEXT NOT NULL, user_id TEXT NOT NULL,
key TEXT NOT NULL, key TEXT NOT NULL,
value TEXT NOT NULL, value TEXT NOT NULL,
updated_at TEXT NOT NULL DEFAULT (datetime('now')), updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
PRIMARY KEY (user_id, key) PRIMARY KEY (user_id, key)
); );
@@ -558,24 +558,24 @@ CREATE INDEX IF NOT EXISTS idx_heartbeat_next_run ON heartbeat_state(next_run);
-- Pre-populate leak detection patterns (matches PostgreSQL V2 migration). -- Pre-populate leak detection patterns (matches PostgreSQL V2 migration).
INSERT OR IGNORE INTO leak_detection_patterns (id, name, pattern, severity, action, enabled, created_at) VALUES INSERT OR IGNORE INTO leak_detection_patterns (id, name, pattern, severity, action, enabled, created_at) VALUES
('550e8400-e29b-41d4-a716-446655440001', 'openai_api_key', 'sk-(?:proj-)?[a-zA-Z0-9]{20,}(?:T3BlbkFJ[a-zA-Z0-9_-]*)?', 'critical', 'block', 1, datetime('now')), ('550e8400-e29b-41d4-a716-446655440001', 'openai_api_key', 'sk-(?:proj-)?[a-zA-Z0-9]{20,}(?:T3BlbkFJ[a-zA-Z0-9_-]*)?', 'critical', 'block', 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
('550e8400-e29b-41d4-a716-446655440002', 'anthropic_api_key', 'sk-ant-api[a-zA-Z0-9_-]{90,}', 'critical', 'block', 1, datetime('now')), ('550e8400-e29b-41d4-a716-446655440002', 'anthropic_api_key', 'sk-ant-api[a-zA-Z0-9_-]{90,}', 'critical', 'block', 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
('550e8400-e29b-41d4-a716-446655440003', 'aws_access_key', 'AKIA[0-9A-Z]{16}', 'critical', 'block', 1, datetime('now')), ('550e8400-e29b-41d4-a716-446655440003', 'aws_access_key', 'AKIA[0-9A-Z]{16}', 'critical', 'block', 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
('550e8400-e29b-41d4-a716-446655440004', 'aws_secret_key', '(?<![A-Za-z0-9/+=])[A-Za-z0-9/+=]{40}(?![A-Za-z0-9/+=])', 'high', 'block', 1, datetime('now')), ('550e8400-e29b-41d4-a716-446655440004', 'aws_secret_key', '(?<![A-Za-z0-9/+=])[A-Za-z0-9/+=]{40}(?![A-Za-z0-9/+=])', 'high', 'block', 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
('550e8400-e29b-41d4-a716-446655440005', 'github_token', 'gh[pousr]_[A-Za-z0-9_]{36,}', 'critical', 'block', 1, datetime('now')), ('550e8400-e29b-41d4-a716-446655440005', 'github_token', 'gh[pousr]_[A-Za-z0-9_]{36,}', 'critical', 'block', 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
('550e8400-e29b-41d4-a716-446655440006', 'github_fine_grained_pat', 'github_pat_[a-zA-Z0-9]{22}_[a-zA-Z0-9]{59}', 'critical', 'block', 1, datetime('now')), ('550e8400-e29b-41d4-a716-446655440006', 'github_fine_grained_pat', 'github_pat_[a-zA-Z0-9]{22}_[a-zA-Z0-9]{59}', 'critical', 'block', 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
('550e8400-e29b-41d4-a716-446655440007', 'stripe_api_key', 'sk_(?:live|test)_[a-zA-Z0-9]{24,}', 'critical', 'block', 1, datetime('now')), ('550e8400-e29b-41d4-a716-446655440007', 'stripe_api_key', 'sk_(?:live|test)_[a-zA-Z0-9]{24,}', 'critical', 'block', 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
('550e8400-e29b-41d4-a716-446655440008', 'nearai_session', 'sess_[a-zA-Z0-9]{32,}', 'critical', 'block', 1, datetime('now')), ('550e8400-e29b-41d4-a716-446655440008', 'nearai_session', 'sess_[a-zA-Z0-9]{32,}', 'critical', 'block', 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
('550e8400-e29b-41d4-a716-446655440009', 'bearer_token', 'Bearer\s+[a-zA-Z0-9_-]{20,}', 'high', 'redact', 1, datetime('now')), ('550e8400-e29b-41d4-a716-446655440009', 'bearer_token', 'Bearer\s+[a-zA-Z0-9_-]{20,}', 'high', 'redact', 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
('550e8400-e29b-41d4-a716-44665544000a', 'pem_private_key', '-----BEGIN\s+(?:RSA\s+)?PRIVATE\s+KEY-----', 'critical', 'block', 1, datetime('now')), ('550e8400-e29b-41d4-a716-44665544000a', 'pem_private_key', '-----BEGIN\s+(?:RSA\s+)?PRIVATE\s+KEY-----', 'critical', 'block', 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
('550e8400-e29b-41d4-a716-44665544000b', 'ssh_private_key', '-----BEGIN\s+(?:OPENSSH|EC|DSA)\s+PRIVATE\s+KEY-----', 'critical', 'block', 1, datetime('now')), ('550e8400-e29b-41d4-a716-44665544000b', 'ssh_private_key', '-----BEGIN\s+(?:OPENSSH|EC|DSA)\s+PRIVATE\s+KEY-----', 'critical', 'block', 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
('550e8400-e29b-41d4-a716-44665544000c', 'google_api_key', 'AIza[0-9A-Za-z_-]{35}', 'high', 'block', 1, datetime('now')), ('550e8400-e29b-41d4-a716-44665544000c', 'google_api_key', 'AIza[0-9A-Za-z_-]{35}', 'high', 'block', 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
('550e8400-e29b-41d4-a716-44665544000d', 'slack_token', 'xox[baprs]-[0-9a-zA-Z-]{10,}', 'high', 'block', 1, datetime('now')), ('550e8400-e29b-41d4-a716-44665544000d', 'slack_token', 'xox[baprs]-[0-9a-zA-Z-]{10,}', 'high', 'block', 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
('550e8400-e29b-41d4-a716-44665544000e', 'discord_token', '[MN][A-Za-z\d]{23,}\.[\w-]{6}\.[\w-]{27}', 'high', 'block', 1, datetime('now')), ('550e8400-e29b-41d4-a716-44665544000e', 'discord_token', '[MN][A-Za-z\d]{23,}\.[\w-]{6}\.[\w-]{27}', 'high', 'block', 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
('550e8400-e29b-41d4-a716-44665544000f', 'twilio_api_key', 'SK[a-fA-F0-9]{32}', 'high', 'block', 1, datetime('now')), ('550e8400-e29b-41d4-a716-44665544000f', 'twilio_api_key', 'SK[a-fA-F0-9]{32}', 'high', 'block', 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
('550e8400-e29b-41d4-a716-446655440010', 'sendgrid_api_key', 'SG\.[a-zA-Z0-9_-]{22}\.[a-zA-Z0-9_-]{43}', 'high', 'block', 1, datetime('now')), ('550e8400-e29b-41d4-a716-446655440010', 'sendgrid_api_key', 'SG\.[a-zA-Z0-9_-]{22}\.[a-zA-Z0-9_-]{43}', 'high', 'block', 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
('550e8400-e29b-41d4-a716-446655440011', 'mailchimp_api_key', '[a-f0-9]{32}-us[0-9]{1,2}', 'medium', 'block', 1, datetime('now')), ('550e8400-e29b-41d4-a716-446655440011', 'mailchimp_api_key', '[a-f0-9]{32}-us[0-9]{1,2}', 'medium', 'block', 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
('550e8400-e29b-41d4-a716-446655440012', 'high_entropy_hex', '(?<![a-fA-F0-9])[a-fA-F0-9]{64}(?![a-fA-F0-9])', 'medium', 'warn', 1, datetime('now')); ('550e8400-e29b-41d4-a716-446655440012', 'high_entropy_hex', '(?<![a-fA-F0-9])[a-fA-F0-9]{64}(?![a-fA-F0-9])', 'medium', 'warn', 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'));
"#; "#;
@@ -613,7 +613,7 @@ CREATE TABLE IF NOT EXISTS memory_chunks_new (
chunk_index INTEGER NOT NULL, chunk_index INTEGER NOT NULL,
content TEXT NOT NULL, content TEXT NOT NULL,
embedding BLOB, embedding BLOB,
created_at TEXT NOT NULL DEFAULT (datetime('now')), created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
UNIQUE (document_id, chunk_index) UNIQUE (document_id, chunk_index)
); );
+227 -123
View File
@@ -1,10 +1,13 @@
//! Success evaluation for jobs. //! Success evaluation for jobs.
use std::sync::Arc;
use async_trait::async_trait; use async_trait::async_trait;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use crate::context::{ActionRecord, JobContext}; use crate::context::{ActionRecord, JobContext};
use crate::error::EvaluationError; use crate::error::EvaluationError;
use crate::llm::LlmProvider;
/// Result of evaluating job success. /// Result of evaluating job success.
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -61,132 +64,233 @@ pub trait SuccessEvaluator: Send + Sync {
) -> Result<EvaluationResult, EvaluationError>; ) -> Result<EvaluationResult, EvaluationError>;
} }
/// Rule-based success evaluator.
pub struct RuleBasedEvaluator {
/// Minimum success rate for actions.
min_action_success_rate: f64,
/// Maximum allowed failures.
max_failures: u32,
}
impl RuleBasedEvaluator {
/// Create a new rule-based evaluator.
pub fn new() -> Self {
Self {
min_action_success_rate: 0.8,
max_failures: 3,
}
}
/// Set minimum action success rate.
#[allow(dead_code)] // Public API for configuring evaluation threshold
pub fn with_min_success_rate(mut self, rate: f64) -> Self {
self.min_action_success_rate = rate;
self
}
/// Set maximum failures.
#[allow(dead_code)] // Public API for configuring failure tolerance
pub fn with_max_failures(mut self, max: u32) -> Self {
self.max_failures = max;
self
}
}
impl Default for RuleBasedEvaluator {
fn default() -> Self {
Self::new()
}
}
#[async_trait]
impl SuccessEvaluator for RuleBasedEvaluator {
async fn evaluate(
&self,
job: &JobContext,
actions: &[ActionRecord],
_output: Option<&str>,
) -> Result<EvaluationResult, EvaluationError> {
let mut issues = Vec::new();
// Check if there were any actions
if actions.is_empty() {
return Ok(EvaluationResult::failure(
"No actions were taken",
vec!["No actions recorded".to_string()],
));
}
// Calculate action success rate
let successful = actions.iter().filter(|a| a.success).count();
let total = actions.len();
let success_rate = successful as f64 / total as f64;
if success_rate < self.min_action_success_rate {
issues.push(format!(
"Action success rate {:.1}% below threshold {:.1}%",
success_rate * 100.0,
self.min_action_success_rate * 100.0
));
}
// Count failures
let failures = actions.iter().filter(|a| !a.success).count() as u32;
if failures > self.max_failures {
issues.push(format!(
"Too many failures: {} (max {})",
failures, self.max_failures
));
}
// Check for critical errors
for action in actions.iter().filter(|a| !a.success) {
if let Some(ref error) = action.error
&& (error.to_lowercase().contains("critical")
|| error.to_lowercase().contains("fatal"))
{
issues.push(format!("Critical error in {}: {}", action.tool_name, error));
}
}
// Check job state
if job.state != crate::context::JobState::Completed
&& job.state != crate::context::JobState::Submitted
{
issues.push(format!("Job not in completed state: {:?}", job.state));
}
// Calculate quality score
let quality_score = if issues.is_empty() {
let base_score = (success_rate * 80.0) as u32;
let completion_bonus = if job.state == crate::context::JobState::Completed {
20
} else {
0
};
(base_score + completion_bonus).min(100)
} else {
((success_rate * 50.0) as u32).min(50)
};
if issues.is_empty() {
Ok(EvaluationResult::success(
format!(
"Job completed successfully with {}/{} actions succeeding ({:.1}%)",
successful,
total,
success_rate * 100.0
),
quality_score,
))
} else {
Ok(EvaluationResult {
success: false,
confidence: 0.85,
reasoning: format!("Job had {} issues", issues.len()),
issues,
suggestions: vec![
"Review failed actions for common patterns".to_string(),
"Consider adjusting retry logic".to_string(),
],
quality_score,
})
}
}
}
/// LLM-based success evaluator for more nuanced evaluation.
pub struct LlmEvaluator {
llm: Arc<dyn LlmProvider>,
}
impl LlmEvaluator {
/// Create a new LLM-based evaluator.
#[allow(dead_code)] // Public API for LLM-based evaluation
pub fn new(llm: Arc<dyn LlmProvider>) -> Self {
Self { llm }
}
}
#[async_trait]
impl SuccessEvaluator for LlmEvaluator {
async fn evaluate(
&self,
job: &JobContext,
actions: &[ActionRecord],
output: Option<&str>,
) -> Result<EvaluationResult, EvaluationError> {
// Build evaluation prompt
let actions_summary: Vec<String> = actions
.iter()
.map(|a| {
format!(
"- {}: {} ({})",
a.tool_name,
if a.success { "success" } else { "failed" },
a.error.as_deref().unwrap_or("ok")
)
})
.collect();
let prompt = format!(
r#"Evaluate if this job was completed successfully.
Job: {}
Description: {}
State: {:?}
Actions taken:
{}
{}
Respond in JSON format:
{{
"success": true/false,
"confidence": 0.0-1.0,
"reasoning": "...",
"issues": ["..."],
"suggestions": ["..."],
"quality_score": 0-100
}}"#,
job.title,
job.description,
job.state,
actions_summary.join("\n"),
output
.map(|o| format!("Output:\n{}", o))
.unwrap_or_default()
);
let request =
crate::llm::CompletionRequest::new(vec![crate::llm::ChatMessage::user(prompt)])
.with_max_tokens(1024)
.with_temperature(0.1);
let response = self
.llm
.complete(request)
.await
.map_err(|e| EvaluationError::Failed {
job_id: job.job_id,
reason: e.to_string(),
})?;
// Parse the response
let result: EvaluationResult =
serde_json::from_str(&response.content).map_err(|e| EvaluationError::Failed {
job_id: job.job_id,
reason: format!("Failed to parse LLM evaluation: {}", e),
})?;
Ok(result)
}
}
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
use crate::context::{ActionRecord, JobContext}; use crate::context::JobContext;
use crate::error::EvaluationError;
/// Rule-based success evaluator (test-only; no production callers).
struct RuleBasedEvaluator {
min_action_success_rate: f64,
max_failures: u32,
}
impl RuleBasedEvaluator {
fn new() -> Self {
Self {
min_action_success_rate: 0.8,
max_failures: 3,
}
}
fn with_min_success_rate(mut self, rate: f64) -> Self {
self.min_action_success_rate = rate;
self
}
fn with_max_failures(mut self, max: u32) -> Self {
self.max_failures = max;
self
}
}
impl Default for RuleBasedEvaluator {
fn default() -> Self {
Self::new()
}
}
#[async_trait::async_trait]
impl SuccessEvaluator for RuleBasedEvaluator {
async fn evaluate(
&self,
job: &JobContext,
actions: &[ActionRecord],
_output: Option<&str>,
) -> Result<EvaluationResult, EvaluationError> {
let mut issues = Vec::new();
if actions.is_empty() {
return Ok(EvaluationResult::failure(
"No actions were taken",
vec!["No actions recorded".to_string()],
));
}
let successful = actions.iter().filter(|a| a.success).count();
let total = actions.len();
let success_rate = successful as f64 / total as f64;
if success_rate < self.min_action_success_rate {
issues.push(format!(
"Action success rate {:.1}% below threshold {:.1}%",
success_rate * 100.0,
self.min_action_success_rate * 100.0
));
}
let failures = actions.iter().filter(|a| !a.success).count() as u32;
if failures > self.max_failures {
issues.push(format!(
"Too many failures: {} (max {})",
failures, self.max_failures
));
}
for action in actions.iter().filter(|a| !a.success) {
if let Some(ref error) = action.error
&& (error.to_lowercase().contains("critical")
|| error.to_lowercase().contains("fatal"))
{
issues.push(format!("Critical error in {}: {}", action.tool_name, error));
}
}
if job.state != crate::context::JobState::Completed
&& job.state != crate::context::JobState::Submitted
{
issues.push(format!("Job not in completed state: {:?}", job.state));
}
let quality_score = if issues.is_empty() {
let base_score = (success_rate * 80.0) as u32;
let completion_bonus = if job.state == crate::context::JobState::Completed {
20
} else {
0
};
(base_score + completion_bonus).min(100)
} else {
((success_rate * 50.0) as u32).min(50)
};
if issues.is_empty() {
Ok(EvaluationResult::success(
format!(
"Job completed successfully with {}/{} actions succeeding ({:.1}%)",
successful,
total,
success_rate * 100.0
),
quality_score,
))
} else {
Ok(EvaluationResult {
success: false,
confidence: 0.85,
reasoning: format!("Job had {} issues", issues.len()),
issues,
suggestions: vec![
"Review failed actions for common patterns".to_string(),
"Consider adjusting retry logic".to_string(),
],
quality_score,
})
}
}
}
#[tokio::test] #[tokio::test]
async fn test_rule_based_evaluator_success() { async fn test_rule_based_evaluator_success() {
+32
View File
@@ -1405,6 +1405,38 @@ impl ExtensionManager {
Ok(()) Ok(())
} }
#[allow(dead_code)] // Used by upcoming hot-activation flow
async fn install_bundled_channel_from_artifacts(
&self,
name: &str,
) -> Result<InstallResult, ExtensionError> {
// Check if already installed
let channel_wasm = self.wasm_channels_dir.join(format!("{}.wasm", name));
if channel_wasm.exists() {
return Err(ExtensionError::AlreadyInstalled(name.to_string()));
}
crate::channels::wasm::install_bundled_channel(name, &self.wasm_channels_dir, false)
.await
.map_err(ExtensionError::InstallFailed)?;
tracing::info!(
"Installed bundled channel '{}' to {}",
name,
self.wasm_channels_dir.display()
);
Ok(InstallResult {
name: name.to_string(),
kind: ExtensionKind::WasmChannel,
message: format!(
"Channel '{}' installed. \
Run tool_auth('{}') to configure authentication, then activate.",
name, name,
),
})
}
/// Install a WASM extension from local build artifacts (WasmBuildable source). /// Install a WASM extension from local build artifacts (WasmBuildable source).
/// ///
/// Resolves the build directory (relative to `CARGO_MANIFEST_DIR` or absolute), /// Resolves the build directory (relative to `CARGO_MANIFEST_DIR` or absolute),
-3
View File
@@ -241,9 +241,6 @@ impl Store {
tool_output_stash: std::sync::Arc::new(tokio::sync::RwLock::new( tool_output_stash: std::sync::Arc::new(tokio::sync::RwLock::new(
std::collections::HashMap::new(), std::collections::HashMap::new(),
)), )),
// TODO(#661): persist user_timezone in agent_jobs table so
// background/routine jobs retain the session's timezone context.
user_timezone: "UTC".to_string(),
})) }))
} }
None => Ok(None), None => Ok(None),
-1
View File
@@ -66,7 +66,6 @@ pub mod service;
pub mod settings; pub mod settings;
pub mod setup; pub mod setup;
pub mod skills; pub mod skills;
pub mod timezone;
pub mod tools; pub mod tools;
pub mod tracing_fmt; pub mod tracing_fmt;
pub mod transcription; pub mod transcription;
+4 -21
View File
@@ -58,10 +58,8 @@ pub fn create_llm_provider(
config: &LlmConfig, config: &LlmConfig,
session: Arc<SessionManager>, session: Arc<SessionManager>,
) -> Result<Arc<dyn LlmProvider>, LlmError> { ) -> Result<Arc<dyn LlmProvider>, LlmError> {
let timeout = config.request_timeout_secs;
if config.backend == "nearai" || config.backend == "near_ai" || config.backend == "near" { if config.backend == "nearai" || config.backend == "near_ai" || config.backend == "near" {
return create_llm_provider_with_config(&config.nearai, session, timeout); return create_llm_provider_with_config(&config.nearai, session);
} }
let reg_config = config let reg_config = config
@@ -81,7 +79,6 @@ pub fn create_llm_provider(
pub fn create_llm_provider_with_config( pub fn create_llm_provider_with_config(
config: &NearAiConfig, config: &NearAiConfig,
session: Arc<SessionManager>, session: Arc<SessionManager>,
request_timeout_secs: u64,
) -> Result<Arc<dyn LlmProvider>, LlmError> { ) -> Result<Arc<dyn LlmProvider>, LlmError> {
let auth_mode = if config.api_key.is_some() { let auth_mode = if config.api_key.is_some() {
"API key" "API key"
@@ -92,14 +89,9 @@ pub fn create_llm_provider_with_config(
model = %config.model, model = %config.model,
base_url = %config.base_url, base_url = %config.base_url,
auth = auth_mode, auth = auth_mode,
timeout_secs = request_timeout_secs,
"Using NEAR AI (Chat Completions API)" "Using NEAR AI (Chat Completions API)"
); );
Ok(Arc::new(NearAiChatProvider::new_with_timeout( Ok(Arc::new(NearAiChatProvider::new(config.clone(), session)?))
config.clone(),
session,
request_timeout_secs,
)?))
} }
/// Create a provider from a registry-resolved config. /// Create a provider from a registry-resolved config.
@@ -373,11 +365,7 @@ pub fn build_provider_chain(
let llm: Arc<dyn LlmProvider> = if let Some(ref cheap_model) = config.nearai.cheap_model { let llm: Arc<dyn LlmProvider> = if let Some(ref cheap_model) = config.nearai.cheap_model {
let mut cheap_config = config.nearai.clone(); let mut cheap_config = config.nearai.clone();
cheap_config.model = cheap_model.clone(); cheap_config.model = cheap_model.clone();
let cheap = create_llm_provider_with_config( let cheap = create_llm_provider_with_config(&cheap_config, session.clone())?;
&cheap_config,
session.clone(),
config.request_timeout_secs,
)?;
let cheap: Arc<dyn LlmProvider> = if retry_config.max_retries > 0 { let cheap: Arc<dyn LlmProvider> = if retry_config.max_retries > 0 {
Arc::new(RetryProvider::new(cheap, retry_config.clone())) Arc::new(RetryProvider::new(cheap, retry_config.clone()))
} else { } else {
@@ -409,11 +397,7 @@ pub fn build_provider_chain(
} }
let mut fallback_config = config.nearai.clone(); let mut fallback_config = config.nearai.clone();
fallback_config.model = fallback_model.clone(); fallback_config.model = fallback_model.clone();
let fallback = create_llm_provider_with_config( let fallback = create_llm_provider_with_config(&fallback_config, session.clone())?;
&fallback_config,
session.clone(),
config.request_timeout_secs,
)?;
tracing::info!( tracing::info!(
primary = %llm.model_name(), primary = %llm.model_name(),
fallback = %fallback.model_name(), fallback = %fallback.model_name(),
@@ -519,7 +503,6 @@ mod tests {
session: SessionConfig::default(), session: SessionConfig::default(),
nearai: test_nearai_config(), nearai: test_nearai_config(),
provider: None, provider: None,
request_timeout_secs: 120,
} }
} }
+4 -15
View File
@@ -58,28 +58,17 @@ impl NearAiChatProvider {
/// By default this enables tool-message flattening for compatibility with /// By default this enables tool-message flattening for compatibility with
/// providers that reject `role: "tool"` messages. /// providers that reject `role: "tool"` messages.
pub fn new(config: NearAiConfig, session: Arc<SessionManager>) -> Result<Self, LlmError> { pub fn new(config: NearAiConfig, session: Arc<SessionManager>) -> Result<Self, LlmError> {
Self::new_with_options(config, session, true, 120) Self::new_with_flatten(config, session, true)
} }
/// Create a new provider with a custom request timeout. /// Create a chat completions provider with configurable tool-message flattening.
pub fn new_with_timeout( pub fn new_with_flatten(
config: NearAiConfig,
session: Arc<SessionManager>,
request_timeout_secs: u64,
) -> Result<Self, LlmError> {
Self::new_with_options(config, session, true, request_timeout_secs)
}
/// Create a chat completions provider with configurable tool-message flattening
/// and request timeout.
pub fn new_with_options(
config: NearAiConfig, config: NearAiConfig,
session: Arc<SessionManager>, session: Arc<SessionManager>,
flatten_tool_messages: bool, flatten_tool_messages: bool,
request_timeout_secs: u64,
) -> Result<Self, LlmError> { ) -> Result<Self, LlmError> {
let client = Client::builder() let client = Client::builder()
.timeout(std::time::Duration::from_secs(request_timeout_secs)) .timeout(std::time::Duration::from_secs(120))
.build() .build()
.map_err(|e| LlmError::RequestFailed { .map_err(|e| LlmError::RequestFailed {
provider: "nearai_chat".to_string(), provider: "nearai_chat".to_string(),
+12 -2
View File
@@ -11,6 +11,7 @@ use crate::llm::{
ChatMessage, CompletionRequest, LlmProvider, Role, ToolCall, ToolCompletionRequest, ChatMessage, CompletionRequest, LlmProvider, Role, ToolCall, ToolCompletionRequest,
ToolDefinition, ToolDefinition,
}; };
use crate::safety::SafetyLayer;
/// Token the agent returns when it has nothing to say (e.g. in group chats). /// Token the agent returns when it has nothing to say (e.g. in group chats).
/// The dispatcher should check for this and suppress the message. /// The dispatcher should check for this and suppress the message.
@@ -342,6 +343,8 @@ pub struct RespondOutput {
/// Reasoning engine for the agent. /// Reasoning engine for the agent.
pub struct Reasoning { pub struct Reasoning {
llm: Arc<dyn LlmProvider>, llm: Arc<dyn LlmProvider>,
#[allow(dead_code)] // Will be used for sanitizing tool outputs
safety: Arc<SafetyLayer>,
/// Optional workspace for loading identity/system prompts. /// Optional workspace for loading identity/system prompts.
workspace_system_prompt: Option<String>, workspace_system_prompt: Option<String>,
/// Optional skill context block to inject into system prompt. /// Optional skill context block to inject into system prompt.
@@ -359,9 +362,10 @@ pub struct Reasoning {
impl Reasoning { impl Reasoning {
/// Create a new reasoning engine. /// Create a new reasoning engine.
pub fn new(llm: Arc<dyn LlmProvider>) -> Self { pub fn new(llm: Arc<dyn LlmProvider>, safety: Arc<SafetyLayer>) -> Self {
Self { Self {
llm, llm,
safety,
workspace_system_prompt: None, workspace_system_prompt: None,
skill_context: None, skill_context: None,
channel: None, channel: None,
@@ -2113,9 +2117,15 @@ That's my plan."#;
// ---- System prompt building tests (issue #565) ---- // ---- System prompt building tests (issue #565) ----
fn make_test_reasoning() -> Reasoning { fn make_test_reasoning() -> Reasoning {
use crate::config::SafetyConfig;
use crate::safety::SafetyLayer;
use crate::testing::StubLlm; use crate::testing::StubLlm;
let llm = Arc::new(StubLlm::new("test")); let llm = Arc::new(StubLlm::new("test"));
Reasoning::new(llm) let safety = Arc::new(SafetyLayer::new(&SafetyConfig {
max_output_length: 100_000,
injection_check_enabled: false,
}));
Reasoning::new(llm, safety)
} }
#[test] #[test]
+1 -2
View File
@@ -200,10 +200,9 @@ impl SessionManager {
let status = response.status(); let status = response.status();
let body = response.text().await.unwrap_or_default(); let body = response.text().await.unwrap_or_default();
let preview = crate::agent::truncate_for_preview(&body, 200);
Err(LlmError::SessionRenewalFailed { Err(LlmError::SessionRenewalFailed {
provider: "nearai".to_string(), provider: "nearai".to_string(),
reason: format!("Validation failed: HTTP {status}: {preview}"), reason: format!("Validation failed: HTTP {}: {}", status, body),
}) })
} }
+6 -23
View File
@@ -145,24 +145,6 @@ async fn async_main() -> anyhow::Result<()> {
} }
} }
// ── PID lock (prevent multiple instances) ────────────────────────
let _pid_lock = match ironclaw::bootstrap::PidLock::acquire() {
Ok(lock) => Some(lock),
Err(ironclaw::bootstrap::PidLockError::AlreadyRunning { pid }) => {
anyhow::bail!(
"Another IronClaw instance is already running (PID {}). \
If this is incorrect, remove the stale PID file: {}",
pid,
ironclaw::bootstrap::pid_lock_path().display()
);
}
Err(e) => {
eprintln!("Warning: Could not acquire PID lock: {}", e);
eprintln!("Continuing without PID lock protection.");
None
}
};
// ── Agent startup ────────────────────────────────────────────────── // ── Agent startup ──────────────────────────────────────────────────
// Enhanced first-run detection // Enhanced first-run detection
@@ -184,12 +166,13 @@ async fn async_main() -> anyhow::Result<()> {
let config = match Config::from_env_with_toml(toml_path).await { let config = match Config::from_env_with_toml(toml_path).await {
Ok(c) => c, Ok(c) => c,
Err(ironclaw::error::ConfigError::MissingRequired { key, hint }) => { Err(ironclaw::error::ConfigError::MissingRequired { key, hint }) => {
anyhow::bail!( eprintln!("Configuration error: Missing required setting '{}'", key);
"Configuration error: Missing required setting '{}'. {}. \ eprintln!(" {}", hint);
Run 'ironclaw onboard' to configure, or set the required environment variables.", eprintln!();
key, eprintln!(
hint "Run 'ironclaw onboard' to configure, or set the required environment variables."
); );
std::process::exit(1);
} }
Err(e) => return Err(e.into()), Err(e) => return Err(e.into()),
}; };
-53
View File
@@ -42,10 +42,6 @@ pub struct Settings {
#[serde(default)] #[serde(default)]
pub secrets_master_key_source: KeySource, pub secrets_master_key_source: KeySource,
/// Generated master key hex (env var mode only, written to .env by wizard).
#[serde(default, skip_serializing)]
pub secrets_master_key_hex: Option<String>,
// === Step 3: Inference Provider === // === Step 3: Inference Provider ===
/// LLM backend: "nearai", "anthropic", "openai", "ollama", "openai_compatible". /// LLM backend: "nearai", "anthropic", "openai", "ollama", "openai_compatible".
#[serde(default)] #[serde(default)]
@@ -295,18 +291,6 @@ pub struct HeartbeatSettings {
/// User ID to notify on heartbeat findings. /// User ID to notify on heartbeat findings.
#[serde(default)] #[serde(default)]
pub notify_user: Option<String>, pub notify_user: Option<String>,
/// Hour (0-23) when quiet hours start (heartbeat skipped).
#[serde(default)]
pub quiet_hours_start: Option<u32>,
/// Hour (0-23) when quiet hours end (heartbeat resumes).
#[serde(default)]
pub quiet_hours_end: Option<u32>,
/// Timezone for quiet hours evaluation (IANA name, e.g. "America/New_York").
#[serde(default)]
pub timezone: Option<String>,
} }
fn default_heartbeat_interval() -> u64 { fn default_heartbeat_interval() -> u64 {
@@ -320,9 +304,6 @@ impl Default for HeartbeatSettings {
interval_secs: default_heartbeat_interval(), interval_secs: default_heartbeat_interval(),
notify_channel: None, notify_channel: None,
notify_user: None, notify_user: None,
quiet_hours_start: None,
quiet_hours_end: None,
timezone: None,
} }
} }
} }
@@ -370,10 +351,6 @@ pub struct AgentSettings {
/// When true, skip tool approval checks entirely. For benchmarks/CI. /// When true, skip tool approval checks entirely. For benchmarks/CI.
#[serde(default)] #[serde(default)]
pub auto_approve_tools: bool, pub auto_approve_tools: bool,
/// Default timezone for new sessions (IANA name, e.g. "America/New_York").
#[serde(default = "default_timezone")]
pub default_timezone: String,
} }
fn default_agent_name() -> String { fn default_agent_name() -> String {
@@ -408,10 +385,6 @@ fn default_max_tool_iterations() -> usize {
50 50
} }
fn default_timezone() -> String {
"UTC".to_string()
}
fn default_true() -> bool { fn default_true() -> bool {
true true
} }
@@ -429,7 +402,6 @@ impl Default for AgentSettings {
session_idle_timeout_secs: default_session_idle_timeout(), session_idle_timeout_secs: default_session_idle_timeout(),
max_tool_iterations: default_max_tool_iterations(), max_tool_iterations: default_max_tool_iterations(),
auto_approve_tools: false, auto_approve_tools: false,
default_timezone: default_timezone(),
} }
} }
} }
@@ -1202,31 +1174,6 @@ mod tests {
assert_eq!(loaded.heartbeat.interval_secs, 900); assert_eq!(loaded.heartbeat.interval_secs, 900);
} }
/// Regression test: /model command must persist selected_model to TOML config.
/// Prior to the fix, `set_model()` only changed the in-memory provider and the
/// choice was lost on restart.
#[test]
fn toml_selected_model_update_persists() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("config.toml");
// Start with a config that has a different model.
let settings = Settings {
selected_model: Some("old-model".to_string()),
..Default::default()
};
settings.save_toml(&path).unwrap();
// Simulate what persist_selected_model does: load, update, save.
let mut loaded = Settings::load_toml(&path).unwrap().unwrap();
loaded.selected_model = Some("new-model".to_string());
loaded.save_toml(&path).unwrap();
// Verify the change survived a reload.
let reloaded = Settings::load_toml(&path).unwrap().unwrap();
assert_eq!(reloaded.selected_model, Some("new-model".to_string()));
}
#[test] #[test]
fn toml_missing_file_returns_none() { fn toml_missing_file_returns_none() {
let result = Settings::load_toml(std::path::Path::new("/tmp/nonexistent_config.toml")); let result = Settings::load_toml(std::path::Path::new("/tmp/nonexistent_config.toml"));
+12 -106
View File
@@ -769,28 +769,13 @@ impl SetupWizard {
print_success("Master key generated and stored in OS keychain"); print_success("Master key generated and stored in OS keychain");
} }
1 => { 1 => {
// Env var mode — generate key, init crypto, and persist to .env // Env var mode
print_info("Generate a key and add it to your environment:");
let key_hex = crate::secrets::keychain::generate_master_key_hex(); let key_hex = crate::secrets::keychain::generate_master_key_hex();
// Initialize crypto so subsequent wizard steps (channel setup,
// API key storage) can encrypt secrets immediately.
self.secrets_crypto = Some(Arc::new(
SecretsCrypto::new(SecretString::from(key_hex.clone()))
.map_err(|e| SetupError::Config(e.to_string()))?,
));
// Make visible to optional_env() for any subsequent config resolution.
crate::config::inject_single_var("SECRETS_MASTER_KEY", &key_hex);
// Store hex for write_bootstrap_env to persist to ~/.ironclaw/.env.
self.settings.secrets_master_key_hex = Some(key_hex.clone());
println!(); println!();
print_info("Master key generated and will be saved to ~/.ironclaw/.env"); println!(" export SECRETS_MASTER_KEY={}", key_hex);
println!(); println!();
println!(" SECRETS_MASTER_KEY={}", key_hex); print_info("Add this to your shell profile or .env file.");
println!();
print_info("You can also copy this to another .env file or CI secrets.");
self.settings.secrets_master_key_source = KeySource::Env; self.settings.secrets_master_key_source = KeySource::Env;
print_success("Configured for environment variable"); print_success("Configured for environment variable");
@@ -1037,11 +1022,10 @@ impl SetupWizard {
/// Anthropic OAuth setup: extract token from `claude login` credentials. /// Anthropic OAuth setup: extract token from `claude login` credentials.
async fn setup_anthropic_oauth(&mut self) -> Result<(), SetupError> { async fn setup_anthropic_oauth(&mut self) -> Result<(), SetupError> {
// Clear model only when switching providers (old model may be invalid) self.settings.llm_backend = Some("anthropic".to_string());
if self.settings.llm_backend.as_deref() != Some("anthropic") { if self.settings.selected_model.is_some() {
self.settings.selected_model = None; self.settings.selected_model = None;
} }
self.settings.llm_backend = Some("anthropic".to_string());
// Try to extract existing OAuth token from Claude Code credentials // Try to extract existing OAuth token from Claude Code credentials
if let Some(token) = crate::config::ClaudeCodeConfig::extract_oauth_token() { if let Some(token) = crate::config::ClaudeCodeConfig::extract_oauth_token() {
@@ -1135,11 +1119,10 @@ impl SetupWizard {
other => other, other => other,
}); });
// Clear model only when switching providers (old model may be invalid) self.settings.llm_backend = Some(backend.to_string());
if self.settings.llm_backend.as_deref() != Some(backend) { if self.settings.selected_model.is_some() {
self.settings.selected_model = None; self.settings.selected_model = None;
} }
self.settings.llm_backend = Some(backend.to_string());
// Check env var first // Check env var first
if let Ok(existing) = std::env::var(env_var) { if let Ok(existing) = std::env::var(env_var) {
@@ -1198,11 +1181,10 @@ impl SetupWizard {
&mut self, &mut self,
def: &crate::llm::ProviderDefinition, def: &crate::llm::ProviderDefinition,
) -> Result<(), SetupError> { ) -> Result<(), SetupError> {
// Clear model only when switching providers (old model may be invalid) self.settings.llm_backend = Some(def.id.clone());
if self.settings.llm_backend.as_deref() != Some(&def.id) { if self.settings.selected_model.is_some() {
self.settings.selected_model = None; self.settings.selected_model = None;
} }
self.settings.llm_backend = Some(def.id.clone());
let default_url = self let default_url = self
.settings .settings
@@ -1237,11 +1219,10 @@ impl SetupWizard {
secret_name: &str, secret_name: &str,
display_name: &str, display_name: &str,
) -> Result<(), SetupError> { ) -> Result<(), SetupError> {
// Clear model only when switching providers (old model may be invalid) self.settings.llm_backend = Some(backend_id.to_string());
if self.settings.llm_backend.as_deref() != Some(backend_id) { if self.settings.selected_model.is_some() {
self.settings.selected_model = None; self.settings.selected_model = None;
} }
self.settings.llm_backend = Some(backend_id.to_string());
let existing_url = self let existing_url = self
.settings .settings
@@ -1495,7 +1476,6 @@ impl SetupWizard {
smart_routing_cascade: true, smart_routing_cascade: true,
}, },
provider: None, provider: None,
request_timeout_secs: 120,
}; };
match create_llm_provider(&config, session) { match create_llm_provider(&config, session) {
@@ -2344,12 +2324,6 @@ impl SetupWizard {
env_vars.push(("NEARAI_API_KEY".to_string(), api_key)); env_vars.push(("NEARAI_API_KEY".to_string(), api_key));
} }
// Secrets master key (env var mode): write to .env so it's available
// on next startup before the DB is connected.
if let Some(ref key_hex) = self.settings.secrets_master_key_hex {
env_vars.push(("SECRETS_MASTER_KEY".to_string(), key_hex.clone()));
}
// Always write ONBOARD_COMPLETED so that check_onboard_needed() // Always write ONBOARD_COMPLETED so that check_onboard_needed()
// (which runs before the DB is connected) knows to skip re-onboarding. // (which runs before the DB is connected) knows to skip re-onboarding.
if self.settings.onboard_completed { if self.settings.onboard_completed {
@@ -3525,48 +3499,6 @@ mod tests {
} }
} }
/// Regression test for #600: re-running provider setup for the same backend
/// must NOT clear selected_model. Only switching to a different backend should.
#[test]
fn test_same_provider_preserves_selected_model() {
let mut wizard = SetupWizard::new();
wizard.settings.llm_backend = Some("ollama".to_string());
wizard.settings.selected_model = Some("llama3".to_string());
// Simulate re-entering the same provider -- model should survive
// (This is the check that each setup_* function now performs)
if wizard.settings.llm_backend.as_deref() != Some("ollama") {
wizard.settings.selected_model = None;
}
wizard.settings.llm_backend = Some("ollama".to_string());
assert_eq!(
wizard.settings.selected_model.as_deref(),
Some("llama3"),
"model should be preserved when re-selecting the same provider"
);
}
/// Regression test for #600: switching to a different provider must clear
/// selected_model since the old model may not be valid for the new backend.
#[test]
fn test_different_provider_clears_selected_model() {
let mut wizard = SetupWizard::new();
wizard.settings.llm_backend = Some("ollama".to_string());
wizard.settings.selected_model = Some("llama3".to_string());
// Simulate switching to a different provider -- model should be cleared
if wizard.settings.llm_backend.as_deref() != Some("openai") {
wizard.settings.selected_model = None;
}
wizard.settings.llm_backend = Some("openai".to_string());
assert!(
wizard.settings.selected_model.is_none(),
"model should be cleared when switching providers"
);
}
#[tokio::test] #[tokio::test]
async fn test_run_provider_setup_no_setup_hint() { async fn test_run_provider_setup_no_setup_hint() {
// A provider with setup: None should not error. It should set the // A provider with setup: None should not error. It should set the
@@ -3604,30 +3536,4 @@ mod tests {
"backend should be set even without setup hint" "backend should be set even without setup hint"
); );
} }
/// Regression test for #666: env-var security option must initialize
/// secrets_crypto so subsequent steps can encrypt API keys.
#[test]
fn test_env_var_security_initializes_crypto() {
use crate::secrets::SecretsCrypto;
use secrecy::SecretString;
// Simulate what option 1 in step_security() does after the fix:
let key_hex = crate::secrets::keychain::generate_master_key_hex();
// The fix: create SecretsCrypto from the generated key.
// Before the fix, this was skipped, leaving secrets_crypto = None.
let crypto = SecretsCrypto::new(SecretString::from(key_hex.clone()));
assert!(
crypto.is_ok(),
"generated key hex must produce valid SecretsCrypto"
);
// Verify the key is stored for bootstrap env persistence.
let settings = Settings {
secrets_master_key_hex: Some(key_hex),
..Settings::default()
};
assert!(settings.secrets_master_key_hex.is_some());
}
} }
-1
View File
@@ -1009,7 +1009,6 @@ mod tests {
enabled: true, enabled: true,
trigger: Trigger::Cron { trigger: Trigger::Cron {
schedule: "0 * * * *".to_string(), schedule: "0 * * * *".to_string(),
timezone: None,
}, },
action: RoutineAction::Lightweight { action: RoutineAction::Lightweight {
prompt: "Check status".to_string(), prompt: "Check status".to_string(),
-110
View File
@@ -1,110 +0,0 @@
//! Timezone resolution and utilities.
use chrono::{DateTime, NaiveDate, Utc};
use chrono_tz::Tz;
/// Resolve the effective timezone from a priority chain.
///
/// Priority: client_tz > user_setting > config_default > UTC
pub fn resolve_timezone(
client_tz: Option<&str>,
user_setting: Option<&str>,
config_default: &str,
) -> Tz {
// Try each in priority order, skipping invalid values
for candidate in [client_tz, user_setting, Some(config_default)] {
if let Some(tz) = candidate.and_then(parse_timezone) {
return tz;
}
}
Tz::UTC
}
/// Parse a timezone string (IANA name) into a `Tz`.
pub fn parse_timezone(s: &str) -> Option<Tz> {
s.parse::<Tz>().ok()
}
/// Get today's date in the given timezone.
pub fn today_in_tz(tz: Tz) -> NaiveDate {
Utc::now().with_timezone(&tz).date_naive()
}
/// Get the current time in the given timezone.
pub fn now_in_tz(tz: Tz) -> DateTime<Tz> {
Utc::now().with_timezone(&tz)
}
/// Detect the system's timezone, falling back to UTC.
pub fn detect_system_timezone() -> Tz {
iana_time_zone::get_timezone()
.ok()
.and_then(|s| parse_timezone(&s))
.unwrap_or(Tz::UTC)
}
#[cfg(test)]
mod tests {
use chrono::Datelike;
use super::*;
#[test]
fn test_resolve_client_wins() {
let tz = resolve_timezone(Some("America/New_York"), Some("Europe/London"), "UTC");
assert_eq!(tz, chrono_tz::America::New_York);
}
#[test]
fn test_resolve_user_setting_fallback() {
let tz = resolve_timezone(None, Some("Europe/London"), "UTC");
assert_eq!(tz, chrono_tz::Europe::London);
}
#[test]
fn test_resolve_config_fallback() {
let tz = resolve_timezone(None, None, "Asia/Tokyo");
assert_eq!(tz, chrono_tz::Asia::Tokyo);
}
#[test]
fn test_resolve_all_none_utc() {
let tz = resolve_timezone(None, None, "UTC");
assert_eq!(tz, Tz::UTC);
}
#[test]
fn test_resolve_invalid_client_skipped() {
let tz = resolve_timezone(Some("Fake/Zone"), Some("Europe/London"), "UTC");
assert_eq!(tz, chrono_tz::Europe::London);
}
#[test]
fn test_parse_valid() {
assert_eq!(
parse_timezone("America/Chicago"),
Some(chrono_tz::America::Chicago)
);
}
#[test]
fn test_parse_invalid() {
assert_eq!(parse_timezone("Fake/Zone"), None);
}
#[test]
fn test_detect_system_tz() {
// Should always return a valid Tz (at minimum UTC)
let tz = detect_system_timezone();
let _ = now_in_tz(tz); // Should not panic
}
#[test]
fn test_today_in_tz_returns_valid_date() {
let date = today_in_tz(Tz::UTC);
// Verify it returns a valid date (year, month, day are all positive)
assert!(date.year() > 0);
assert!((1..=12).contains(&date.month()));
assert!((1..=31).contains(&date.day()));
}
}
+16 -4
View File
@@ -43,6 +43,7 @@ use crate::error::ToolError as AgentToolError;
use crate::llm::{ use crate::llm::{
ChatMessage, LlmProvider, Reasoning, ReasoningContext, RespondResult, ToolDefinition, ChatMessage, LlmProvider, Reasoning, ReasoningContext, RespondResult, ToolDefinition,
}; };
use crate::safety::SafetyLayer;
use crate::tools::ToolRegistry; use crate::tools::ToolRegistry;
use crate::tools::tool::{ApprovalRequirement, Tool, ToolError, ToolOutput}; use crate::tools::tool::{ApprovalRequirement, Tool, ToolError, ToolOutput};
@@ -250,18 +251,29 @@ pub trait SoftwareBuilder: Send + Sync {
pub struct LlmSoftwareBuilder { pub struct LlmSoftwareBuilder {
config: BuilderConfig, config: BuilderConfig,
llm: Arc<dyn LlmProvider>, llm: Arc<dyn LlmProvider>,
safety: Arc<SafetyLayer>,
tools: Arc<ToolRegistry>, tools: Arc<ToolRegistry>,
} }
impl LlmSoftwareBuilder { impl LlmSoftwareBuilder {
/// Create a new LLM-based software builder. /// Create a new LLM-based software builder.
pub fn new(config: BuilderConfig, llm: Arc<dyn LlmProvider>, tools: Arc<ToolRegistry>) -> Self { pub fn new(
config: BuilderConfig,
llm: Arc<dyn LlmProvider>,
safety: Arc<SafetyLayer>,
tools: Arc<ToolRegistry>,
) -> Self {
// Ensure build directory exists // Ensure build directory exists
if let Err(e) = std::fs::create_dir_all(&config.build_dir) { if let Err(e) = std::fs::create_dir_all(&config.build_dir) {
tracing::warn!("Failed to create build directory: {}", e); tracing::warn!("Failed to create build directory: {}", e);
} }
Self { config, llm, tools } Self {
config,
llm,
safety,
tools,
}
} }
/// Get the build tools available for the build loop. /// Get the build tools available for the build loop.
@@ -509,7 +521,7 @@ Create alongside the .wasm file to grant capabilities:
let mut iteration = 0; let mut iteration = 0;
// Create reasoning engine // Create reasoning engine
let reasoning = Reasoning::new(self.llm.clone()); let reasoning = Reasoning::new(self.llm.clone(), self.safety.clone());
// Build initial context // Build initial context
let tool_defs = self.get_build_tools().await; let tool_defs = self.get_build_tools().await;
@@ -810,7 +822,7 @@ Create alongside the .wasm file to grant capabilities:
impl SoftwareBuilder for LlmSoftwareBuilder { impl SoftwareBuilder for LlmSoftwareBuilder {
async fn analyze(&self, description: &str) -> Result<BuildRequirement, AgentToolError> { async fn analyze(&self, description: &str) -> Result<BuildRequirement, AgentToolError> {
// Use LLM to parse the description // Use LLM to parse the description
let reasoning = Reasoning::new(self.llm.clone()); let reasoning = Reasoning::new(self.llm.clone(), self.safety.clone());
let prompt = format!( let prompt = format!(
r#"Analyze this software requirement and extract structured information. r#"Analyze this software requirement and extract structured information.
+4 -5
View File
@@ -172,7 +172,7 @@ impl Tool for MemoryWriteTool {
async fn execute( async fn execute(
&self, &self,
params: serde_json::Value, params: serde_json::Value,
ctx: &JobContext, _ctx: &JobContext,
) -> Result<ToolOutput, ToolError> { ) -> Result<ToolOutput, ToolError> {
let start = std::time::Instant::now(); let start = std::time::Instant::now();
@@ -239,12 +239,11 @@ impl Tool for MemoryWriteTool {
paths::MEMORY.to_string() paths::MEMORY.to_string()
} }
"daily_log" => { "daily_log" => {
let tz = crate::timezone::parse_timezone(&ctx.user_timezone)
.unwrap_or(chrono_tz::Tz::UTC);
self.workspace self.workspace
.append_daily_log_tz(content, tz) .append_daily_log(content)
.await .await
.map_err(|e| ToolError::ExecutionFailed(format!("Write failed: {}", e)))? .map_err(|e| ToolError::ExecutionFailed(format!("Write failed: {}", e)))?;
format!("daily/{}.md", chrono::Utc::now().format("%Y-%m-%d"))
} }
"heartbeat" => { "heartbeat" => {
if append { if append {
+48 -182
View File
@@ -105,47 +105,42 @@ impl Tool for MessageTool {
async fn execute( async fn execute(
&self, &self,
params: serde_json::Value, params: serde_json::Value,
ctx: &JobContext, _ctx: &JobContext,
) -> Result<ToolOutput, ToolError> { ) -> Result<ToolOutput, ToolError> {
let start = std::time::Instant::now(); let start = std::time::Instant::now();
let content = require_str(&params, "content")?; let content = require_str(&params, "content")?;
// Get channel: use param → conversation default → job metadata → None (broadcast all) // Get channel: use param or fall back to default
let channel: Option<String> = let channel = if let Some(c) = params.get("channel").and_then(|v| v.as_str()) {
if let Some(c) = params.get("channel").and_then(|v| v.as_str()) { c.to_string()
Some(c.to_string()) } else {
} else if let Some(c) = self self.default_channel
.default_channel
.read() .read()
.unwrap_or_else(|e| e.into_inner()) .unwrap_or_else(|e| e.into_inner())
.clone() .clone()
{ .ok_or_else(|| {
Some(c) ToolError::ExecutionFailed(
} else { "No channel specified and no active conversation. Provide channel parameter."
ctx.metadata .to_string(),
.get("notify_channel") )
.and_then(|v| v.as_str()) })?
.map(|c| c.to_string()) };
};
// Get target: use param → conversation default → job metadata // Get target: use param or fall back to default
let target = if let Some(t) = params.get("target").and_then(|v| v.as_str()) { let target = if let Some(t) = params.get("target").and_then(|v| v.as_str()) {
t.to_string() t.to_string()
} else if let Some(t) = self
.default_target
.read()
.unwrap_or_else(|e| e.into_inner())
.clone()
{
t
} else if let Some(t) = ctx.metadata.get("notify_user").and_then(|v| v.as_str()) {
t.to_string()
} else { } else {
return Err(ToolError::ExecutionFailed( self.default_target
"No target specified and no active conversation. Provide target parameter." .read()
.to_string(), .unwrap_or_else(|e| e.into_inner())
)); .clone()
.ok_or_else(|| {
ToolError::ExecutionFailed(
"No target specified and no active conversation. Provide target parameter."
.to_string(),
)
})?
}; };
let attachments: Vec<String> = match params.get("attachments") { let attachments: Vec<String> = match params.get("attachments") {
@@ -186,80 +181,37 @@ impl Tool for MessageTool {
response = response.with_attachments(attachments); response = response.with_attachments(attachments);
} }
if let Some(ref channel) = channel { match self
// Send to a specific channel .channel_manager
match self .broadcast(&channel, &target, response)
.channel_manager .await
.broadcast(channel, &target, response) {
.await Ok(()) => {
{
Ok(()) => {
tracing::info!(
message_sent = true,
channel = %channel,
target = %target,
attachments = attachment_count,
"Message sent via message tool"
);
let msg = format!("Sent message to {}:{}", channel, target);
Ok(ToolOutput::text(msg, start.elapsed()))
}
Err(e) => {
let available = self.channel_manager.channel_names().await.join(", ");
let err_msg = if available.is_empty() {
format!(
"Failed to send to {}:{}: {}. No channels connected.",
channel, target, e
)
} else {
format!(
"Failed to send to {}:{}. Available channels: {}. Error: {}",
channel, target, available, e
)
};
Err(ToolError::ExecutionFailed(err_msg))
}
}
} else {
// No channel specified — broadcast to all channels (routine with notify.channel = None)
let results = self.channel_manager.broadcast_all(&target, response).await;
let mut succeeded = Vec::new();
let mut failed: Vec<&str> = Vec::new();
for (ch, result) in &results {
match result {
Ok(()) => succeeded.push(ch.as_str()),
Err(e) => {
tracing::warn!(
channel = %ch,
target = %target,
"broadcast_all: channel failed: {}", e
);
failed.push(ch.as_str());
}
}
}
if succeeded.is_empty() {
let err_msg = if failed.is_empty() {
"No channels connected.".to_string()
} else {
format!("All channels failed: {}", failed.join(", "))
};
Err(ToolError::ExecutionFailed(err_msg))
} else {
tracing::info!( tracing::info!(
message_sent = true, message_sent = true,
channels = ?succeeded, channel = %channel,
target = %target, target = %target,
attachments = attachment_count, attachments = attachment_count,
"Message broadcast via message tool" "Message sent via message tool"
);
let msg = format!(
"Broadcast message to {} (target: {})",
succeeded.join(", "),
target
); );
let msg = format!("Sent message to {}:{}", channel, target);
Ok(ToolOutput::text(msg, start.elapsed())) Ok(ToolOutput::text(msg, start.elapsed()))
} }
Err(e) => {
let available = self.channel_manager.channel_names().await.join(", ");
let err_msg = if available.is_empty() {
format!(
"Failed to send to {}:{}: {}. No channels connected.",
channel, target, e
)
} else {
format!(
"Failed to send to {}:{}. Available channels: {}. Error: {}",
channel, target, available, e
)
};
Err(ToolError::ExecutionFailed(err_msg))
}
} }
} }
@@ -624,90 +576,4 @@ mod tests {
ApprovalRequirement::Never, ApprovalRequirement::Never,
); );
} }
#[tokio::test]
async fn message_tool_falls_back_to_job_metadata() {
// Regression: when no conversation context is set (e.g. routine full-job),
// the message tool should fall back to notify_channel/notify_user from
// JobContext metadata instead of returning "No target specified".
let tool = MessageTool::new(Arc::new(ChannelManager::new()));
let mut ctx = crate::context::JobContext::new("routine-job", "price alert");
ctx.metadata = serde_json::json!({
"notify_channel": "telegram",
"notify_user": "123456789",
});
// No set_context called — simulates a routine full-job worker
let result = tool
.execute(serde_json::json!({"content": "NEAR price is $5"}), &ctx)
.await;
// Should fail at channel broadcast (no real channel), NOT at
// "No target specified and no active conversation"
assert!(result.is_err());
let err = result.unwrap_err().to_string();
assert!(
!err.contains("No target specified"),
"Should not get 'No target specified' when metadata has notify_user, got: {}",
err
);
assert!(
!err.contains("No channel specified"),
"Should not get 'No channel specified' when metadata has notify_channel, got: {}",
err
);
}
#[tokio::test]
async fn message_tool_no_metadata_still_errors() {
// When neither conversation context nor metadata is set, should still
// return a clear error (target resolution fails).
let tool = MessageTool::new(Arc::new(ChannelManager::new()));
let ctx = crate::context::JobContext::new("orphan-job", "no notify config");
let result = tool
.execute(serde_json::json!({"content": "hello"}), &ctx)
.await;
assert!(result.is_err());
let err = result.unwrap_err().to_string();
assert!(
err.contains("No target specified"),
"Expected 'No target specified' error, got: {}",
err
);
}
#[tokio::test]
async fn message_tool_broadcasts_all_when_no_channel() {
// Regression: when notify.channel is None but notify_user is set,
// the message tool should attempt broadcast_all instead of erroring
// with "No channel specified".
let tool = MessageTool::new(Arc::new(ChannelManager::new()));
let mut ctx = crate::context::JobContext::new("routine-job", "price alert");
ctx.metadata = serde_json::json!({
"notify_user": "123456789",
});
let result = tool
.execute(serde_json::json!({"content": "NEAR price is $5"}), &ctx)
.await;
// Should fail because no channels are registered (empty ChannelManager),
// NOT because "No channel specified".
assert!(result.is_err());
let err = result.unwrap_err().to_string();
assert!(
!err.contains("No channel specified"),
"Should not get 'No channel specified' when broadcasting, got: {}",
err
);
assert!(
err.contains("No channels connected") || err.contains("All channels failed"),
"Expected channel delivery error, got: {}",
err
);
}
} }
+10 -68
View File
@@ -107,10 +107,6 @@ impl Tool for RoutineCreateTool {
"notify_user": { "notify_user": {
"type": "string", "type": "string",
"description": "User/target to notify (e.g. username, chat ID). Defaults to 'default'." "description": "User/target to notify (e.g. username, chat ID). Defaults to 'default'."
},
"timezone": {
"type": "string",
"description": "IANA timezone for cron schedule evaluation (e.g. 'America/New_York'). Defaults to UTC."
} }
}, },
"required": ["name", "trigger_type", "prompt"] "required": ["name", "trigger_type", "prompt"]
@@ -147,26 +143,12 @@ impl Tool for RoutineCreateTool {
"cron trigger requires 'schedule'".to_string(), "cron trigger requires 'schedule'".to_string(),
) )
})?; })?;
let timezone = params
.get("timezone")
.and_then(|v| v.as_str())
.map(|tz| {
crate::timezone::parse_timezone(tz)
.map(|_| tz.to_string())
.ok_or_else(|| {
ToolError::InvalidParameters(format!(
"invalid IANA timezone: '{tz}'"
))
})
})
.transpose()?;
// Validate cron expression // Validate cron expression
next_cron_fire(schedule, timezone.as_deref()).map_err(|e| { next_cron_fire(schedule).map_err(|e| {
ToolError::InvalidParameters(format!("invalid cron schedule: {e}")) ToolError::InvalidParameters(format!("invalid cron schedule: {e}"))
})?; })?;
Trigger::Cron { Trigger::Cron {
schedule: schedule.to_string(), schedule: schedule.to_string(),
timezone,
} }
} }
"event" => { "event" => {
@@ -246,12 +228,8 @@ impl Tool for RoutineCreateTool {
.unwrap_or(300); .unwrap_or(300);
// Compute next fire time for cron // Compute next fire time for cron
let next_fire = if let Trigger::Cron { let next_fire = if let Trigger::Cron { ref schedule } = trigger {
ref schedule, next_cron_fire(schedule).unwrap_or(None)
ref timezone,
} = trigger
{
next_cron_fire(schedule, timezone.as_deref()).unwrap_or(None)
} else { } else {
None None
}; };
@@ -434,10 +412,6 @@ impl Tool for RoutineUpdateTool {
"type": "string", "type": "string",
"description": "New cron schedule (for cron triggers)" "description": "New cron schedule (for cron triggers)"
}, },
"timezone": {
"type": "string",
"description": "IANA timezone for cron schedule (e.g. 'America/New_York'). Only valid for cron triggers."
},
"description": { "description": {
"type": "string", "type": "string",
"description": "New description" "description": "New description"
@@ -479,47 +453,15 @@ impl Tool for RoutineUpdateTool {
} }
} }
// Validate timezone param if provided if let Some(schedule) = params.get("schedule").and_then(|v| v.as_str()) {
let new_timezone = params // Validate
.get("timezone") next_cron_fire(schedule)
.and_then(|v| v.as_str()) .map_err(|e| ToolError::InvalidParameters(format!("invalid cron schedule: {e}")))?;
.map(|tz| {
crate::timezone::parse_timezone(tz)
.map(|_| tz.to_string())
.ok_or_else(|| {
ToolError::InvalidParameters(format!("invalid IANA timezone: '{tz}'"))
})
})
.transpose()?;
let new_schedule = params.get("schedule").and_then(|v| v.as_str()); routine.trigger = Trigger::Cron {
schedule: schedule.to_string(),
if new_schedule.is_some() || new_timezone.is_some() {
// Extract existing cron fields (cloned to avoid borrow conflict)
let existing_cron = match &routine.trigger {
Trigger::Cron { schedule, timezone } => Some((schedule.clone(), timezone.clone())),
_ => None,
}; };
routine.next_fire_at = next_cron_fire(schedule).unwrap_or(None);
if let Some((old_schedule, old_tz)) = existing_cron {
let effective_schedule = new_schedule.unwrap_or(&old_schedule);
let effective_tz = new_timezone.or(old_tz);
// Validate
next_cron_fire(effective_schedule, effective_tz.as_deref()).map_err(|e| {
ToolError::InvalidParameters(format!("invalid cron schedule: {e}"))
})?;
routine.trigger = Trigger::Cron {
schedule: effective_schedule.to_string(),
timezone: effective_tz.clone(),
};
routine.next_fire_at =
next_cron_fire(effective_schedule, effective_tz.as_deref()).unwrap_or(None);
} else {
return Err(ToolError::InvalidParameters(
"Cannot update schedule or timezone on a non-cron routine.".to_string(),
));
}
} }
self.store self.store
+2 -46
View File
@@ -48,7 +48,7 @@ impl Tool for TimeTool {
async fn execute( async fn execute(
&self, &self,
params: serde_json::Value, params: serde_json::Value,
ctx: &JobContext, _ctx: &JobContext,
) -> Result<ToolOutput, ToolError> { ) -> Result<ToolOutput, ToolError> {
let start = std::time::Instant::now(); let start = std::time::Instant::now();
@@ -57,15 +57,10 @@ impl Tool for TimeTool {
let result = match operation { let result = match operation {
"now" => { "now" => {
let now = Utc::now(); let now = Utc::now();
let tz =
crate::timezone::parse_timezone(&ctx.user_timezone).unwrap_or(chrono_tz::UTC);
let local = now.with_timezone(&tz);
serde_json::json!({ serde_json::json!({
"iso": now.to_rfc3339(), "iso": now.to_rfc3339(),
"unix": now.timestamp(), "unix": now.timestamp(),
"unix_millis": now.timestamp_millis(), "unix_millis": now.timestamp_millis()
"local_iso": local.to_rfc3339(),
"timezone": tz.name()
}) })
} }
"parse" => { "parse" => {
@@ -117,42 +112,3 @@ impl Tool for TimeTool {
false // Internal tool, no external data false // Internal tool, no external data
} }
} }
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn test_now_includes_local_time_when_timezone_set() {
let tool = TimeTool;
let mut ctx = JobContext::with_user("test", "chat", "test");
ctx.user_timezone = "America/New_York".to_string();
let output = tool
.execute(serde_json::json!({"operation": "now"}), &ctx)
.await
.expect("execute");
assert!(
output.result.get("local_iso").is_some(),
"should have local_iso"
);
assert_eq!(
output.result["timezone"].as_str(),
Some("America/New_York"),
"should report timezone"
);
}
#[tokio::test]
async fn test_now_includes_utc_timezone_by_default() {
let tool = TimeTool;
let ctx = JobContext::with_user("test", "chat", "test");
// Default user_timezone is "UTC" which is a valid IANA timezone
let output = tool
.execute(serde_json::json!({"operation": "now"}), &ctx)
.await
.expect("execute");
assert!(output.result.get("iso").is_some(), "should have iso");
assert_eq!(output.result["timezone"].as_str(), Some("UTC"));
}
}
+2 -123
View File
@@ -261,9 +261,9 @@ impl McpClient {
if !response.status().is_success() { if !response.status().is_success() {
let status = response.status(); let status = response.status();
let body = response.text().await.unwrap_or_default(); let body = response.text().await.unwrap_or_default();
let preview = sanitize_error_body(&body);
return Err(ToolError::ExternalService(format!( return Err(ToolError::ExternalService(format!(
"MCP server returned status: {status} - {preview}", "MCP server returned status: {} - {}",
status, body
))); )));
} }
@@ -548,58 +548,6 @@ impl Tool for McpToolWrapper {
} }
} }
/// Sanitize an HTTP error response body for safe display.
///
/// Detects full HTML error pages (containing `<html` or `<!DOCTYPE`) and
/// strips all tags, collapsing whitespace. Non-HTML bodies are left
/// intact. In both cases the result is truncated to 200 *characters*
/// (char-boundary safe) so that large payloads don't bloat error messages.
///
/// See #263 — raw HTML error pages were propagating through the error
/// chain into the web UI, causing a white screen.
fn sanitize_error_body(body: &str) -> String {
const MAX_CHARS: usize = 200;
// Only strip tags when the body looks like a full HTML document.
// Plain text that happens to contain `<` / `>` (e.g. log lines,
// comparison expressions) is left untouched.
let lower = body.to_ascii_lowercase();
let is_html_document = lower.contains("<html") || lower.contains("<!doctype");
let text = if is_html_document {
let stripped = body
.chars()
.fold((String::new(), false), |(mut out, in_tag), c| {
if c == '<' {
(out, true)
} else if c == '>' {
(out, false)
} else if !in_tag {
out.push(c);
(out, false)
} else {
(out, true)
}
})
.0;
stripped.split_whitespace().collect::<Vec<_>>().join(" ")
} else {
body.to_string()
};
// Truncate at a char boundary (safe for multi-byte UTF-8).
if text.chars().count() > MAX_CHARS {
let byte_offset = text
.char_indices()
.nth(MAX_CHARS)
.map(|(i, _)| i)
.unwrap_or(text.len());
format!("{}... ({} bytes total)", &text[..byte_offset], body.len())
} else {
text
}
}
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
@@ -792,73 +740,4 @@ mod tests {
}; };
assert!(!tool.requires_approval()); assert!(!tool.requires_approval());
} }
// Regression tests for #263: HTML error bodies must not propagate raw
// markup through the error chain into the web UI.
#[test]
fn test_sanitize_error_body_strips_html_tags() {
let html =
r#"<!DOCTYPE html><html><body><h1>422 Error</h1><p>Invalid token</p></body></html>"#;
let result = sanitize_error_body(html);
assert!(!result.contains('<'), "HTML tags must be stripped");
assert!(!result.contains('>'), "HTML tags must be stripped");
assert!(result.contains("422 Error"));
assert!(result.contains("Invalid token"));
}
#[test]
fn test_sanitize_error_body_truncates_large_html_page() {
let html = format!(
"<html><body><p>{}</p></body></html>",
"error detail ".repeat(50)
);
let result = sanitize_error_body(&html);
assert!(result.contains("..."));
assert!(result.contains("bytes total)"));
assert!(!result.contains('<'));
}
#[test]
fn test_sanitize_error_body_passes_short_plain_text() {
assert_eq!(sanitize_error_body("Not Found"), "Not Found");
}
#[test]
fn test_sanitize_error_body_truncates_long_plain_text() {
let long = "x".repeat(300);
let result = sanitize_error_body(&long);
assert!(result.contains("..."));
assert!(result.contains("300 bytes total)"));
}
#[test]
fn test_sanitize_error_body_multibyte_no_panic() {
// 300 CJK characters = 900 bytes; truncation must land on a
// char boundary, not in the middle of a multi-byte sequence.
let cjk = "错误".repeat(150);
let result = sanitize_error_body(&cjk);
assert!(result.contains("..."));
// Must be valid UTF-8 (would have panicked otherwise).
assert!(result.is_char_boundary(result.len()));
}
#[test]
fn test_sanitize_error_body_strips_uppercase_html() {
let html = "<HTML><BODY><H1>500 Internal Server Error</H1></BODY></HTML>";
let result = sanitize_error_body(html);
assert!(
!result.contains('<'),
"uppercase HTML tags must be stripped"
);
assert!(result.contains("500 Internal Server Error"));
}
#[test]
fn test_sanitize_error_body_preserves_angle_brackets_in_non_html() {
// Text with < and > that is NOT an HTML document should be
// left untouched (e.g. log lines, comparison expressions).
let text = "value < 10 and value > 0";
assert_eq!(sanitize_error_body(text), text);
}
} }
+4 -1
View File
@@ -10,6 +10,7 @@ use crate::db::Database;
use crate::extensions::ExtensionManager; use crate::extensions::ExtensionManager;
use crate::llm::{LlmProvider, ToolDefinition}; use crate::llm::{LlmProvider, ToolDefinition};
use crate::orchestrator::job_manager::ContainerJobManager; use crate::orchestrator::job_manager::ContainerJobManager;
use crate::safety::SafetyLayer;
use crate::secrets::SecretsStore; use crate::secrets::SecretsStore;
use crate::skills::catalog::SkillCatalog; use crate::skills::catalog::SkillCatalog;
use crate::skills::registry::SkillRegistry; use crate::skills::registry::SkillRegistry;
@@ -484,15 +485,17 @@ impl ToolRegistry {
pub async fn register_builder_tool( pub async fn register_builder_tool(
self: &Arc<Self>, self: &Arc<Self>,
llm: Arc<dyn LlmProvider>, llm: Arc<dyn LlmProvider>,
safety: Arc<SafetyLayer>,
config: Option<BuilderConfig>, config: Option<BuilderConfig>,
) { ) {
// First register dev tools needed by the builder // First register dev tools needed by the builder
self.register_dev_tools(); self.register_dev_tools();
// Create the builder (arg order: config, llm, tools) // Create the builder (arg order: config, llm, safety, tools)
let builder = Arc::new(LlmSoftwareBuilder::new( let builder = Arc::new(LlmSoftwareBuilder::new(
config.unwrap_or_default(), config.unwrap_or_default(),
llm, llm,
safety,
Arc::clone(self), Arc::clone(self),
)); ));
+1 -1
View File
@@ -133,7 +133,7 @@ impl WorkerRuntime {
.await?; .await?;
// Create reasoning engine // Create reasoning engine
let reasoning = Reasoning::new(self.llm.clone()); let reasoning = Reasoning::new(self.llm.clone(), self.safety.clone());
// Build initial context // Build initial context
let mut reason_ctx = ReasoningContext::new().with_job(&job.description); let mut reason_ctx = ReasoningContext::new().with_job(&job.description);
+116
View File
@@ -113,6 +113,79 @@ pub fn chunk_document(content: &str, config: ChunkConfig) -> Vec<String> {
chunks chunks
} }
/// Split content by paragraphs first, then chunk.
///
/// This is better for preserving semantic boundaries.
#[allow(dead_code)] // Alternative chunking strategy for paragraph-aware indexing
pub fn chunk_by_paragraphs(content: &str, config: ChunkConfig) -> Vec<String> {
if content.is_empty() {
return Vec::new();
}
// Split by double newlines (paragraphs)
let paragraphs: Vec<&str> = content
.split("\n\n")
.map(|p| p.trim())
.filter(|p| !p.is_empty())
.collect();
if paragraphs.is_empty() {
return chunk_document(content, config);
}
let mut chunks = Vec::new();
let mut current_chunk = String::new();
let mut current_word_count = 0;
for paragraph in paragraphs {
let para_words = paragraph.split_whitespace().count();
// If this paragraph alone exceeds chunk size, chunk it separately
if para_words > config.chunk_size {
// Flush current chunk first
if !current_chunk.is_empty() {
chunks.push(current_chunk.trim().to_string());
current_chunk = String::new();
current_word_count = 0;
}
// Chunk the large paragraph
let para_chunks = chunk_document(paragraph, config.clone());
chunks.extend(para_chunks);
continue;
}
// Check if adding this paragraph would exceed chunk size
if current_word_count + para_words > config.chunk_size {
// Flush current chunk
if !current_chunk.is_empty() {
chunks.push(current_chunk.trim().to_string());
}
current_chunk = paragraph.to_string();
current_word_count = para_words;
} else {
// Add paragraph to current chunk
if !current_chunk.is_empty() {
current_chunk.push_str("\n\n");
}
current_chunk.push_str(paragraph);
current_word_count += para_words;
}
}
// Flush remaining content
if !current_chunk.is_empty() {
// If too small, merge with previous chunk if possible
if current_word_count < config.min_chunk_size && !chunks.is_empty() {
let last = chunks.pop().unwrap();
chunks.push(format!("{}\n\n{}", last, current_chunk.trim()));
} else {
chunks.push(current_chunk.trim().to_string());
}
}
chunks
}
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
@@ -180,6 +253,49 @@ mod tests {
assert_eq!(config.step_size(), 85); assert_eq!(config.step_size(), 85);
} }
#[test]
fn test_paragraph_chunking() {
let config = ChunkConfig::default().with_chunk_size(20);
let content = "First paragraph with some words.\n\nSecond paragraph with different content.\n\nThird paragraph here.";
let chunks = chunk_by_paragraphs(content, config);
// Should preserve paragraph boundaries
assert!(!chunks.is_empty());
for chunk in &chunks {
// No chunk should start or end with \n\n
assert!(!chunk.starts_with("\n"));
assert!(!chunk.ends_with("\n"));
}
}
#[test]
fn test_large_paragraph_handling() {
let config = ChunkConfig {
chunk_size: 10,
overlap_percent: 0.15,
min_chunk_size: 3, // Low threshold for test
};
// Create a paragraph with 30 words
let large_para = (1..=30)
.map(|i| format!("word{}", i))
.collect::<Vec<_>>()
.join(" ");
let content = format!("Short intro.\n\n{}\n\nShort outro.", large_para);
let chunks = chunk_by_paragraphs(&content, config);
// Should have multiple chunks due to large paragraph
// 30 words + 2 intro + 2 outro = 34 words, chunk_size=10
// Expect at least 3 chunks
assert!(
chunks.len() >= 3,
"Expected at least 3 chunks for 34 words with chunk_size=10, got {}",
chunks.len()
);
}
#[test] #[test]
fn test_min_chunk_size_merging() { fn test_min_chunk_size_merging() {
let config = ChunkConfig { let config = ChunkConfig {
-644
View File
@@ -1,644 +0,0 @@
//! LanceDB-backed vector store for workspace memory chunks.
//!
//! Provides an alternative to pgvector/libsql for semantic search when the
//! `lancedb` feature is enabled. Documents and metadata stay in the main
//! database; this store holds chunk embeddings for vector similarity search.
//!
//! Configuration:
//! LANCEDB_PATH=~/.ironclaw/lancedb # Default
//! VECTOR_BACKEND=lancedb # Use LanceDB for vector search
/// Default embedding dimension (text-embedding-3-small).
/// Override by passing the actual provider dimension to `LanceDbVectorStore::new()`.
pub const DEFAULT_EMBEDDING_DIM: i32 = 1536;
#[cfg(feature = "lancedb")]
mod impl_lancedb {
use std::sync::Arc;
use arrow_array::types::Float32Type;
use arrow_array::{Array, FixedSizeListArray, RecordBatch, RecordBatchIterator, StringArray};
use arrow_schema::{DataType, Field, Schema};
use async_trait::async_trait;
use futures::StreamExt;
use lancedb::query::{ExecutableQuery, QueryBase};
use uuid::Uuid;
use super::DEFAULT_EMBEDDING_DIM;
use crate::error::WorkspaceError;
use crate::workspace::search::RankedResult;
use crate::workspace::vector_store::VectorStore;
const TABLE_NAME: &str = "memory_chunks";
/// Escapes a string for safe use in LanceDB predicate expressions.
/// Uses SQL-style escaping: single quotes are doubled to prevent injection.
fn escape_predicate_value(s: &str) -> String {
s.replace('\'', "''")
}
/// LanceDB-backed vector store.
///
/// The `update_embedding` method uses delete-then-insert (not atomic).
/// LanceDB does not support transactions, so a crash between the two
/// operations can lose the embedding for that chunk. This is acceptable
/// for personal workspace sizes where data can be reindexed.
pub struct LanceDbVectorStore {
db: Arc<lancedb::Connection>,
table_name: String,
embedding_dim: i32,
schema: Arc<Schema>,
table: tokio::sync::OnceCell<lancedb::Table>,
}
impl LanceDbVectorStore {
/// Create a new LanceDB store at the given path.
///
/// `embedding_dim` should match `EmbeddingProvider::dimension()`.
/// Pass `None` to use the default (1536, text-embedding-3-small).
pub async fn new(
path: impl AsRef<std::path::Path>,
embedding_dim: Option<usize>,
) -> Result<Self, WorkspaceError> {
let path_str = path
.as_ref()
.to_str()
.ok_or_else(|| WorkspaceError::SearchFailed {
reason: "Invalid LanceDB path".to_string(),
})?;
let db = lancedb::connect(path_str).execute().await.map_err(|e| {
WorkspaceError::SearchFailed {
reason: format!("Failed to connect to LanceDB: {}", e),
}
})?;
let dim = embedding_dim.unwrap_or(DEFAULT_EMBEDDING_DIM as usize) as i32;
let schema = Arc::new(Self::build_schema(dim));
let store = Self {
db: Arc::new(db),
table_name: TABLE_NAME.to_string(),
embedding_dim: dim,
schema,
table: tokio::sync::OnceCell::new(),
};
store.ensure_table().await?;
Ok(store)
}
async fn ensure_table(&self) -> Result<(), WorkspaceError> {
let tables = self.db.table_names().execute().await.map_err(|e| {
WorkspaceError::SearchFailed {
reason: format!("Failed to list tables: {}", e),
}
})?;
if tables.iter().any(|t| t == &self.table_name) {
return Ok(());
}
self.db
.create_empty_table(&self.table_name, self.schema.clone())
.execute()
.await
.map_err(|e| WorkspaceError::SearchFailed {
reason: format!("Failed to create table: {}", e),
})?;
// Index creation is deferred — brute-force search via
// bypass_vector_index() works without a pre-built index and is
// sufficient for personal workspace sizes.
Ok(())
}
/// Get or open the cached table handle.
async fn table(&self) -> Result<&lancedb::Table, WorkspaceError> {
self.table
.get_or_try_init(|| async {
self.db
.open_table(&self.table_name)
.execute()
.await
.map_err(|e| WorkspaceError::SearchFailed {
reason: format!("Failed to open table: {}", e),
})
})
.await
}
fn build_schema(embedding_dim: i32) -> Schema {
Schema::new(vec![
Field::new("chunk_id", DataType::Utf8, false),
Field::new("document_id", DataType::Utf8, false),
Field::new("document_path", DataType::Utf8, false),
Field::new("user_id", DataType::Utf8, false),
Field::new("agent_id", DataType::Utf8, true),
Field::new("content", DataType::Utf8, false),
Field::new(
"vector",
DataType::FixedSizeList(
Arc::new(Field::new("item", DataType::Float32, true)),
embedding_dim,
),
false,
),
])
}
}
#[async_trait]
impl VectorStore for LanceDbVectorStore {
async fn store_embedding(
&self,
chunk_id: Uuid,
document_id: Uuid,
document_path: &str,
user_id: &str,
agent_id: Option<Uuid>,
content: &str,
embedding: &[f32],
) -> Result<(), WorkspaceError> {
if embedding.len() != self.embedding_dim as usize {
return Err(WorkspaceError::EmbeddingFailed {
reason: format!(
"Embedding dimension {} does not match expected {}",
embedding.len(),
self.embedding_dim
),
});
}
let table = self.table().await?;
let chunk_ids = StringArray::from(vec![chunk_id.to_string()]);
let document_ids = StringArray::from(vec![document_id.to_string()]);
let document_paths = StringArray::from(vec![document_path]);
let user_ids = StringArray::from(vec![user_id]);
let agent_ids = StringArray::from(vec![agent_id.map(|a| a.to_string())]);
let contents = StringArray::from(vec![content]);
let vec_values: Vec<Option<f32>> = embedding.iter().map(|&x| Some(x)).collect();
let vectors = FixedSizeListArray::from_iter_primitive::<Float32Type, _, _>(
vec![Some(vec_values)],
self.embedding_dim,
);
let batch = RecordBatch::try_new(
self.schema.clone(),
vec![
Arc::new(chunk_ids),
Arc::new(document_ids),
Arc::new(document_paths),
Arc::new(user_ids),
Arc::new(agent_ids),
Arc::new(contents),
Arc::new(vectors),
],
)
.map_err(|e| WorkspaceError::EmbeddingFailed {
reason: format!("Failed to create record batch: {}", e),
})?;
let batches =
RecordBatchIterator::new(vec![Ok(batch)].into_iter(), self.schema.clone());
table
.add(Box::new(batches) as Box<dyn arrow_array::RecordBatchReader + Send>)
.execute()
.await
.map_err(|e| WorkspaceError::EmbeddingFailed {
reason: format!("Failed to store embedding: {}", e),
})?;
Ok(())
}
async fn update_embedding(
&self,
chunk_id: Uuid,
document_id: Uuid,
document_path: &str,
user_id: &str,
agent_id: Option<Uuid>,
content: &str,
embedding: &[f32],
) -> Result<(), WorkspaceError> {
let table = self.table().await?;
table
.delete(&format!(
"chunk_id = '{}'",
escape_predicate_value(&chunk_id.to_string())
))
.await
.map_err(|e| WorkspaceError::EmbeddingFailed {
reason: format!("Failed to delete chunk for update: {}", e),
})?;
self.store_embedding(
chunk_id,
document_id,
document_path,
user_id,
agent_id,
content,
embedding,
)
.await
}
async fn delete_embeddings(&self, document_id: Uuid) -> Result<(), WorkspaceError> {
let table = self.table().await?;
table
.delete(&format!(
"document_id = '{}'",
escape_predicate_value(&document_id.to_string())
))
.await
.map_err(|e| WorkspaceError::EmbeddingFailed {
reason: format!("Failed to delete embeddings: {}", e),
})?;
Ok(())
}
async fn vector_search(
&self,
user_id: &str,
agent_id: Option<Uuid>,
embedding: &[f32],
limit: usize,
) -> Result<Vec<RankedResult>, WorkspaceError> {
let table = self.table().await?;
let filter = if let Some(aid) = agent_id {
format!(
"user_id = '{}' AND agent_id = '{}'",
escape_predicate_value(user_id),
escape_predicate_value(&aid.to_string())
)
} else {
format!(
"user_id = '{}' AND agent_id IS NULL",
escape_predicate_value(user_id)
)
};
let query = table
.query()
.nearest_to(embedding)
.map_err(|e| WorkspaceError::SearchFailed {
reason: format!("Invalid query vector: {}", e),
})?
.only_if(&filter)
.bypass_vector_index()
.limit(limit);
let mut stream = ExecutableQuery::execute(&query).await.map_err(|e| {
WorkspaceError::SearchFailed {
reason: format!("Vector search failed: {}", e),
}
})?;
let mut results = Vec::new();
let mut rank: u32 = 1;
while let Some(batch_result) = stream.next().await {
let batch = batch_result.map_err(|e| WorkspaceError::SearchFailed {
reason: format!("Stream error: {}", e),
})?;
let chunk_id_col = batch.column_by_name("chunk_id").ok_or_else(|| {
WorkspaceError::SearchFailed {
reason: "chunk_id column missing".to_string(),
}
})?;
let document_id_col = batch.column_by_name("document_id").ok_or_else(|| {
WorkspaceError::SearchFailed {
reason: "document_id column missing".to_string(),
}
})?;
let document_path_col = batch.column_by_name("document_path").ok_or_else(|| {
WorkspaceError::SearchFailed {
reason: "document_path column missing".to_string(),
}
})?;
let content_col = batch.column_by_name("content").ok_or_else(|| {
WorkspaceError::SearchFailed {
reason: "content column missing".to_string(),
}
})?;
let chunk_ids = chunk_id_col
.as_any()
.downcast_ref::<StringArray>()
.ok_or_else(|| WorkspaceError::SearchFailed {
reason: "chunk_id wrong type".to_string(),
})?;
let document_ids = document_id_col
.as_any()
.downcast_ref::<StringArray>()
.ok_or_else(|| WorkspaceError::SearchFailed {
reason: "document_id wrong type".to_string(),
})?;
let document_paths = document_path_col
.as_any()
.downcast_ref::<StringArray>()
.ok_or_else(|| WorkspaceError::SearchFailed {
reason: "document_path wrong type".to_string(),
})?;
let contents = content_col
.as_any()
.downcast_ref::<StringArray>()
.ok_or_else(|| WorkspaceError::SearchFailed {
reason: "content wrong type".to_string(),
})?;
for i in 0..batch.num_rows() {
let raw_chunk_id = chunk_ids.value(i);
let chunk_id =
raw_chunk_id
.parse::<Uuid>()
.map_err(|e| WorkspaceError::SearchFailed {
reason: format!("Invalid chunk_id UUID '{}': {}", raw_chunk_id, e),
})?;
let raw_document_id = document_ids.value(i);
let document_id = raw_document_id.parse::<Uuid>().map_err(|e| {
WorkspaceError::SearchFailed {
reason: format!(
"Invalid document_id UUID '{}': {}",
raw_document_id, e
),
}
})?;
let document_path = document_paths.value(i).to_string();
let content = contents.value(i).to_string();
results.push(RankedResult {
chunk_id,
document_id,
document_path,
content,
rank,
});
rank += 1;
}
}
Ok(results)
}
}
}
#[cfg(feature = "lancedb")]
pub use impl_lancedb::LanceDbVectorStore;
#[cfg(all(test, feature = "lancedb"))]
mod tests {
use tempfile::TempDir;
use uuid::Uuid;
use super::{DEFAULT_EMBEDDING_DIM, LanceDbVectorStore};
use crate::workspace::vector_store::VectorStore;
fn make_embedding(seed: f32) -> Vec<f32> {
(0..DEFAULT_EMBEDDING_DIM as usize)
.map(|i| (seed * (i as f32 + 1.0)).sin())
.collect()
}
#[tokio::test]
async fn test_insert_and_vector_search() {
let dir = TempDir::new().unwrap();
let store = LanceDbVectorStore::new(dir.path(), None).await.unwrap();
let chunk_id = Uuid::new_v4();
let document_id = Uuid::new_v4();
let user_id = "user1";
let content = "Rust is a systems programming language";
let embedding = make_embedding(1.0);
store
.store_embedding(
chunk_id,
document_id,
"test.md",
user_id,
None,
content,
&embedding,
)
.await
.unwrap();
let results = store
.vector_search(user_id, None, &embedding, 5)
.await
.unwrap();
assert_eq!(results.len(), 1);
assert_eq!(results[0].chunk_id, chunk_id);
assert_eq!(results[0].document_id, document_id);
assert_eq!(results[0].content, content);
assert_eq!(results[0].rank, 1);
}
#[tokio::test]
async fn test_insert_multiple_and_search_returns_ordered() {
let dir = TempDir::new().unwrap();
let store = LanceDbVectorStore::new(dir.path(), None).await.unwrap();
let doc_id = Uuid::new_v4();
let user_id = "user1";
for (i, seed) in [1.0, 2.0, 3.0].iter().enumerate() {
store
.store_embedding(
Uuid::new_v4(),
doc_id,
"test.md",
user_id,
None,
&format!("content {}", i),
&make_embedding(*seed),
)
.await
.unwrap();
}
let query_emb = make_embedding(2.0);
let results = store
.vector_search(user_id, None, &query_emb, 5)
.await
.unwrap();
assert_eq!(results.len(), 3);
let contents: Vec<_> = results.iter().map(|r| r.content.as_str()).collect();
assert!(contents.contains(&"content 0"));
assert!(contents.contains(&"content 1"));
assert!(contents.contains(&"content 2"));
}
#[tokio::test]
async fn test_delete_chunks() {
let dir = TempDir::new().unwrap();
let store = LanceDbVectorStore::new(dir.path(), None).await.unwrap();
let doc_id = Uuid::new_v4();
let user_id = "user1";
store
.store_embedding(
Uuid::new_v4(),
doc_id,
"test.md",
user_id,
None,
"content",
&make_embedding(1.0),
)
.await
.unwrap();
let results = store
.vector_search(user_id, None, &make_embedding(1.0), 5)
.await
.unwrap();
assert_eq!(results.len(), 1);
store.delete_embeddings(doc_id).await.unwrap();
let results_after = store
.vector_search(user_id, None, &make_embedding(1.0), 5)
.await
.unwrap();
assert!(results_after.is_empty());
}
#[tokio::test]
async fn test_update_chunk_embedding() {
let dir = TempDir::new().unwrap();
let store = LanceDbVectorStore::new(dir.path(), None).await.unwrap();
let chunk_id = Uuid::new_v4();
let doc_id = Uuid::new_v4();
let user_id = "user1";
let content = "original content";
store
.store_embedding(
chunk_id,
doc_id,
"test.md",
user_id,
None,
content,
&make_embedding(1.0),
)
.await
.unwrap();
let new_embedding = make_embedding(5.0);
store
.update_embedding(
chunk_id,
doc_id,
"test.md",
user_id,
None,
content,
&new_embedding,
)
.await
.unwrap();
let results = store
.vector_search(user_id, None, &new_embedding, 5)
.await
.unwrap();
assert_eq!(results.len(), 1);
assert_eq!(results[0].chunk_id, chunk_id);
}
#[tokio::test]
async fn test_vector_search_filters_by_user_and_agent() {
let dir = TempDir::new().unwrap();
let store = LanceDbVectorStore::new(dir.path(), None).await.unwrap();
let doc_id = Uuid::new_v4();
let embedding = make_embedding(1.0);
store
.store_embedding(
Uuid::new_v4(),
doc_id,
"test.md",
"user1",
None,
"user1 content",
&embedding,
)
.await
.unwrap();
store
.store_embedding(
Uuid::new_v4(),
doc_id,
"test.md",
"user2",
None,
"user2 content",
&embedding,
)
.await
.unwrap();
let results_user1 = store
.vector_search("user1", None, &embedding, 5)
.await
.unwrap();
assert_eq!(results_user1.len(), 1);
assert_eq!(results_user1[0].content, "user1 content");
let results_user2 = store
.vector_search("user2", None, &embedding, 5)
.await
.unwrap();
assert_eq!(results_user2.len(), 1);
assert_eq!(results_user2[0].content, "user2 content");
let results_wrong_user = store
.vector_search("user3", None, &embedding, 5)
.await
.unwrap();
assert!(results_wrong_user.is_empty());
}
#[tokio::test]
async fn test_insert_rejects_wrong_embedding_dim() {
let dir = TempDir::new().unwrap();
let store = LanceDbVectorStore::new(dir.path(), None).await.unwrap();
let wrong_dim: Vec<f32> = vec![1.0; 100];
let err = store
.store_embedding(
Uuid::new_v4(),
Uuid::new_v4(),
"test.md",
"user1",
None,
"content",
&wrong_dim,
)
.await
.unwrap_err();
assert!(matches!(
err,
crate::error::WorkspaceError::EmbeddingFailed { .. }
));
}
}
+22 -216
View File
@@ -42,26 +42,20 @@
mod chunker; mod chunker;
mod document; mod document;
pub mod embeddings; mod embeddings;
pub mod hygiene; pub mod hygiene;
#[cfg(feature = "lancedb")]
pub mod lancedb_store;
#[cfg(feature = "postgres")] #[cfg(feature = "postgres")]
mod repository; mod repository;
mod search; mod search;
pub mod vector_store;
pub use chunker::{ChunkConfig, chunk_document}; pub use chunker::{ChunkConfig, chunk_document};
pub use document::{MemoryChunk, MemoryDocument, WorkspaceEntry, paths}; pub use document::{MemoryChunk, MemoryDocument, WorkspaceEntry, paths};
pub use embeddings::{ pub use embeddings::{
EmbeddingProvider, MockEmbeddings, NearAiEmbeddings, OllamaEmbeddings, OpenAiEmbeddings, EmbeddingProvider, MockEmbeddings, NearAiEmbeddings, OllamaEmbeddings, OpenAiEmbeddings,
}; };
#[cfg(feature = "lancedb")]
pub use lancedb_store::{DEFAULT_EMBEDDING_DIM, LanceDbVectorStore};
#[cfg(feature = "postgres")] #[cfg(feature = "postgres")]
pub use repository::Repository; pub use repository::Repository;
pub use search::{RankedResult, SearchConfig, SearchResult, reciprocal_rank_fusion}; pub use search::{RankedResult, SearchConfig, SearchResult, reciprocal_rank_fusion};
pub use vector_store::VectorStore;
use std::sync::Arc; use std::sync::Arc;
@@ -338,12 +332,6 @@ pub struct Workspace {
storage: WorkspaceStorage, storage: WorkspaceStorage,
/// Embedding provider for semantic search. /// Embedding provider for semantic search.
embeddings: Option<Arc<dyn EmbeddingProvider>>, embeddings: Option<Arc<dyn EmbeddingProvider>>,
/// Optional external vector store for semantic search.
///
/// When set, embeddings are stored here instead of (or in addition to)
/// the database's built-in vector support, and hybrid search uses this
/// for the vector component while FTS comes from the database.
vector_store: Option<Arc<dyn VectorStore>>,
} }
impl Workspace { impl Workspace {
@@ -355,7 +343,6 @@ impl Workspace {
agent_id: None, agent_id: None,
storage: WorkspaceStorage::Repo(Repository::new(pool)), storage: WorkspaceStorage::Repo(Repository::new(pool)),
embeddings: None, embeddings: None,
vector_store: None,
} }
} }
@@ -368,7 +355,6 @@ impl Workspace {
agent_id: None, agent_id: None,
storage: WorkspaceStorage::Db(db), storage: WorkspaceStorage::Db(db),
embeddings: None, embeddings: None,
vector_store: None,
} }
} }
@@ -384,17 +370,6 @@ impl Workspace {
self self
} }
/// Set an external vector store for semantic search.
///
/// When set, vector operations (store/search/delete embeddings) use this
/// store instead of the database's built-in vector support. FTS continues
/// to use the database. Hybrid search combines FTS from the database with
/// vector results from this store via RRF.
pub fn with_vector_store(mut self, store: Arc<dyn VectorStore>) -> Self {
self.vector_store = Some(store);
self
}
/// Get the user ID. /// Get the user ID.
pub fn user_id(&self) -> &str { pub fn user_id(&self) -> &str {
&self.user_id &self.user_id
@@ -483,21 +458,9 @@ impl Workspace {
/// Delete a file. /// Delete a file.
/// ///
/// Also deletes associated chunks (from both DB and external vector store). /// Also deletes associated chunks.
pub async fn delete(&self, path: &str) -> Result<(), WorkspaceError> { pub async fn delete(&self, path: &str) -> Result<(), WorkspaceError> {
let path = normalize_path(path); let path = normalize_path(path);
// Clean up external vector store before DB cascade deletes chunks
if let Some(ref vs) = self.vector_store
&& let Ok(doc) = self
.storage
.get_document_by_path(&self.user_id, self.agent_id, &path)
.await
&& let Err(e) = vs.delete_embeddings(doc.id).await
{
tracing::warn!("Failed to delete embeddings from vector store: {}", e);
}
self.storage self.storage
.delete_document_by_path(&self.user_id, self.agent_id, &path) .delete_document_by_path(&self.user_id, self.agent_id, &path)
.await .await
@@ -602,26 +565,11 @@ impl Workspace {
/// ///
/// Daily logs are raw, append-only notes for the current day. /// Daily logs are raw, append-only notes for the current day.
pub async fn append_daily_log(&self, entry: &str) -> Result<(), WorkspaceError> { pub async fn append_daily_log(&self, entry: &str) -> Result<(), WorkspaceError> {
self.append_daily_log_tz(entry, chrono_tz::Tz::UTC) let today = Utc::now().date_naive();
.await
.map(|_| ())
}
/// Append an entry to today's daily log using the given timezone.
///
/// Returns the path that was written to (e.g. `daily/2024-01-15.md`).
pub async fn append_daily_log_tz(
&self,
entry: &str,
tz: chrono_tz::Tz,
) -> Result<String, WorkspaceError> {
let now = crate::timezone::now_in_tz(tz);
let today = now.date_naive();
let path = format!("daily/{}.md", today.format("%Y-%m-%d")); let path = format!("daily/{}.md", today.format("%Y-%m-%d"));
let timestamp = now.format("%H:%M:%S"); let timestamp = Utc::now().format("%H:%M:%S");
let timestamped_entry = format!("[{}] {}", timestamp, entry); let timestamped_entry = format!("[{}] {}", timestamp, entry);
self.append(&path, &timestamped_entry).await?; self.append(&path, &timestamped_entry).await
Ok(path)
} }
// ==================== System Prompt ==================== // ==================== System Prompt ====================
@@ -636,18 +584,6 @@ impl Workspace {
self.system_prompt_for_context(false).await self.system_prompt_for_context(false).await
} }
/// Build the system prompt with timezone-aware daily log dates.
///
/// Uses the given timezone to determine "today" and "yesterday" for daily log injection.
pub async fn system_prompt_for_context_tz(
&self,
is_group_chat: bool,
tz: chrono_tz::Tz,
) -> Result<String, WorkspaceError> {
self.system_prompt_for_context_inner(is_group_chat, Some(tz))
.await
}
/// Build the system prompt, optionally excluding personal memory. /// Build the system prompt, optionally excluding personal memory.
/// ///
/// When `is_group_chat` is true, MEMORY.md is excluded to prevent /// When `is_group_chat` is true, MEMORY.md is excluded to prevent
@@ -655,16 +591,6 @@ impl Workspace {
pub async fn system_prompt_for_context( pub async fn system_prompt_for_context(
&self, &self,
is_group_chat: bool, is_group_chat: bool,
) -> Result<String, WorkspaceError> {
self.system_prompt_for_context_inner(is_group_chat, None)
.await
}
/// Inner implementation for system prompt building.
async fn system_prompt_for_context_inner(
&self,
is_group_chat: bool,
tz: Option<chrono_tz::Tz>,
) -> Result<String, WorkspaceError> { ) -> Result<String, WorkspaceError> {
let mut parts = Vec::new(); let mut parts = Vec::new();
@@ -719,10 +645,7 @@ impl Workspace {
} }
// Add today's memory context (last 2 days of daily logs) // Add today's memory context (last 2 days of daily logs)
let today = match tz { let today = Utc::now().date_naive();
Some(t) => crate::timezone::today_in_tz(t),
None => Utc::now().date_naive(),
};
let yesterday = today.pred_opt().unwrap_or(today); let yesterday = today.pred_opt().unwrap_or(today);
for date in [today, yesterday] { for date in [today, yesterday] {
@@ -762,67 +685,20 @@ impl Workspace {
query: &str, query: &str,
config: SearchConfig, config: SearchConfig,
) -> Result<Vec<SearchResult>, WorkspaceError> { ) -> Result<Vec<SearchResult>, WorkspaceError> {
// Generate embedding for semantic search only when vector search is enabled // Generate embedding for semantic search if provider available
let embedding = let embedding = if let Some(ref provider) = self.embeddings {
if config.use_vector { Some(
if let Some(ref provider) = self.embeddings { provider
Some(provider.embed(query).await.map_err(|e| { .embed(query)
WorkspaceError::EmbeddingFailed { .await
reason: e.to_string(), .map_err(|e| WorkspaceError::EmbeddingFailed {
} reason: e.to_string(),
})?) })?,
} else { )
None } else {
} None
} else { };
None
};
// When an external vector store is configured, do FTS from the
// database and vector search from the store, then fuse with RRF.
if let Some(ref vs) = self.vector_store {
// FTS from database (disable vector to avoid double-searching)
let fts_results = if config.use_fts {
let fts_config = SearchConfig {
use_fts: true,
use_vector: false,
..config.clone()
};
let fts_search = self
.storage
.hybrid_search(&self.user_id, self.agent_id, query, None, &fts_config)
.await?;
fts_search
.into_iter()
.enumerate()
.map(|(i, r)| RankedResult {
chunk_id: r.chunk_id,
document_id: r.document_id,
document_path: r.document_path,
content: r.content,
rank: (i + 1) as u32,
})
.collect()
} else {
Vec::new()
};
// Vector search from external store
let vector_results = if config.use_vector {
if let Some(ref emb) = embedding {
vs.vector_search(&self.user_id, self.agent_id, emb, config.pre_fusion_limit)
.await?
} else {
Vec::new()
}
} else {
Vec::new()
};
return Ok(reciprocal_rank_fusion(fts_results, vector_results, &config));
}
// No external vector store — use database's built-in hybrid search
self.storage self.storage
.hybrid_search( .hybrid_search(
&self.user_id, &self.user_id,
@@ -844,13 +720,7 @@ impl Workspace {
// Chunk the content // Chunk the content
let chunks = chunk_document(&doc.content, ChunkConfig::default()); let chunks = chunk_document(&doc.content, ChunkConfig::default());
// Delete old embeddings from external vector store FIRST — if this fails, // Delete old chunks
// we abort before touching DB chunks, keeping the document consistent.
if let Some(ref vs) = self.vector_store {
vs.delete_embeddings(document_id).await?;
}
// Delete old chunks from database
self.storage.delete_chunks(document_id).await?; self.storage.delete_chunks(document_id).await?;
// Insert new chunks // Insert new chunks
@@ -868,33 +738,9 @@ impl Workspace {
None None
}; };
// When an external vector store is active, skip writing embeddings self.storage
// to the DB (they'd never be queried from there). .insert_chunk(document_id, index as i32, &content, embedding.as_deref())
let db_embedding = if self.vector_store.is_some() {
None
} else {
embedding.as_deref()
};
let chunk_id = self
.storage
.insert_chunk(document_id, index as i32, &content, db_embedding)
.await?; .await?;
// Sync embedding to external vector store (propagate errors to
// avoid leaving a document with deleted-then-missing embeddings).
if let (Some(vs), Some(emb)) = (&self.vector_store, &embedding) {
vs.store_embedding(
chunk_id,
document_id,
&doc.path,
&doc.user_id,
doc.agent_id,
&content,
emb,
)
.await?;
}
} }
Ok(()) Ok(())
@@ -1139,23 +985,6 @@ impl Workspace {
.get_chunks_without_embeddings(&self.user_id, self.agent_id, 100) .get_chunks_without_embeddings(&self.user_id, self.agent_id, 100)
.await?; .await?;
// Prefetch document metadata to avoid N+1 queries when syncing to vector store
let doc_map: std::collections::HashMap<Uuid, crate::workspace::document::MemoryDocument> =
if self.vector_store.is_some() {
let mut map = std::collections::HashMap::new();
for chunk in &chunks {
if !map.contains_key(&chunk.document_id)
&& let Ok(doc) =
self.storage.get_document_by_id(chunk.document_id).await
{
map.insert(doc.id, doc);
}
}
map
} else {
std::collections::HashMap::new()
};
let mut count = 0; let mut count = 0;
for chunk in chunks { for chunk in chunks {
match provider.embed(&chunk.content).await { match provider.embed(&chunk.content).await {
@@ -1163,29 +992,6 @@ impl Workspace {
self.storage self.storage
.update_chunk_embedding(chunk.id, &embedding) .update_chunk_embedding(chunk.id, &embedding)
.await?; .await?;
// Sync to external vector store
if let Some(ref vs) = self.vector_store
&& let Some(doc) = doc_map.get(&chunk.document_id)
&& let Err(e) = vs
.update_embedding(
chunk.id,
chunk.document_id,
&doc.path,
&doc.user_id,
doc.agent_id,
&chunk.content,
&embedding,
)
.await
{
tracing::warn!(
"Failed to sync embedding to vector store for chunk {}: {}",
chunk.id,
e
);
}
count += 1; count += 1;
} }
Err(e) => { Err(e) => {
-71
View File
@@ -1,71 +0,0 @@
//! Vector store abstraction for workspace semantic search.
//!
//! Separates vector search from the main `Database` trait so that
//! third-party vector backends (LanceDB, Qdrant, Pinecone, etc.) can
//! be added by implementing a 4-method trait instead of wrapping the
//! entire ~80-method `Database` trait.
//!
//! When no external vector store is configured, the built-in database
//! vector support (pgvector / libsql_vector_idx) is used via the
//! `Database::hybrid_search` method directly.
use async_trait::async_trait;
use uuid::Uuid;
use crate::error::WorkspaceError;
use crate::workspace::search::RankedResult;
/// External vector store for semantic search.
///
/// Implementations hold chunk embeddings and perform vector similarity
/// queries. Document/chunk metadata and FTS stay in the main database;
/// only embeddings live here.
///
/// # Adding a new backend
///
/// 1. Implement this trait for your backend (4 methods).
/// 2. Feature-gate the module (`#[cfg(feature = "mybackend")]`).
/// 3. Pass `Arc<dyn VectorStore>` to `Workspace::with_vector_store()`.
///
/// That's it — no Database wrapper, no delegation boilerplate.
#[async_trait]
#[allow(clippy::too_many_arguments)]
pub trait VectorStore: Send + Sync {
/// Store an embedding for a chunk.
async fn store_embedding(
&self,
chunk_id: Uuid,
document_id: Uuid,
document_path: &str,
user_id: &str,
agent_id: Option<Uuid>,
content: &str,
embedding: &[f32],
) -> Result<(), WorkspaceError>;
/// Update an existing chunk's embedding (delete + re-insert is fine).
async fn update_embedding(
&self,
chunk_id: Uuid,
document_id: Uuid,
document_path: &str,
user_id: &str,
agent_id: Option<Uuid>,
content: &str,
embedding: &[f32],
) -> Result<(), WorkspaceError>;
/// Delete all embeddings for a document.
async fn delete_embeddings(&self, document_id: Uuid) -> Result<(), WorkspaceError>;
/// Vector similarity search, filtered by user and optional agent.
///
/// Returns results ranked by similarity (rank 1 = most similar).
async fn vector_search(
&self,
user_id: &str,
agent_id: Option<Uuid>,
embedding: &[f32],
limit: usize,
) -> Result<Vec<RankedResult>, WorkspaceError>;
}
-10
View File
@@ -242,16 +242,6 @@ mod tests {
"create_job should return a job_id: {:?}", "create_job should return a job_id: {:?}",
create_result.1 create_result.1
); );
assert!(
create_result.1.contains("in_progress"),
"create_job should dispatch through the scheduler, not stay pending: {:?}",
create_result.1
);
assert!(
!create_result.1.contains("scheduler unavailable"),
"create_job should not fall back to the unscheduled path: {:?}",
create_result.1
);
let status_result = results let status_result = results
.iter() .iter()
.find(|(n, _)| n == "job_status") .find(|(n, _)| n == "job_status")
+15 -8
View File
@@ -20,8 +20,9 @@ mod tests {
use ironclaw::agent::routine_engine::RoutineEngine; use ironclaw::agent::routine_engine::RoutineEngine;
use ironclaw::agent::{HeartbeatConfig, HeartbeatRunner}; use ironclaw::agent::{HeartbeatConfig, HeartbeatRunner};
use ironclaw::channels::IncomingMessage; use ironclaw::channels::IncomingMessage;
use ironclaw::config::RoutineConfig; use ironclaw::config::{RoutineConfig, SafetyConfig};
use ironclaw::db::Database; use ironclaw::db::Database;
use ironclaw::safety::SafetyLayer;
use ironclaw::workspace::Workspace; use ironclaw::workspace::Workspace;
use ironclaw::workspace::hygiene::HygieneConfig; use ironclaw::workspace::hygiene::HygieneConfig;
@@ -117,7 +118,6 @@ mod tests {
"cron-test", "cron-test",
Trigger::Cron { Trigger::Cron {
schedule: "* * * * *".to_string(), schedule: "* * * * *".to_string(),
timezone: None,
}, },
"Check system status.", "Check system status.",
); );
@@ -203,7 +203,6 @@ mod tests {
thread_id: None, thread_id: None,
received_at: Utc::now(), received_at: Utc::now(),
metadata: serde_json::json!({}), metadata: serde_json::json!({}),
timezone: None,
attachments: Vec::new(), attachments: Vec::new(),
}; };
let fired = engine.check_event_triggers(&matching_msg).await; let fired = engine.check_event_triggers(&matching_msg).await;
@@ -225,7 +224,6 @@ mod tests {
thread_id: None, thread_id: None,
received_at: Utc::now(), received_at: Utc::now(),
metadata: serde_json::json!({}), metadata: serde_json::json!({}),
timezone: None,
attachments: Vec::new(), attachments: Vec::new(),
}; };
let fired_neg = engine.check_event_triggers(&non_matching_msg).await; let fired_neg = engine.check_event_triggers(&non_matching_msg).await;
@@ -290,7 +288,6 @@ mod tests {
thread_id: None, thread_id: None,
received_at: Utc::now(), received_at: Utc::now(),
metadata: serde_json::json!({}), metadata: serde_json::json!({}),
timezone: None,
attachments: Vec::new(), attachments: Vec::new(),
}; };
let fired1 = engine.check_event_triggers(&msg).await; let fired1 = engine.check_event_triggers(&msg).await;
@@ -345,6 +342,10 @@ mod tests {
}], }],
); );
let llm = Arc::new(TraceLlm::from_trace(trace)); let llm = Arc::new(TraceLlm::from_trace(trace));
let safety = Arc::new(SafetyLayer::new(&SafetyConfig {
max_output_length: 100_000,
injection_check_enabled: false,
}));
let (tx, mut rx) = tokio::sync::mpsc::channel(16); let (tx, mut rx) = tokio::sync::mpsc::channel(16);
@@ -356,8 +357,9 @@ mod tests {
state_dir: _tmp.path().to_path_buf(), state_dir: _tmp.path().to_path_buf(),
}; };
let runner = HeartbeatRunner::new(HeartbeatConfig::default(), hygiene_config, ws, llm) let runner =
.with_response_channel(tx); HeartbeatRunner::new(HeartbeatConfig::default(), hygiene_config, ws, llm, safety)
.with_response_channel(tx);
let result = runner.check_heartbeat().await; let result = runner.check_heartbeat().await;
match result { match result {
@@ -394,6 +396,10 @@ mod tests {
// LLM should NOT be called, so provide a trace that would panic if called. // LLM should NOT be called, so provide a trace that would panic if called.
let trace = LlmTrace::single_turn("test-heartbeat-skip", "skip", vec![]); let trace = LlmTrace::single_turn("test-heartbeat-skip", "skip", vec![]);
let llm = Arc::new(TraceLlm::from_trace(trace)); let llm = Arc::new(TraceLlm::from_trace(trace));
let safety = Arc::new(SafetyLayer::new(&SafetyConfig {
max_output_length: 100_000,
injection_check_enabled: false,
}));
let hygiene_config = HygieneConfig { let hygiene_config = HygieneConfig {
enabled: false, enabled: false,
@@ -403,7 +409,8 @@ mod tests {
state_dir: _tmp.path().to_path_buf(), state_dir: _tmp.path().to_path_buf(),
}; };
let runner = HeartbeatRunner::new(HeartbeatConfig::default(), hygiene_config, ws, llm); let runner =
HeartbeatRunner::new(HeartbeatConfig::default(), hygiene_config, ws, llm, safety);
let result = runner.check_heartbeat().await; let result = runner.check_heartbeat().await;
assert!( assert!(
+3 -1
View File
@@ -15,6 +15,7 @@ use ironclaw::{
config::Config, config::Config,
history::Store, history::Store,
llm::{create_llm_provider, create_session_manager}, llm::{create_llm_provider, create_session_manager},
safety::SafetyLayer,
workspace::Workspace, workspace::Workspace,
}; };
@@ -92,7 +93,8 @@ async fn test_heartbeat_end_to_end() {
let hb_config = ironclaw::agent::HeartbeatConfig::default(); let hb_config = ironclaw::agent::HeartbeatConfig::default();
let hygiene_config = ironclaw::workspace::hygiene::HygieneConfig::default(); let hygiene_config = ironclaw::workspace::hygiene::HygieneConfig::default();
let runner = HeartbeatRunner::new(hb_config, hygiene_config, workspace, llm); let safety = Arc::new(SafetyLayer::new(&config.safety));
let runner = HeartbeatRunner::new(hb_config, hygiene_config, workspace, llm, safety);
let result = runner.check_heartbeat().await; let result = runner.check_heartbeat().await;
-123
View File
@@ -1,123 +0,0 @@
//! Integration tests for LanceDB vector store with Workspace composition.
//!
//! Requires: cargo test --features "libsql,lancedb"
//!
//! Verifies that Workspace correctly composes FTS from libSQL with vector
//! search from LanceDB via the VectorStore trait.
#![cfg(all(feature = "libsql", feature = "lancedb"))]
use std::sync::Arc;
use ironclaw::db::Database;
use ironclaw::db::libsql::LibSqlBackend;
use ironclaw::workspace::{LanceDbVectorStore, SearchConfig, Workspace};
use tempfile::TempDir;
const EMBEDDING_DIM: usize = 1536;
fn make_embedding(seed: f32) -> Vec<f32> {
(0..EMBEDDING_DIM)
.map(|i| (seed * (i as f32 + 1.0)).sin())
.collect()
}
/// Mock embedding provider that returns deterministic embeddings.
struct FixedEmbeddings {
embedding: Vec<f32>,
}
#[async_trait::async_trait]
impl ironclaw::workspace::EmbeddingProvider for FixedEmbeddings {
fn dimension(&self) -> usize {
EMBEDDING_DIM
}
fn model_name(&self) -> &str {
"fixed-test"
}
fn max_input_length(&self) -> usize {
8192
}
async fn embed(
&self,
_text: &str,
) -> Result<Vec<f32>, ironclaw::workspace::embeddings::EmbeddingError> {
Ok(self.embedding.clone())
}
}
async fn setup_workspace() -> (Workspace, TempDir, TempDir) {
// Use a temp file (not :memory:) because libSQL in-memory DBs are connection-local
let db_dir = TempDir::new().unwrap();
let db_path = db_dir.path().join("test.db");
let libsql = LibSqlBackend::new_local(&db_path).await.unwrap();
libsql.run_migrations().await.unwrap();
let lancedb_dir = TempDir::new().unwrap();
let store = LanceDbVectorStore::new(lancedb_dir.path(), None).await.unwrap();
let embedding = make_embedding(1.0);
let ws = Workspace::new_with_db("test_user", Arc::new(libsql) as Arc<dyn Database>)
.with_vector_store(Arc::new(store))
.with_embeddings(Arc::new(FixedEmbeddings { embedding }));
(ws, lancedb_dir, db_dir)
}
#[tokio::test]
async fn test_workspace_hybrid_search_with_lancedb() {
let (ws, _keep_lance, _keep_db) = setup_workspace().await;
// Write a document — this triggers chunking + embedding + LanceDB sync
ws.write(
"context/rust.md",
"Rust is a systems programming language focused on safety.",
)
.await
.unwrap();
// Hybrid search: FTS for "Rust" + vector from LanceDB
let results = ws.search("Rust", 5).await.unwrap();
assert!(!results.is_empty(), "hybrid search should return results");
assert!(results[0].content.contains("Rust"));
}
#[tokio::test]
async fn test_workspace_delete_removes_from_lancedb() {
let (ws, _keep_lance, _keep_db) = setup_workspace().await;
ws.write("notes/deleted.md", "Content to be deleted.")
.await
.unwrap();
let before = ws.search("deleted", 5).await.unwrap();
assert_eq!(before.len(), 1);
ws.delete("notes/deleted.md").await.unwrap();
let after = ws.search("deleted", 5).await.unwrap();
assert!(after.is_empty());
}
#[tokio::test]
async fn test_workspace_vector_only_search_uses_lancedb() {
let (ws, _keep_lance, _keep_db) = setup_workspace().await;
ws.write("sync/test.md", "Semantic content for vector search")
.await
.unwrap();
// Vector-only search should find via LanceDB even with non-matching FTS query
let config = SearchConfig::default().vector_only().with_limit(5);
let results = ws
.search_with_config("nonexistent_fts_term", config)
.await
.unwrap();
assert_eq!(results.len(), 1);
assert!(results[0].content.contains("Semantic content"));
}
+8 -9
View File
@@ -545,14 +545,16 @@ impl TestRigBuilder {
.await .await
.expect("AppBuilder::build_all() failed in test rig"); .expect("AppBuilder::build_all() failed in test rig");
let scheduler_slot: ironclaw::tools::builtin::SchedulerSlot =
Arc::new(tokio::sync::RwLock::new(None));
// 6. Register job tools, routine tools, and extra tools. // 6. Register job tools, routine tools, and extra tools.
{ {
use ironclaw::context::ContextManager;
let ctx_mgr = Arc::new(ContextManager::new(
components.config.agent.max_parallel_jobs,
));
components.tools.register_job_tools( components.tools.register_job_tools(
Arc::clone(&components.context_manager), ctx_mgr,
Some(scheduler_slot.clone()), None,
None, None,
components.db.clone(), components.db.clone(),
None, None,
@@ -655,13 +657,10 @@ impl TestRigBuilder {
None, // heartbeat_config None, // heartbeat_config
None, // hygiene_config None, // hygiene_config
routine_config, routine_config,
Some(Arc::clone(&components.context_manager)), None, // context_manager
None, // session_manager None, // session_manager
); );
// Match main.rs: fill the scheduler slot once Agent::new has created it.
*scheduler_slot.write().await = Some(agent.scheduler());
// 9. Spawn agent in background task. // 9. Spawn agent in background task.
let agent_handle = tokio::spawn(async move { let agent_handle = tokio::spawn(async move {
if let Err(e) = agent.run().await { if let Err(e) = agent.run().await {
+26 -35
View File
@@ -513,41 +513,32 @@ impl LlmProvider for TraceLlm {
} }
async fn complete(&self, request: CompletionRequest) -> Result<CompletionResponse, LlmError> { async fn complete(&self, request: CompletionRequest) -> Result<CompletionResponse, LlmError> {
// complete() is called when Reasoning has force_text=true (no tools let step = self.next_step(&request.messages)?;
// available). Skip any remaining ToolCalls steps in the trace and match step.response {
// return the next Text step, since in real usage the LLM would TraceResponse::Text {
// produce text when no tools are offered. content,
loop { input_tokens,
let step = self.next_step(&request.messages)?; output_tokens,
match step.response { } => Ok(CompletionResponse {
TraceResponse::Text { content,
content, input_tokens,
input_tokens, output_tokens,
output_tokens, finish_reason: FinishReason::Stop,
} => { cache_read_input_tokens: 0,
return Ok(CompletionResponse { cache_creation_input_tokens: 0,
content, }),
input_tokens, TraceResponse::ToolCalls { .. } => Err(LlmError::RequestFailed {
output_tokens, provider: self.model_name.clone(),
finish_reason: FinishReason::Stop, reason: "TraceLlm::complete() called but current step is a tool_calls response; \
cache_read_input_tokens: 0, use complete_with_tools() instead"
cache_creation_input_tokens: 0, .to_string(),
}); }),
} TraceResponse::UserInput { .. } => Err(LlmError::RequestFailed {
TraceResponse::ToolCalls { .. } => { provider: self.model_name.clone(),
// Skip tool_calls steps — complete() is called in reason: "TraceLlm::complete() encountered a user_input step; \
// force_text mode so the LLM can't use tools anyway. these should have been filtered out during construction"
continue; .to_string(),
} }),
TraceResponse::UserInput { .. } => {
return Err(LlmError::RequestFailed {
provider: self.model_name.clone(),
reason: "TraceLlm::complete() encountered a user_input step; \
these should have been filtered out during construction"
.to_string(),
});
}
}
} }
} }
+9 -16
View File
@@ -571,29 +571,22 @@ mod trace_llm_tests {
} }
#[tokio::test] #[tokio::test]
async fn complete_skips_tool_calls_step() { async fn complete_errors_on_tool_calls_step() {
// complete() is called in force_text mode where tools aren't available.
// When the trace has a ToolCalls step followed by a Text step, complete()
// should skip the ToolCalls and return the Text response.
let trace = LlmTrace::single_turn( let trace = LlmTrace::single_turn(
"test-model", "test-model",
"hi", "hi",
vec![ vec![tool_calls_step(vec![simple_tool_call("echo")], 10, 5)],
tool_calls_step(vec![simple_tool_call("echo")], 10, 5),
text_step("skipped past tools", 20, 8),
],
); );
let llm = TraceLlm::from_trace(trace); let llm = TraceLlm::from_trace(trace);
let resp = llm let result = llm.complete(make_completion_request("hi")).await;
.complete(make_completion_request("hi"))
.await
.expect("complete() should skip ToolCalls and return the Text step");
assert_eq!(resp.content, "skipped past tools"); assert!(result.is_err());
assert_eq!(resp.input_tokens, 20); let err_msg = result.unwrap_err().to_string();
assert_eq!(resp.output_tokens, 8); assert!(
assert_eq!(resp.finish_reason, FinishReason::Stop); err_msg.contains("tool_calls"),
"Expected 'tool_calls' in error: {err_msg}"
);
} }
#[tokio::test] #[tokio::test]