tic: run per-user under a system-agent scheduler, with a run log
Nightly Build / build (push) Successful in 6m58s
Nightly Build / build (push) Successful in 6m58s
Reframe TIC from an ownerless global loop into a per-user system agent.
The events it reads live in each user's own encrypted mcp_events, the
connectors that produced them run in that user's container, and the
notifications go to that user's hub — so the previous design (built
against the ownerless Conversation bundle, writing into system.db and
notifying a hub with no subscribers) was inert by construction.
Core changes
- TicManager owns no timer and no user list. It now exposes
run_for(user_id, pool, sessions, hub): one tick for one user, over
deps unpacked from that user's UserContext. Removed from the
Conversation bundle; Skald::tic_manager() is gone.
- New spawn_system_agents in wiring.rs: one instance-wide loop, spawned
post-construction with a Weak<Skald> (like spawn_user_lifecycle).
Each pass walks the directory and runs TIC for one user at a time —
sequential, because a pass is N container round-trips and N LLM calls
nobody is waiting on. A ConfigKeyUpdated on the interval key cuts the
current wait short; enabled is re-read per pass.
- A user whose database is still locked is skipped (normal, not an
error): the pool is the unlock token, so a user who hasn't logged in
since restart has no readable events and nowhere to record a skip.
- The configured tic.security_group is re-checked per user through
run_context::reconcile_group_for_user — a restricted member never
gets a tool set their role wouldn't grant; unconfigured starts from
role_default_run_context, never None (None = catch-all = wider).
- New system_agent_runs owner table (no user_id column — the file is
the owner): start/finish split so a crash leaves a visible 'running'
row, swept to 'failed' by the next start; safe because the scheduler
is sequential and single-instance. An idle tick writes nothing.
- counting_notify wraps the notify tool so the run log can report
notifications emitted without the tool knowing it's counted.
- The session's event channel is drained by a spawned task instead of
a dropped receiver — the translator awaits its sends and would wedge
at capacity.
EventLog::{Persist,Discard} on McpManager::new
- mcp_events is an owner table and its only reader (TIC) is per-user,
so an event is something that happened to someone. The per-user
runtime gets Persist; the ownerless global runtime gets Discard (its
pool is system.db, rows would be unattributable and unread).
API + UI
- GET /api/system-agents/runs: the caller's own run history, scoped
through require_context with no admin override (same promise as the
rest of the private pool).
- web/components/system-agents.js replaces tic-sessions.js. The old
#tic debug page inferred runs from leftover ephemeral sessions; the
new #system-agents page (sidebar group 'extensions', visible to
everyone — the data is the caller's own) reads the real run log.
- i18n: tic.* keys replaced with system_agents.* in en/it/fr.
Docs
- New docs/system-agents.md (user-facing: what TIC does, why it runs
per person, why a run can be missing). Updated docs/settings.md and
docs/index.md.
- agents/tic/AGENT.md reframed per-user: events are that person's,
memory is user-memory/ (private) — never shared-memory/.
- CLAUDE.md records the system-agents design and the EventLog seam.
This commit is contained in:
@@ -29,6 +29,7 @@ pub mod scheduled_jobs;
|
||||
pub mod scratchpad;
|
||||
pub mod shared_folders;
|
||||
pub mod sources;
|
||||
pub mod system_agent_runs;
|
||||
pub mod tool_permission_groups;
|
||||
pub mod users;
|
||||
|
||||
@@ -895,6 +896,45 @@ pub async fn create_owner_tables(pool: &SqlitePool) -> Result<()> {
|
||||
.execute(pool)
|
||||
.await?;
|
||||
|
||||
// Execution log of the **system agents** — the background agents the instance
|
||||
// runs on a user's behalf without being asked (TIC is the first and, today,
|
||||
// the only one). The sibling of `job_runs`: same shape, but keyed on the agent
|
||||
// instead of a scheduled job, because a system agent has no user-authored row
|
||||
// to point at.
|
||||
//
|
||||
// Owner table, and that is the whole privacy story: a run of TIC summarises
|
||||
// what landed in this user's inbox, so it belongs in *their* encrypted file
|
||||
// and nowhere else. There is deliberately no `user_id` column — the file is
|
||||
// the owner (§5.1). An admin reading `system.db` learns nothing about it.
|
||||
//
|
||||
// A user whose database is still locked is skipped by the scheduler and
|
||||
// produces no row at all: the only file that could hold it is the one we
|
||||
// cannot open. Hence no 'skipped' status — the skip is a log line (§9).
|
||||
sqlx::query(
|
||||
"CREATE TABLE IF NOT EXISTS system_agent_runs (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
agent_id TEXT NOT NULL,
|
||||
session_id INTEGER,
|
||||
started_at TEXT NOT NULL,
|
||||
completed_at TEXT,
|
||||
duration_ms INTEGER,
|
||||
status TEXT NOT NULL
|
||||
CHECK(status IN ('running', 'completed', 'failed', 'cancelled')),
|
||||
stats TEXT,
|
||||
error TEXT,
|
||||
created_at TEXT NOT NULL DEFAULT (datetime('now'))
|
||||
)",
|
||||
)
|
||||
.execute(pool)
|
||||
.await?;
|
||||
|
||||
sqlx::query(
|
||||
"CREATE INDEX IF NOT EXISTS idx_system_agent_runs_agent
|
||||
ON system_agent_runs (agent_id, created_at DESC)",
|
||||
)
|
||||
.execute(pool)
|
||||
.await?;
|
||||
|
||||
sqlx::query(
|
||||
"CREATE TABLE IF NOT EXISTS mcp_events (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
@@ -1098,6 +1138,7 @@ mod tests {
|
||||
one("INSERT INTO scheduled_jobs (id, title, cron, prompt, session_id) VALUES (1, 't', '* * * * *', 'p', 1)")
|
||||
.await.unwrap();
|
||||
one("INSERT INTO job_runs (job_id, started_at, status) VALUES (1, 'now', 'completed')").await.unwrap();
|
||||
one("INSERT INTO system_agent_runs (agent_id, started_at, status) VALUES ('tic', 'now', 'running')").await.unwrap();
|
||||
// Owner table with a BARE `catalog_name` ref — proves it stands alone with
|
||||
// FKs on (an owner→registry FK here would die on this INSERT).
|
||||
one("INSERT INTO mcp_user_servers (name, catalog_name, source) VALUES ('u', 'whatsapp', 'local_script')").await.unwrap();
|
||||
|
||||
@@ -0,0 +1,123 @@
|
||||
//! Execution log of the system agents (blueprint §13).
|
||||
//!
|
||||
//! One row per run, in the **user's own** database — a system agent runs on a
|
||||
//! user's behalf, over their events, so its trace is theirs (see the table
|
||||
//! comment in [`super::create_owner_tables`]). There is no `user_id` column
|
||||
//! because the file is the owner.
|
||||
//!
|
||||
//! The write is split in two, unlike [`super::job_runs`]: [`start`] before the
|
||||
//! agent runs, [`finish`] after. A run that never reaches `finish` — the process
|
||||
//! died mid-turn — stays `running` and is swept to `failed` by the next [`start`]
|
||||
//! for the same agent, which is safe because the scheduler is sequential and
|
||||
//! single-instance: no live run can be in that state when a new one begins.
|
||||
|
||||
use anyhow::Result;
|
||||
use sqlx::SqlitePool;
|
||||
|
||||
/// Terminal statuses. `running` is the transient one written by [`start`].
|
||||
pub const STATUS_COMPLETED: &str = "completed";
|
||||
pub const STATUS_FAILED: &str = "failed";
|
||||
pub const STATUS_CANCELLED: &str = "cancelled";
|
||||
|
||||
#[derive(Debug, Clone, sqlx::FromRow)]
|
||||
pub struct SystemAgentRun {
|
||||
pub id: i64,
|
||||
pub agent_id: String,
|
||||
pub session_id: Option<i64>,
|
||||
pub started_at: String,
|
||||
pub completed_at: Option<String>,
|
||||
pub duration_ms: Option<i64>,
|
||||
pub status: String,
|
||||
/// Free-form JSON with the agent's own counters (TIC: events processed,
|
||||
/// notifications emitted). Never the event contents.
|
||||
pub stats: Option<String>,
|
||||
pub error: Option<String>,
|
||||
pub created_at: String,
|
||||
}
|
||||
|
||||
/// Open a run: sweep any leftover `running` row for this agent, then insert.
|
||||
pub async fn start(pool: &SqlitePool, agent_id: &str) -> Result<i64> {
|
||||
sqlx::query(
|
||||
"UPDATE system_agent_runs
|
||||
SET status = 'failed', error = 'interrupted (server restarted)',
|
||||
completed_at = datetime('now')
|
||||
WHERE agent_id = ? AND status = 'running'",
|
||||
)
|
||||
.bind(agent_id)
|
||||
.execute(pool)
|
||||
.await?;
|
||||
|
||||
let id = sqlx::query_scalar::<_, i64>(
|
||||
"INSERT INTO system_agent_runs (agent_id, started_at, status)
|
||||
VALUES (?, datetime('now'), 'running')
|
||||
RETURNING id",
|
||||
)
|
||||
.bind(agent_id)
|
||||
.fetch_one(pool)
|
||||
.await?;
|
||||
Ok(id)
|
||||
}
|
||||
|
||||
/// Close a run. `duration_ms` is computed by the caller, which holds the
|
||||
/// `Instant` — `datetime('now')` has second resolution and a tick is often faster.
|
||||
pub async fn finish(
|
||||
pool: &SqlitePool,
|
||||
run_id: i64,
|
||||
status: &str,
|
||||
session_id: Option<i64>,
|
||||
duration_ms: i64,
|
||||
stats: Option<&str>,
|
||||
error: Option<&str>,
|
||||
) -> Result<()> {
|
||||
sqlx::query(
|
||||
"UPDATE system_agent_runs
|
||||
SET status = ?, session_id = ?, completed_at = datetime('now'),
|
||||
duration_ms = ?, stats = ?, error = ?
|
||||
WHERE id = ?",
|
||||
)
|
||||
.bind(status)
|
||||
.bind(session_id)
|
||||
.bind(duration_ms)
|
||||
.bind(stats)
|
||||
.bind(error)
|
||||
.bind(run_id)
|
||||
.execute(pool)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Newest-first page of runs, optionally narrowed to one agent.
|
||||
pub async fn list(
|
||||
pool: &SqlitePool,
|
||||
agent_id: Option<&str>,
|
||||
limit: i64,
|
||||
offset: i64,
|
||||
) -> Result<Vec<SystemAgentRun>> {
|
||||
let rows = sqlx::query_as::<_, SystemAgentRun>(
|
||||
"SELECT id, agent_id, session_id, started_at, completed_at, duration_ms,
|
||||
status, stats, error, created_at
|
||||
FROM system_agent_runs
|
||||
WHERE (? IS NULL OR agent_id = ?)
|
||||
ORDER BY id DESC
|
||||
LIMIT ? OFFSET ?",
|
||||
)
|
||||
.bind(agent_id)
|
||||
.bind(agent_id)
|
||||
.bind(limit)
|
||||
.bind(offset)
|
||||
.fetch_all(pool)
|
||||
.await?;
|
||||
Ok(rows)
|
||||
}
|
||||
|
||||
/// Total matching [`list`]'s filter, for pagination.
|
||||
pub async fn count(pool: &SqlitePool, agent_id: Option<&str>) -> Result<i64> {
|
||||
let total = sqlx::query_scalar::<_, i64>(
|
||||
"SELECT COUNT(*) FROM system_agent_runs WHERE (? IS NULL OR agent_id = ?)",
|
||||
)
|
||||
.bind(agent_id)
|
||||
.bind(agent_id)
|
||||
.fetch_one(pool)
|
||||
.await?;
|
||||
Ok(total)
|
||||
}
|
||||
@@ -59,13 +59,50 @@ pub struct McpManager {
|
||||
data_root: PathBuf,
|
||||
}
|
||||
|
||||
/// Whether a runtime's server-pushed notifications are persisted to `mcp_events`.
|
||||
///
|
||||
/// `mcp_events` is an **owner** table and its only consumer is TIC, which is
|
||||
/// per-user: an event is something that happened to *someone*. The global
|
||||
/// runtime has no owner — its pool is `system.db` — so persisting there would
|
||||
/// produce rows nobody can attribute and nobody will ever read.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum EventLog {
|
||||
/// Per-user runtime: notifications land in that user's `mcp_events`.
|
||||
Persist,
|
||||
/// Global runtime: notifications are dropped after the diagnostic log line.
|
||||
Discard,
|
||||
}
|
||||
|
||||
impl McpManager {
|
||||
pub fn new(pool: Arc<SqlitePool>, shutdown: CancellationToken, data_root: impl Into<PathBuf>) -> Self {
|
||||
pub fn new(
|
||||
pool: Arc<SqlitePool>,
|
||||
shutdown: CancellationToken,
|
||||
data_root: impl Into<PathBuf>,
|
||||
event_log: EventLog,
|
||||
) -> Self {
|
||||
let (notification_tx, notification_rx) = mpsc::unbounded_channel::<McpNotification>();
|
||||
let (log_tx, log_rx) = mpsc::unbounded_channel::<McpLogLine>();
|
||||
|
||||
let pool_bg = pool.clone();
|
||||
tokio::spawn(Self::notification_consumer(pool_bg, notification_rx, shutdown.clone()));
|
||||
match event_log {
|
||||
EventLog::Persist => {
|
||||
tokio::spawn(Self::notification_consumer(pool_bg, notification_rx, shutdown.clone()));
|
||||
}
|
||||
// Still drain the channel: the senders are unbounded, but a receiver
|
||||
// dropped here would make every `send` fail and log noise per event.
|
||||
EventLog::Discard => {
|
||||
let sd = shutdown.clone();
|
||||
tokio::spawn(async move {
|
||||
let mut rx = notification_rx;
|
||||
loop {
|
||||
tokio::select! {
|
||||
_ = sd.cancelled() => break,
|
||||
msg = rx.recv() => if msg.is_none() { break },
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
tokio::spawn(logs::log_consumer(log_rx, shutdown));
|
||||
|
||||
Self {
|
||||
|
||||
@@ -38,7 +38,6 @@ use crate::provider::ProviderRegistry;
|
||||
use crate::run_context::RunContextManager;
|
||||
use crate::secrets::SecretsStore;
|
||||
use crate::session::manager::ChatSessionManager;
|
||||
use crate::tic::TicManager;
|
||||
use crate::tool_catalog::ToolCatalog;
|
||||
use crate::tools::ToolRegistry;
|
||||
use crate::transcribe::TranscribeManager;
|
||||
@@ -293,7 +292,6 @@ impl Skald {
|
||||
pub fn manager(&self) -> &Arc<ChatSessionManager> { &self.conversation.manager }
|
||||
pub fn chat_hub(&self) -> &Arc<ChatHub> { &self.conversation.chat_hub }
|
||||
pub fn run_context_manager(&self) -> &Arc<RunContextManager> { &self.conversation.run_context_manager }
|
||||
pub fn tic_manager(&self) -> &Arc<TicManager> { &self.conversation.tic_manager }
|
||||
|
||||
// Interaction
|
||||
pub fn approval(&self) -> &Arc<ApprovalManager> { &self.interaction.approval }
|
||||
|
||||
@@ -35,7 +35,6 @@ use crate::run_context::RunContextManager;
|
||||
use crate::secrets::SecretsStore;
|
||||
use crate::session::handler::{DEFAULT_MAX_PARALLEL_SUBAGENTS, DEFAULT_MAX_TOOL_ROUNDS};
|
||||
use crate::session::manager::ChatSessionManager;
|
||||
use crate::tic::TicManager;
|
||||
use crate::tool_catalog::ToolCatalog;
|
||||
use crate::tool_discovery::ToolDiscovery;
|
||||
use crate::tools::ToolRegistry;
|
||||
@@ -160,7 +159,14 @@ impl Integrations {
|
||||
/// after the elicitation handler is wired) and the plugin manager (plugins are
|
||||
/// injected by `main.rs`; `start_enabled()` runs later, from `WebFrontend`).
|
||||
pub(super) fn build(rt: &Runtime, plugins: Vec<Arc<dyn Plugin>>) -> Self {
|
||||
let mcp = Arc::new(McpManager::new(Arc::clone(&rt.db), rt.shutdown_token.clone(), "data"));
|
||||
// The global runtime has no owner, so its notifications are not persisted:
|
||||
// `mcp_events` is per-user and its only reader (TIC) runs per-user.
|
||||
let mcp = Arc::new(McpManager::new(
|
||||
Arc::clone(&rt.db),
|
||||
rt.shutdown_token.clone(),
|
||||
"data",
|
||||
crate::mcp::EventLog::Discard,
|
||||
));
|
||||
|
||||
let mut plugin_manager = PluginManager::new(Arc::clone(&rt.db));
|
||||
for plugin in plugins {
|
||||
@@ -297,16 +303,12 @@ impl Interaction {
|
||||
}
|
||||
}
|
||||
|
||||
// ── Conversation: session manager + chat hub + run context + TIC ────────────
|
||||
// ── Conversation: session manager + chat hub + run context ──────────────────
|
||||
|
||||
pub(super) struct Conversation {
|
||||
pub(super) manager: Arc<ChatSessionManager>,
|
||||
pub(super) chat_hub: Arc<ChatHub>,
|
||||
pub(super) run_context_manager: Arc<RunContextManager>,
|
||||
/// TIC lives here (rather than in `Tasks`) because it is constructed from and
|
||||
/// drives the conversation stack (session manager + chat hub + run context);
|
||||
/// this keeps every bundle a single-shot `build()` with no two-phase init.
|
||||
pub(super) tic_manager: Arc<TicManager>,
|
||||
}
|
||||
|
||||
impl Conversation {
|
||||
@@ -395,17 +397,12 @@ impl Conversation {
|
||||
chat_hub.register("web").await;
|
||||
chat_hub.register("talk").await;
|
||||
|
||||
let tic_manager = TicManager::new(
|
||||
Arc::clone(&rt.db),
|
||||
Arc::clone(&manager),
|
||||
Arc::clone(&chat_hub),
|
||||
config.tic.clone(),
|
||||
Arc::clone(&rt.config),
|
||||
Arc::clone(&run_context_manager),
|
||||
Arc::clone(&rt.system_bus),
|
||||
);
|
||||
// TIC is deliberately absent: it is a system agent that runs *per user*,
|
||||
// over that user's own events, sessions and hub. Building it here would
|
||||
// bind it to the ownerless stack above (§19) — which is precisely the bug
|
||||
// that made it inert. It is constructed by `wiring::spawn_system_agents`.
|
||||
|
||||
Ok(Conversation { manager, chat_hub, run_context_manager, tic_manager })
|
||||
Ok(Conversation { manager, chat_hub, run_context_manager })
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -31,7 +31,7 @@ use bundles::{Conversation, Infra, Integrations, Interaction, Media, Models, Tas
|
||||
use runtime::Runtime;
|
||||
use user_context::{UserContextFactory, UserContextRegistry};
|
||||
pub use user_context::UserContext;
|
||||
use wiring::{spawn_background, spawn_user_lifecycle, wire};
|
||||
use wiring::{spawn_background, spawn_system_agents, spawn_user_lifecycle, wire};
|
||||
|
||||
pub struct Skald {
|
||||
rt: Runtime,
|
||||
@@ -111,6 +111,10 @@ impl Skald {
|
||||
// can only be spawned once the instance exists (blueprint §6).
|
||||
spawn_user_lifecycle(&skald);
|
||||
|
||||
// Likewise the system-agent scheduler: it resolves a per-user runtime for
|
||||
// each user it runs an agent for (blueprint §13).
|
||||
spawn_system_agents(&skald, config.tic.clone());
|
||||
|
||||
Ok(skald)
|
||||
}
|
||||
|
||||
|
||||
@@ -241,6 +241,9 @@ impl UserContextFactory {
|
||||
Arc::clone(&pool),
|
||||
user_shutdown.clone(),
|
||||
"data",
|
||||
// This user's connectors push into this user's `mcp_events`, which is
|
||||
// what TIC reads on their behalf.
|
||||
crate::mcp::EventLog::Persist,
|
||||
));
|
||||
// NOTE: per-user MCP elicitation (interactive connector login, §15) is
|
||||
// deferred — api-key connectors don't need it. Wire the user's
|
||||
|
||||
@@ -2,19 +2,22 @@
|
||||
//! spawns, each concentrated in one readable place instead of being scattered
|
||||
//! through the constructor.
|
||||
//!
|
||||
//! Owner-bound background loops (cron, session-cancel, ticket-listener, tic) have
|
||||
//! Owner-bound background loops (cron, session-cancel, ticket-listener) have
|
||||
//! moved per-user into `UserContextFactory::build`. What remains here are the
|
||||
//! instance-wide tasks: LLM-log cleanup on the registry pool, MCP server
|
||||
//! initialization, and the user-lifecycle reconciler (which needs the finished
|
||||
//! `Arc<Skald>` and is therefore spawned separately, after construction).
|
||||
//! initialization, and the two that need the finished `Arc<Skald>` and are
|
||||
//! therefore spawned separately, after construction — the user-lifecycle
|
||||
//! reconciler and the system-agent scheduler.
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use core_api::system_bus::{RecvError, SystemEvent};
|
||||
use tracing::{info, warn};
|
||||
|
||||
use crate::config::CoreConfig;
|
||||
use crate::config::{CoreConfig, TicConfig};
|
||||
use crate::elicitation::ElicitationBridge;
|
||||
use crate::tic::{TicManager, TIC_INTERVAL_MINUTES_KEY};
|
||||
|
||||
use super::bundles::{Conversation, Integrations, Interaction, Tasks};
|
||||
use super::runtime::Runtime;
|
||||
@@ -41,10 +44,11 @@ pub(super) fn wire(
|
||||
|
||||
/// Spawns the instance-wide background tasks.
|
||||
///
|
||||
/// Owner-bound loops (cron, session-cancel, ticket-listener, tic) are **not**
|
||||
/// spawned here — they run per-user inside `UserContext`. Session cancellation is
|
||||
/// handled directly by the API handlers (which have `AuthUser` and resolve the
|
||||
/// per-user context). TIC is deferred until connectors return (§13).
|
||||
/// Owner-bound loops (cron, session-cancel, ticket-listener) are **not** spawned
|
||||
/// here — they run per-user inside `UserContext`. Session cancellation is handled
|
||||
/// directly by the API handlers (which have `AuthUser` and resolve the per-user
|
||||
/// context). The system-agent scheduler needs the finished instance and lives in
|
||||
/// [`spawn_system_agents`].
|
||||
pub(super) fn spawn_background(
|
||||
rt: &Runtime,
|
||||
_tasks: &Tasks,
|
||||
@@ -169,3 +173,107 @@ pub(super) fn spawn_user_lifecycle(skald: &Arc<super::Skald>) {
|
||||
info!("user-lifecycle: reconciler stopped");
|
||||
});
|
||||
}
|
||||
|
||||
/// Spawns the **system-agent scheduler** — the instance-wide timer that runs the
|
||||
/// background agents nobody asked for (today: TIC).
|
||||
///
|
||||
/// One loop, not one per user. Every pass walks the user directory and runs the
|
||||
/// agent for each user **sequentially**: a pass means N container round-trips and
|
||||
/// N LLM calls, and doing them concurrently would spike the box every interval
|
||||
/// for no gain — nobody is waiting on a background tick.
|
||||
///
|
||||
/// A user whose database is still locked is **skipped**, and that is the normal
|
||||
/// case rather than an error: the pool is the unlock token (§9), so a user who
|
||||
/// has not logged in since the last restart has no readable events, no session
|
||||
/// store, and no place to record the skip. It is logged at INFO and the pass
|
||||
/// moves on; their events keep accumulating and are picked up by the first pass
|
||||
/// after they log in.
|
||||
///
|
||||
/// Spawned after `Skald` is fully built, like [`spawn_user_lifecycle`] and for
|
||||
/// the same reason: it resolves each user's runtime through `Skald::user_context`.
|
||||
/// The back-reference is [`std::sync::Weak`].
|
||||
pub(super) fn spawn_system_agents(skald: &Arc<super::Skald>, tic_config: TicConfig) {
|
||||
let weak = Arc::downgrade(skald);
|
||||
let shutdown = skald.rt.shutdown_token.clone();
|
||||
let mut sys_rx = skald.rt.system_bus.subscribe();
|
||||
|
||||
let tic = TicManager::new(
|
||||
tic_config,
|
||||
Arc::clone(&skald.rt.config),
|
||||
Arc::clone(&skald.rt.db),
|
||||
);
|
||||
|
||||
skald.rt.supervisor.spawn("system-agents", async move {
|
||||
info!("system-agents: scheduler started");
|
||||
|
||||
'outer: loop {
|
||||
// Re-read the interval each pass so a Settings change lands without a
|
||||
// restart; a live change also cuts the current wait short.
|
||||
let wait = Duration::from_secs(tic.interval_secs().await);
|
||||
let deadline = tokio::time::sleep(wait);
|
||||
tokio::pin!(deadline);
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
_ = shutdown.cancelled() => break 'outer,
|
||||
_ = &mut deadline => break,
|
||||
ev = sys_rx.recv() => match ev {
|
||||
Ok(SystemEvent::ConfigKeyUpdated { key, .. })
|
||||
if key == TIC_INTERVAL_MINUTES_KEY =>
|
||||
{
|
||||
info!("system-agents: interval changed, rescheduling");
|
||||
continue 'outer;
|
||||
}
|
||||
Err(RecvError::Closed) => break 'outer,
|
||||
_ => {}
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
let Some(skald) = weak.upgrade() else { break };
|
||||
tic_pass(&skald, &tic).await;
|
||||
}
|
||||
|
||||
info!("system-agents: scheduler stopped");
|
||||
});
|
||||
}
|
||||
|
||||
/// One TIC pass over the whole directory, one user at a time.
|
||||
async fn tic_pass(skald: &Arc<super::Skald>, tic: &Arc<TicManager>) {
|
||||
if !tic.is_enabled().await {
|
||||
return;
|
||||
}
|
||||
|
||||
let users = match skald.users().list().await {
|
||||
Ok(u) => u,
|
||||
Err(e) => {
|
||||
warn!(error = %e, "system-agents: cannot list users, skipping this pass");
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
for user in users.into_iter().filter(|u| u.active) {
|
||||
if skald.rt.shutdown_token.is_cancelled() {
|
||||
break;
|
||||
}
|
||||
|
||||
if !skald.users().is_unlocked(&user.id) {
|
||||
info!(
|
||||
user = %user.id, username = %user.username,
|
||||
"TIC: skipped — the user's database is still encrypted (not logged in since the last restart)",
|
||||
);
|
||||
continue;
|
||||
}
|
||||
|
||||
// Unlocked, so this resolves (and is normally already live from their login).
|
||||
let Some(ctx) = skald.user_context(&user.id).await else {
|
||||
warn!(user = %user.id, "TIC: skipped — could not resolve the user's runtime");
|
||||
continue;
|
||||
};
|
||||
|
||||
if let Err(e) = tic.run_for(&user.id, &ctx.pool, &ctx.sessions, &ctx.chat_hub).await {
|
||||
// One user's failure must not end the pass for everyone after them.
|
||||
warn!(user = %user.id, error = %e, "TIC: tick failed");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+241
-147
@@ -1,51 +1,81 @@
|
||||
//! TIC — the background event processor, and the first of the **system agents**.
|
||||
//!
|
||||
//! A system agent runs on a user's behalf without being asked. TIC's job is to
|
||||
//! look at the events the user's connectors pushed since the last tick (new
|
||||
//! mail, a calendar change, a WhatsApp message), decide which of them are worth
|
||||
//! interrupting the user for, and `notify()` those.
|
||||
//!
|
||||
//! **It is per-user, and that is not an implementation detail.** The events it
|
||||
//! reads live in `mcp_events` inside the caller's own encrypted database, the
|
||||
//! connectors that produced them run inside the caller's container, and the
|
||||
//! notification it emits goes to the caller's own hub. This manager therefore
|
||||
//! owns no timer and no user list: it exposes [`TicManager::run_for`], one tick
|
||||
//! for one user, and the instance-wide scheduler
|
||||
//! (`skald::wiring::spawn_system_agents`) decides who to run it for and when —
|
||||
//! sequentially, skipping anyone whose database is still locked.
|
||||
//!
|
||||
//! The run is recorded in `system_agent_runs` in that same user's database, so
|
||||
//! the trace of what TIC did for someone is readable by them and by nobody else.
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::time::Duration;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::time::Instant;
|
||||
|
||||
use sqlx::SqlitePool;
|
||||
use tokio::sync::mpsc;
|
||||
use tracing::{info, warn};
|
||||
|
||||
use core_api::interface_tool::{InterfaceTool, ToolFuture};
|
||||
use core_api::{ConfigProperty, ConfigSet, PropertyType};
|
||||
use core_api::system_bus::{SystemEvent, SystemEventBus};
|
||||
|
||||
use crate::chat_hub::ChatHub;
|
||||
use crate::config::TicConfig;
|
||||
use crate::config_store::GlobalConfigManager;
|
||||
use crate::db::mcp_events;
|
||||
use crate::run_context::{RunContext, RunContextManager};
|
||||
use crate::db::{mcp_events, system_agent_runs};
|
||||
use crate::run_context::{self, RunContext};
|
||||
use crate::session::manager::ChatSessionManager;
|
||||
|
||||
/// The chat `source` TIC's ephemeral sessions carry. Kept distinct from the
|
||||
/// user-facing sources (`web`, `talk`, `telegram`) so a tick never lands in a
|
||||
/// conversation someone is reading.
|
||||
const TIC_SOURCE: &str = "tic";
|
||||
const TIC_AGENT: &str = "tic";
|
||||
/// The agent id, in `agents/tic/`, and the `agent_id` of its `system_agent_runs` rows.
|
||||
pub const TIC_AGENT: &str = "tic";
|
||||
|
||||
pub const TIC_ENABLED_KEY: &str = "tic.enabled";
|
||||
pub const TIC_SECURITY_GROUP_KEY: &str = "tic.security_group";
|
||||
pub const TIC_INTERVAL_MINUTES_KEY: &str = "tic.interval_minutes";
|
||||
pub const TIC_ENABLED_KEY: &str = "tic.enabled";
|
||||
pub const TIC_SECURITY_GROUP_KEY: &str = "tic.security_group";
|
||||
pub const TIC_INTERVAL_MINUTES_KEY: &str = "tic.interval_minutes";
|
||||
|
||||
pub fn config_set() -> ConfigSet {
|
||||
ConfigSet {
|
||||
name: "TIC Agent".into(),
|
||||
description: "TIC is a background agent that monitors all async events generated by connected MCP servers (new emails, calendar updates, WhatsApp messages, etc.). It reads your notification rules from data/notifications.md and your memory to decide — via an LLM call — which events are worth surfacing. Relevant notifications are forwarded to the home agent set via /sethome.".into(),
|
||||
description: "TIC is a background agent that runs for every user, one at a time. For each \
|
||||
user it reads the events their own connectors have pushed since the last run \
|
||||
(new mail, calendar changes, incoming messages), decides — via an LLM call — \
|
||||
which of them are worth surfacing, and sends those to that user as \
|
||||
notifications. It reads only that user's events and writes only to their own \
|
||||
conversation; a user who has not logged in since the last restart is skipped, \
|
||||
because their database is still encrypted. Each run is recorded on the System \
|
||||
agents page, visible to the user it ran for.".into(),
|
||||
properties: vec![
|
||||
ConfigProperty {
|
||||
key: TIC_ENABLED_KEY.into(),
|
||||
name: "Enabled".into(),
|
||||
description: "Enable or disable the TIC agent. When disabled, no MCP events are processed.".into(),
|
||||
description: "Enable or disable the TIC agent for the whole instance. When disabled, no events are processed for anyone.".into(),
|
||||
property_type: PropertyType::Bool,
|
||||
default_value: Some("true".into()),
|
||||
},
|
||||
ConfigProperty {
|
||||
key: TIC_SECURITY_GROUP_KEY.into(),
|
||||
name: "Security Group".into(),
|
||||
description: "Tool permission group applied to each TIC agent session. Leave empty to use the default group.".into(),
|
||||
description: "Tool permission group applied to each TIC run. It is re-checked against each user's own role: a user whose role does not allow this group runs under their role's default group instead. Leave empty to always use the role default.".into(),
|
||||
property_type: PropertyType::SecurityGroup,
|
||||
default_value: None,
|
||||
},
|
||||
ConfigProperty {
|
||||
key: TIC_INTERVAL_MINUTES_KEY.into(),
|
||||
name: "Check Interval (minutes)".into(),
|
||||
description: "How often TIC runs, in minutes. Leave empty to use the value from config.yml (tic.interval_secs).".into(),
|
||||
description: "How often TIC starts a pass over all users, in minutes. Leave empty to use the value from config.yml (tic.interval_secs).".into(),
|
||||
property_type: PropertyType::Int,
|
||||
default_value: Some("15".into()),
|
||||
},
|
||||
@@ -53,90 +83,51 @@ pub fn config_set() -> ConfigSet {
|
||||
}
|
||||
}
|
||||
|
||||
/// What one tick did, for the run log. Counters only — never event contents.
|
||||
pub struct TicRun {
|
||||
pub session_id: i64,
|
||||
pub events_processed: usize,
|
||||
pub notifications_emitted: usize,
|
||||
}
|
||||
|
||||
impl TicRun {
|
||||
fn stats_json(&self) -> String {
|
||||
serde_json::json!({
|
||||
"events_processed": self.events_processed,
|
||||
"notifications_emitted": self.notifications_emitted,
|
||||
})
|
||||
.to_string()
|
||||
}
|
||||
}
|
||||
|
||||
pub struct TicManager {
|
||||
db: Arc<SqlitePool>,
|
||||
session_mgr: Arc<ChatSessionManager>,
|
||||
hub: Arc<ChatHub>,
|
||||
config: TicConfig,
|
||||
config_store: Arc<GlobalConfigManager>,
|
||||
run_context_manager: Arc<RunContextManager>,
|
||||
system_bus: Arc<SystemEventBus>,
|
||||
/// Guards against concurrent ticks (e.g. if a tick takes longer than the interval).
|
||||
running: AtomicBool,
|
||||
config: TicConfig,
|
||||
config_store: Arc<GlobalConfigManager>,
|
||||
/// `system.db` — read to resolve each user's role when validating the
|
||||
/// configured security group. Never written.
|
||||
registry_pool: Arc<SqlitePool>,
|
||||
}
|
||||
|
||||
impl TicManager {
|
||||
pub fn new(
|
||||
db: Arc<SqlitePool>,
|
||||
session_mgr: Arc<ChatSessionManager>,
|
||||
hub: Arc<ChatHub>,
|
||||
config: TicConfig,
|
||||
config_store: Arc<GlobalConfigManager>,
|
||||
run_context_manager: Arc<RunContextManager>,
|
||||
system_bus: Arc<SystemEventBus>,
|
||||
config: TicConfig,
|
||||
config_store: Arc<GlobalConfigManager>,
|
||||
registry_pool: Arc<SqlitePool>,
|
||||
) -> Arc<Self> {
|
||||
Arc::new(Self {
|
||||
db,
|
||||
session_mgr,
|
||||
hub,
|
||||
config,
|
||||
config_store,
|
||||
run_context_manager,
|
||||
system_bus,
|
||||
running: AtomicBool::new(false),
|
||||
})
|
||||
Arc::new(Self { config, config_store, registry_pool })
|
||||
}
|
||||
|
||||
/// Force a tick immediately, ignoring the running guard.
|
||||
/// Intended for manual triggering (e.g. via the `/api/tic/trigger` endpoint).
|
||||
pub async fn tick_now(self: Arc<Self>) {
|
||||
if let Err(e) = self.run_tick().await {
|
||||
warn!(error = %e, "TicManager: forced tick failed");
|
||||
/// Instance-wide on/off switch. Read fresh each pass, so toggling it in
|
||||
/// Settings takes effect at the next pass with no restart.
|
||||
pub async fn is_enabled(&self) -> bool {
|
||||
match self.config_store.get(TIC_ENABLED_KEY).await {
|
||||
Ok(Some(v)) => v != "false",
|
||||
_ => true,
|
||||
}
|
||||
}
|
||||
|
||||
/// Spawn the background timer.
|
||||
/// Subscribes to ConfigKeyUpdated so the interval can be changed at runtime.
|
||||
pub fn start(self: Arc<Self>, shutdown: tokio_util::sync::CancellationToken) -> tokio::task::JoinHandle<()> {
|
||||
tokio::spawn(async move {
|
||||
let mut interval_secs = self.effective_interval_secs().await;
|
||||
info!("TicManager started (interval={}s, batch={})", interval_secs, self.config.batch_size);
|
||||
|
||||
let mut timer = tokio::time::interval(Duration::from_secs(interval_secs));
|
||||
timer.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
|
||||
|
||||
let mut sys_rx = self.system_bus.subscribe();
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
_ = shutdown.cancelled() => {
|
||||
info!("TicManager: stopping");
|
||||
break;
|
||||
}
|
||||
res = sys_rx.recv() => {
|
||||
if let Ok(SystemEvent::ConfigKeyUpdated { key, new_value, .. }) = res {
|
||||
if key == TIC_INTERVAL_MINUTES_KEY {
|
||||
if let Ok(mins) = new_value.parse::<u64>() {
|
||||
let new_secs = mins.max(1) * 60;
|
||||
if new_secs != interval_secs {
|
||||
interval_secs = new_secs;
|
||||
timer = tokio::time::interval(Duration::from_secs(interval_secs));
|
||||
timer.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
|
||||
info!(secs = interval_secs, "TicManager: interval updated");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
_ = timer.tick() => {
|
||||
self.tick().await;
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
async fn effective_interval_secs(&self) -> u64 {
|
||||
/// Seconds between passes: the Settings value wins, else `config.yml`.
|
||||
pub async fn interval_secs(&self) -> u64 {
|
||||
if let Ok(Some(val)) = self.config_store.get(TIC_INTERVAL_MINUTES_KEY).await {
|
||||
if let Ok(mins) = val.parse::<u64>() {
|
||||
if mins > 0 {
|
||||
@@ -147,75 +138,178 @@ impl TicManager {
|
||||
self.config.interval_secs
|
||||
}
|
||||
|
||||
async fn is_enabled(&self) -> bool {
|
||||
match self.config_store.get(TIC_ENABLED_KEY).await {
|
||||
Ok(Some(v)) => v != "false",
|
||||
_ => true,
|
||||
}
|
||||
}
|
||||
|
||||
async fn tick(&self) {
|
||||
if !self.is_enabled().await {
|
||||
return;
|
||||
}
|
||||
// Prevent concurrent ticks.
|
||||
if self.running.compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst).is_err() {
|
||||
warn!("TicManager: previous tick still running, skipping");
|
||||
return;
|
||||
}
|
||||
|
||||
let result = self.run_tick().await;
|
||||
self.running.store(false, Ordering::SeqCst);
|
||||
|
||||
if let Err(e) = result {
|
||||
warn!(error = %e, "TicManager: tick failed");
|
||||
}
|
||||
}
|
||||
|
||||
async fn run_tick(&self) -> anyhow::Result<()> {
|
||||
// 1. Fetch the oldest N unprocessed events.
|
||||
let events = mcp_events::pending_limited(&self.db, self.config.batch_size).await?;
|
||||
/// One tick for one user, over that user's own runtime.
|
||||
///
|
||||
/// `Ok(None)` means there was nothing to do — no pending events — and
|
||||
/// **nothing is written**: an idle tick must not leave a row behind, or the
|
||||
/// run log becomes a heartbeat instead of a history. Any other outcome opens
|
||||
/// a `system_agent_runs` row and closes it, failure included.
|
||||
pub async fn run_for(
|
||||
&self,
|
||||
user_id: &str,
|
||||
pool: &SqlitePool,
|
||||
sessions: &Arc<ChatSessionManager>,
|
||||
hub: &Arc<ChatHub>,
|
||||
) -> anyhow::Result<Option<TicRun>> {
|
||||
let events = mcp_events::pending_limited(pool, self.config.batch_size).await?;
|
||||
if events.is_empty() {
|
||||
return Ok(());
|
||||
return Ok(None);
|
||||
}
|
||||
|
||||
info!(count = events.len(), "TicManager: processing event batch");
|
||||
let run_id = system_agent_runs::start(pool, TIC_AGENT).await?;
|
||||
let started = Instant::now();
|
||||
|
||||
// 2. Mark as processed BEFORE running the agent — avoids double-processing
|
||||
// if the process crashes mid-turn.
|
||||
let ids: Vec<i64> = events.iter().map(|e| e.id).collect();
|
||||
mcp_events::mark_processed(&self.db, &ids).await?;
|
||||
|
||||
// 3. Serialize events into the agent prompt.
|
||||
let prompt = build_prompt(&events);
|
||||
|
||||
// 4. Create a fresh ephemeral session (agent_id = "tic", source = "tic").
|
||||
// We bypass ChatHub entirely — TIC is not a user-facing source and should
|
||||
// not appear in the sources table or consume a broadcast channel.
|
||||
let (session_id, _) = self.session_mgr.create_session(TIC_AGENT, TIC_SOURCE, false, true, None).await?;
|
||||
let handler = self.session_mgr.get_or_create_handler(session_id).await?;
|
||||
handler.set_auto_deny_approvals();
|
||||
|
||||
// 5. Apply run context if configured in DB.
|
||||
if let Ok(Some(rc_id)) = self.config_store.get(TIC_SECURITY_GROUP_KEY).await {
|
||||
if !rc_id.is_empty() {
|
||||
let rc = RunContext::with_security_group(Some(rc_id.clone()));
|
||||
if let Err(e) = self.run_context_manager.set_session_run_context(session_id, Some(&rc)).await {
|
||||
warn!(error = %e, rc_id, "TicManager: failed to set run context");
|
||||
match self.tick(user_id, pool, sessions, hub, events).await {
|
||||
Ok(run) => {
|
||||
system_agent_runs::finish(
|
||||
pool,
|
||||
run_id,
|
||||
system_agent_runs::STATUS_COMPLETED,
|
||||
Some(run.session_id),
|
||||
started.elapsed().as_millis() as i64,
|
||||
Some(&run.stats_json()),
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
info!(
|
||||
user = %user_id,
|
||||
events = run.events_processed,
|
||||
notifications = run.notifications_emitted,
|
||||
"TIC: tick complete",
|
||||
);
|
||||
Ok(Some(run))
|
||||
}
|
||||
Err(e) => {
|
||||
// Best-effort: the tick already failed, a failing log write must not
|
||||
// mask the original error.
|
||||
if let Err(log_err) = system_agent_runs::finish(
|
||||
pool,
|
||||
run_id,
|
||||
system_agent_runs::STATUS_FAILED,
|
||||
None,
|
||||
started.elapsed().as_millis() as i64,
|
||||
None,
|
||||
Some(&e.to_string()),
|
||||
)
|
||||
.await
|
||||
{
|
||||
warn!(user = %user_id, error = %log_err, "TIC: failed to record the failed run");
|
||||
}
|
||||
Err(e)
|
||||
}
|
||||
}
|
||||
|
||||
// 6. Sink for session events — nobody subscribes; drop the receiver immediately
|
||||
// so the channel is drained without buffering.
|
||||
let (tx, _rx) = mpsc::channel(32);
|
||||
let notify = crate::tools::notify::make_tool(Arc::clone(&self.hub), "TIC");
|
||||
|
||||
handler.handle_message(&prompt, None, None, None, None, vec![notify], std::collections::HashMap::new(), tx, true, None, None).await?;
|
||||
|
||||
info!(session_id, count = events.len(), "TicManager: tick complete");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn tick(
|
||||
&self,
|
||||
user_id: &str,
|
||||
pool: &SqlitePool,
|
||||
sessions: &Arc<ChatSessionManager>,
|
||||
hub: &Arc<ChatHub>,
|
||||
events: Vec<mcp_events::McpEvent>,
|
||||
) -> anyhow::Result<TicRun> {
|
||||
info!(user = %user_id, count = events.len(), "TIC: processing event batch");
|
||||
|
||||
// Mark as processed BEFORE running the agent — a crash mid-turn then costs
|
||||
// this batch rather than replaying it forever. The loss is visible: the run
|
||||
// row closes as `failed` with the error.
|
||||
let ids: Vec<i64> = events.iter().map(|e| e.id).collect();
|
||||
mcp_events::mark_processed(pool, &ids).await?;
|
||||
|
||||
let prompt = build_prompt(&events);
|
||||
let rc = self.run_context_for(user_id).await;
|
||||
|
||||
// A fresh ephemeral session per tick (agent_id = "tic", source = "tic").
|
||||
// ChatHub is bypassed: TIC is not a user-facing source and must not take
|
||||
// over the `sources` row of a conversation the user is having.
|
||||
let (session_id, _) = sessions
|
||||
.create_session(TIC_AGENT, TIC_SOURCE, false, true, rc.as_ref())
|
||||
.await?;
|
||||
let handler = sessions.get_or_create_handler(session_id).await?;
|
||||
handler.set_auto_deny_approvals();
|
||||
|
||||
// The session's event stream has no subscriber, but the translator awaits
|
||||
// its sends — a receiver that is merely dropped, or kept and never polled,
|
||||
// wedges the turn at the channel's capacity. Drain it explicitly.
|
||||
let (tx, mut rx) = mpsc::channel(32);
|
||||
tokio::spawn(async move { while rx.recv().await.is_some() {} });
|
||||
|
||||
let (notify, emitted) = counting_notify(Arc::clone(hub));
|
||||
|
||||
handler
|
||||
.handle_message(
|
||||
&prompt,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
vec![notify],
|
||||
std::collections::HashMap::new(),
|
||||
tx,
|
||||
true,
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
|
||||
Ok(TicRun {
|
||||
session_id,
|
||||
events_processed: events.len(),
|
||||
notifications_emitted: emitted.load(Ordering::Relaxed),
|
||||
})
|
||||
}
|
||||
|
||||
/// The security group for this user's tick.
|
||||
///
|
||||
/// The configured group is an instance-wide admin setting, so it cannot be
|
||||
/// applied verbatim to somebody else's session: that would hand a restricted
|
||||
/// member's TIC run a tool set their role never granted. It goes through the
|
||||
/// same seam a persisted group does — [`run_context::reconcile_group_for_user`],
|
||||
/// which degrades it to the user's role default when their role does not allow
|
||||
/// it. With nothing configured we still start from the role default rather than
|
||||
/// `None`, because `None` means the catch-all group, which is *wider*.
|
||||
async fn run_context_for(&self, user_id: &str) -> Option<RunContext> {
|
||||
let configured = self
|
||||
.config_store
|
||||
.get(TIC_SECURITY_GROUP_KEY)
|
||||
.await
|
||||
.ok()
|
||||
.flatten()
|
||||
.filter(|g| !g.is_empty());
|
||||
|
||||
match configured {
|
||||
Some(group) => {
|
||||
let wanted = RunContext::with_security_group(Some(group));
|
||||
run_context::reconcile_group_for_user(&self.registry_pool, user_id, Some(wanted)).await
|
||||
}
|
||||
None => run_context::role_default_run_context(&self.registry_pool, user_id).await,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Wrap the `notify` tool so the run log can report how many notifications the
|
||||
/// tick actually produced, without the tool itself knowing it is being counted.
|
||||
fn counting_notify(hub: Arc<ChatHub>) -> (InterfaceTool, Arc<AtomicUsize>) {
|
||||
let inner = crate::tools::notify::make_tool(hub, "TIC");
|
||||
let counter = Arc::new(AtomicUsize::new(0));
|
||||
|
||||
let handler = {
|
||||
let counter = Arc::clone(&counter);
|
||||
let call = Arc::clone(&inner.handler);
|
||||
Arc::new(move |args: serde_json::Value| {
|
||||
let counter = Arc::clone(&counter);
|
||||
let fut = call(args);
|
||||
Box::pin(async move {
|
||||
let out = fut.await;
|
||||
if out.is_ok() {
|
||||
counter.fetch_add(1, Ordering::Relaxed);
|
||||
}
|
||||
out
|
||||
}) as ToolFuture
|
||||
})
|
||||
};
|
||||
|
||||
(InterfaceTool { definition: inner.definition, handler }, counter)
|
||||
}
|
||||
|
||||
// ── Prompt builder ─────────────────────────────────────────────────────────────
|
||||
|
||||
Reference in New Issue
Block a user