Daniele
2dad4824c9
The `## MCP servers` table lives inside the frozen system prefix, which PrefixCache holds for twenty idle minutes. Refreshing a user's global-access snapshot fixed what `mcp.tools()` offers but left the table describing the world before the change, so an admin could enable a connector, ask for it in an open conversation, and be told in good faith that it does not exist — with the tools sitting right there. Same gap on a reinstall, whose new llm_short_description reached the runtime and not the prompt. Both refreshes now call `invalidate_prefixes()` on the live contexts they were already iterating, the seam the skill tools use. Order matters and runs against the intuition: `render_mcp_list` renders the runtime's in-RAM state, not the DB, so the invalidation goes last — after the snapshot refresh and after the servers restart. Rebuild earlier and the prefix is repopulated from the very descriptions being replaced, with nothing left to invalidate it a second time. In the reinstall that means waiting out a dependency install; those users were already reading a stale table, and an early rebuild would only freeze the stale one in place. Also warn when a feed's connector.json and index disagree on the integer version. The manifest silently wins, so if the index is the lower of the two the strict `feed > installed` comparison is false forever: the connector never offers an Update and nothing anywhere says why.
532 lines
28 KiB
Rust
532 lines
28 KiB
Rust
//! The logical API surface of `Skald`: one accessor per manager, named after the
|
|
//! historical field, delegating into the domain bundle that now owns it. This is
|
|
//! the intentional surface consumers (frontend handlers, plugin context) use — the
|
|
//! bundles themselves stay internal.
|
|
//!
|
|
//! Now that the core is its own crate, this is a real boundary rather than a
|
|
//! convention: everything here is `pub` because the `skald` binary lives outside,
|
|
//! and everything not here is unreachable from it. Promote the block to a
|
|
//! `SkaldApi` trait if the shells ever need to mock it.
|
|
|
|
use std::sync::Arc;
|
|
|
|
use sqlx::SqlitePool;
|
|
use tokio::sync::RwLock;
|
|
use tokio_util::sync::CancellationToken;
|
|
|
|
use core_api::remote::RemoteAccess;
|
|
use core_api::system_bus::SystemEventBus;
|
|
use core_api::user_channel::UserChannelApi;
|
|
|
|
use crate::approval::ApprovalManager;
|
|
use crate::chat_event_bus::ChatEventBus;
|
|
use crate::chat_hub::ChatHub;
|
|
use crate::clarification::ClarificationManager;
|
|
use crate::command::LlmCommandManager;
|
|
use crate::config_store::GlobalConfigManager;
|
|
use crate::cron::TaskManager;
|
|
use crate::elicitation::ElicitationManager;
|
|
use crate::image_generate::ImageGeneratorManager;
|
|
use crate::inbox::Inbox;
|
|
use crate::latex::LatexCompiler;
|
|
use crate::llm::LlmManager;
|
|
use crate::location::LocationManager;
|
|
use crate::mcp::McpManager;
|
|
use crate::memory::MemoryManager;
|
|
use crate::plugin::PluginManager;
|
|
use crate::provider::ProviderRegistry;
|
|
use crate::run_context::RunContextManager;
|
|
use crate::secrets::SecretsStore;
|
|
use crate::session::manager::ChatSessionManager;
|
|
use crate::system_agents::{AgentRunCtx, AgentScope, ManualRun, ManualRunError, SystemAgents};
|
|
use crate::tool_catalog::ToolCatalog;
|
|
use crate::tools::ToolRegistry;
|
|
use crate::transcribe::TranscribeManager;
|
|
use crate::tts::TtsManager;
|
|
use crate::users::UserManager;
|
|
|
|
use super::Skald;
|
|
|
|
impl Skald {
|
|
// Runtime / cross-cutting
|
|
pub fn db(&self) -> &Arc<SqlitePool> { &self.rt.db }
|
|
pub fn users(&self) -> &Arc<UserManager> { &self.rt.users }
|
|
|
|
/// The caller's per-user owner-bound runtime (chat/hub/cron/interaction),
|
|
/// built lazily on first use. `None` when the user's database is still locked
|
|
/// (not logged in). The pool is the unlock token (§9); a present pool means an
|
|
/// unlocked database, so a context can be built for it.
|
|
pub async fn user_context(&self, user_id: &str) -> Option<Arc<super::UserContext>> {
|
|
let pool = self.rt.users.pool_of(user_id)?;
|
|
self.rt_user_contexts().resolve(user_id, pool).await.ok()
|
|
}
|
|
|
|
fn rt_user_contexts(&self) -> &super::user_context::UserContextRegistry { &self.user_contexts }
|
|
|
|
/// Declares the tools the running surface contributes to a chat session
|
|
/// (the SPA's `show_file_to_user`, …). Called once by the shell after
|
|
/// construction: the core owns the tools, the shell owns the policy of who
|
|
/// gets them. Every per-user hub built from here on receives it, and every
|
|
/// path that starts or resumes a turn consults it — see
|
|
/// [`crate::chat_hub::InterfaceToolsBuilder`].
|
|
pub fn set_interface_tools_builder(&self, build: crate::chat_hub::InterfaceToolsBuilder) {
|
|
self.rt_user_contexts().set_interface_tools_builder(build);
|
|
}
|
|
|
|
/// The user's runtime context IF it is already live (built), **without**
|
|
/// building one — used to refresh a logged-in user in place. A user who never
|
|
/// logged in has no snapshot to refresh; their next login builds a fresh one.
|
|
pub async fn user_context_if_live(&self, user_id: &str) -> Option<Arc<super::UserContext>> {
|
|
self.rt_user_contexts().peek(user_id).await
|
|
}
|
|
|
|
/// Revokes a user's live runtime: sessions, owner-bound loops, database key.
|
|
///
|
|
/// Called when a user is **deactivated or deleted**. Writing `active = 0` (or
|
|
/// deleting the row) only stops the *next* login: `login` checks the flag, but
|
|
/// `require_auth` maps token → id without re-reading the row, so a session minted
|
|
/// before the change would keep working, over a pool whose key is still in RAM.
|
|
///
|
|
/// The order is load-bearing:
|
|
///
|
|
/// 1. **Revoke the sessions** — the moment this returns, no token authenticates
|
|
/// as this user.
|
|
/// 2. **Evict the context** — cancels their cron loop, hub and per-user MCP
|
|
/// runtime, so nothing is left to query the pool we are about to close.
|
|
/// 3. **Lock the database** — `close()`s the pool, which invalidates every
|
|
/// surviving clone and drops the DEK (§9). The user is opaque again.
|
|
///
|
|
/// Synchronous by design: this is an authorization invariant, not reconciliation,
|
|
/// so it must not ride the lossy system bus. The Docker half (stop or remove the
|
|
/// container) *is* reconciliation and does ride it.
|
|
///
|
|
/// Idempotent — a user with no live session and a locked database is a no-op.
|
|
pub async fn revoke_user_runtime(&self, user_id: &str) {
|
|
self.sessions().revoke_user(user_id);
|
|
self.rt_user_contexts().evict(user_id).await;
|
|
self.rt.users.lock(user_id).await;
|
|
}
|
|
|
|
/// Re-checks a live user's open sessions against their current role, degrading any
|
|
/// security group the role no longer allows, and tells their open tabs about it.
|
|
///
|
|
/// The durable half of this is in `ChatSessionManager::get_or_create_handler`,
|
|
/// which reconciles on every load; this is the liveness half, for sessions already
|
|
/// in RAM. Synchronous, like [`Self::revoke_user_runtime`] and for the same reason:
|
|
/// narrowing someone's permissions is an authorization change, not reconciliation.
|
|
///
|
|
/// No-op for a user who is not logged in — their next login loads through the
|
|
/// reconcile anyway.
|
|
pub async fn revalidate_security_groups_for_user(&self, user_id: &str) {
|
|
let Some(ctx) = self.user_context_if_live(user_id).await else { return };
|
|
for (session_id, source, group) in ctx.sessions.revalidate_security_groups().await {
|
|
ctx.chat_hub.emit(core_api::events::GlobalEvent {
|
|
source: Some(source),
|
|
// Tagged with the conversation: clients filter per conversation, so
|
|
// an untagged degrade would leave every pill showing the old group.
|
|
session_id: Some(session_id),
|
|
event: core_api::events::ServerEvent::SecurityGroupSelected { group },
|
|
});
|
|
}
|
|
}
|
|
|
|
/// [`Self::revalidate_security_groups_for_user`] for every member of a role —
|
|
/// called when the role's own group set changes, which can narrow many users at
|
|
/// once. Members who are not logged in need nothing.
|
|
pub async fn revalidate_security_groups_for_role(&self, role_id: &str) {
|
|
let users = match crate::db::users::list(self.db()).await {
|
|
Ok(u) => u,
|
|
Err(e) => {
|
|
tracing::warn!(role = %role_id, error = %e,
|
|
"cannot list users to revalidate security groups");
|
|
return;
|
|
}
|
|
};
|
|
for user in users.into_iter().filter(|u| u.role_id == role_id) {
|
|
self.revalidate_security_groups_for_user(&user.id).await;
|
|
}
|
|
}
|
|
|
|
/// Applies a shared-folder membership change to a user (blueprint §6 remount).
|
|
///
|
|
/// A container's bind mounts are fixed at `docker create` time, so the mount set
|
|
/// changes only by recreating the container — done here with a graceful stop
|
|
/// first ([`ContainerManager::recreate`](crate::container::ContainerManager::recreate)).
|
|
/// If the user is **live**, the two snapshot-bound pieces are then refreshed in
|
|
/// place against the fresh container: their filesystem view (which governs both
|
|
/// the host-side fs-tools and `execute_cmd` path routing) and their per-user MCP
|
|
/// runtime (whose `docker exec` children died with the old container). A user
|
|
/// with no live context needs only the recreate — their next login builds a
|
|
/// context that already reflects the change.
|
|
///
|
|
/// Best-effort by contract: the membership row is already committed, so a Docker
|
|
/// hiccup must not fail the caller; the state settles at the next login/boot.
|
|
///
|
|
/// Covers both shared-folder and project membership changes — both feed
|
|
/// `build_user_fs`, so a recreate reflows either mount set.
|
|
pub async fn refresh_user_mounts(&self, user_id: &str) -> anyhow::Result<()> {
|
|
// New mount topology (graceful stop → remove → recreate from current rows).
|
|
self.container().recreate(user_id).await?;
|
|
|
|
let Some(ctx) = self.user_context_if_live(user_id).await else {
|
|
return Ok(()); // not logged in — next login builds a fresh context
|
|
};
|
|
|
|
// fs view: swap the shared cell so every live session picks it up next call.
|
|
let new_fs = crate::container::build_user_fs(self.db(), user_id).await?;
|
|
ctx.sessions.refresh_fs(new_fs);
|
|
|
|
// per-user MCP: the old container's `docker exec` children are gone. Stop the
|
|
// stale handles, then reconnect the activated connectors against the fresh
|
|
// container (same deterministic name).
|
|
ctx.user_mcp.stop_all();
|
|
let rows = crate::db::mcp_user_servers::all_startable(&ctx.pool).await.unwrap_or_default();
|
|
if !rows.is_empty() {
|
|
let container = crate::container::container_name(user_id);
|
|
let mut specs = Vec::with_capacity(rows.len());
|
|
for r in &rows {
|
|
// The home mount (and its `node_modules`/`.pydeps`) survives a
|
|
// recreate, so this is normally a hash-match no-op; it still covers
|
|
// the case where the source changed while the user was logged in.
|
|
crate::mcp::prepare_local_connector(self.db(), user_id, &container, r).await;
|
|
specs.push(crate::mcp::user_row_spec_resolved(r, &container, self.db()).await);
|
|
}
|
|
ctx.user_mcp.connect_all(specs, false).await;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Rebuilds the frozen system prefix of the conversations a skills change made
|
|
/// wrong (blueprint §6).
|
|
///
|
|
/// The prefix is normally left alone until its conversation has been idle for
|
|
/// twenty minutes, which is right for an *injected file* and wrong for the
|
|
/// **index**: an admin who installs a skill and then asks the assistant to use
|
|
/// it would be told, at length and in good faith, that no such skill exists.
|
|
///
|
|
/// Called **directly** by the two skill tools, not through the system bus.
|
|
/// Whoever writes a skill through a tool is inside this process and can say so;
|
|
/// the bus (`SkillsChanged`) is for the other case — someone editing files on
|
|
/// the box — where nothing in-process knows. A miss here is not a lost event,
|
|
/// it is a user who is simply not logged in and whose next login builds a fresh
|
|
/// prefix anyway.
|
|
pub async fn invalidate_prompt_prefix(&self, scope: crate::skills::PromptScope) {
|
|
for ctx in self.rt_user_contexts().all_live().await {
|
|
let concerns = match &scope {
|
|
// The group's tree is in everybody's index.
|
|
crate::skills::PromptScope::Everyone => true,
|
|
crate::skills::PromptScope::User(id) => *id == ctx.user_id,
|
|
};
|
|
if concerns {
|
|
ctx.sessions.loop_runtime().invalidate_prefixes();
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Refresh every live user's global-connector access set in place — call after an
|
|
/// admin enables/deletes a global connector or changes who may use it, so running
|
|
/// sessions see it without a restart (the §7 MCP twin of the §6 fs remount). The
|
|
/// global runtime itself is already updated by the caller (`start_server` /
|
|
/// `stop_server`); this only re-snapshots each user's access filter. Best-effort:
|
|
/// a locked (not-live) user has no snapshot to refresh — their next login rebuilds
|
|
/// it from the now-current tables.
|
|
pub async fn refresh_global_mcp_access(&self) {
|
|
for ctx in self.rt_user_contexts().all_live().await {
|
|
if let Err(e) = ctx.refresh_global_access().await {
|
|
tracing::warn!(user = %ctx.user_id, error = %e, "failed to refresh global MCP access");
|
|
}
|
|
// Then rebuild the frozen prompt prefix, for the reason spelled out in
|
|
// `invalidate_prompt_prefix`: refreshing the snapshot fixes what `mcp.tools()`
|
|
// *offers*, while the `## MCP servers` table the model reads lives inside
|
|
// `base`, which `PrefixCache` holds for twenty idle minutes. Without this the
|
|
// admin enables a connector, asks for it in an open conversation, and is told
|
|
// in good faith that it does not exist — with the tools sitting right there.
|
|
//
|
|
// After the refresh, never before: the table is rendered from the access
|
|
// snapshot we just replaced.
|
|
ctx.sessions.loop_runtime().invalidate_prefixes();
|
|
}
|
|
}
|
|
|
|
/// Pushes a marketplace **reinstall** into every live copy of the connector so
|
|
/// active sessions pick up the new metadata (`llm_short_description`) and code
|
|
/// without a re-login — the reinstall counterpart of the §6/§7 remount helpers.
|
|
/// The reinstall has already rewritten `mcp_catalog`; this reconnects what runs:
|
|
///
|
|
/// - **Global runtime**: install the connector's declared dependencies on the host
|
|
/// (`ensure_installed_host`, once per folder), then for each *enabled*
|
|
/// `mcp_global_servers` row snapshotting this catalog entry, re-snapshot its
|
|
/// `description` from the catalog and restart it, so the running server's in-RAM
|
|
/// description (and code) catches up.
|
|
/// - **Per-user runtimes**: for each live user who has this connector *startable*,
|
|
/// re-copy its files/deps into the container (`prepare_local_connector` — a hash
|
|
/// no-op when the source is unchanged) and restart that one server. The rebuilt
|
|
/// spec now carries the fresh catalog description (see `user_row_spec_resolved`).
|
|
/// - **Prompt prefix**: last, invalidate it for every live user, so the
|
|
/// `## MCP servers` table stops describing the version that was just replaced.
|
|
///
|
|
/// Best-effort: the catalog write already committed, so a Docker/MCP hiccup here
|
|
/// must not fail the reinstall — anything not refreshed settles at the user's next
|
|
/// login. A fresh install (nothing live yet) is a cheap no-op: no row matches.
|
|
pub async fn refresh_connector_after_reinstall(&self, catalog_name: &str) {
|
|
// The metadata the reinstall just wrote — the source of truth to push out.
|
|
let entry = match crate::db::mcp_catalog::get_by_name(self.db(), catalog_name).await {
|
|
Ok(Some(e)) => e,
|
|
Ok(None) => return,
|
|
Err(e) => {
|
|
tracing::warn!(connector = %catalog_name, error = %e, "reinstall refresh: catalog lookup failed");
|
|
return;
|
|
}
|
|
};
|
|
|
|
// 1. Global runtime.
|
|
if let Ok(globals) = crate::db::mcp_global_servers::all_enabled(self.db()).await {
|
|
let live: Vec<_> = globals
|
|
.iter()
|
|
.filter(|g| g.catalog_name.as_deref() == Some(catalog_name))
|
|
.collect();
|
|
|
|
// Dependencies before code. A global connector runs on the host, where
|
|
// nothing reconciles it the way the container reconciler does below, and
|
|
// `ensure_installed_host` was otherwise reachable from `global_enable`
|
|
// alone — so an Update that *adds* a `requirements.txt` landed the file,
|
|
// restarted the server, and never installed what it declared: the
|
|
// connector came back exactly as broken as before, curable only by
|
|
// re-saving its config from the UI.
|
|
//
|
|
// Once per connector folder rather than per row: the deps live beside the
|
|
// files, so two runtime names snapshotting one catalog entry share them.
|
|
// Not hash-guarded, unlike the per-user `ensure_installed` — it leans on
|
|
// `pip`/`npm` being idempotent, so a no-change reinstall pays one fast
|
|
// satisfied-requirements pass. Best-effort like the rest of this function.
|
|
if !live.is_empty() && entry.source == "local_script" {
|
|
match entry.script_path.as_deref().map(crate::mcp::split_script_path) {
|
|
Some(Ok((folder, _))) => {
|
|
if let Err(e) = crate::mcp::ensure_installed_host(folder).await {
|
|
tracing::warn!(connector = %catalog_name, error = %e, "reinstall refresh: global dependency install failed");
|
|
}
|
|
}
|
|
Some(Err(e)) => tracing::warn!(connector = %catalog_name, error = %e, "reinstall refresh: unusable script_path, skipping dependency install"),
|
|
None => tracing::warn!(connector = %catalog_name, "reinstall refresh: local_script entry has no script_path, skipping dependency install"),
|
|
}
|
|
}
|
|
|
|
for g in live {
|
|
if let Err(e) = crate::db::mcp_global_servers::set_description(self.db(), g.id, entry.description.as_deref()).await {
|
|
tracing::warn!(connector = %catalog_name, error = %e, "reinstall refresh: failed to update global description");
|
|
continue;
|
|
}
|
|
match crate::db::mcp_global_servers::get(self.db(), g.id).await {
|
|
Ok(Some(row)) => {
|
|
let spec = crate::mcp::global_row_spec(&row);
|
|
if let Err(e) = self.mcp().start_server(spec).await {
|
|
tracing::warn!(connector = %catalog_name, error = %e, "reinstall refresh: failed to restart global server");
|
|
}
|
|
}
|
|
_ => tracing::warn!(connector = %catalog_name, "reinstall refresh: global row vanished before restart"),
|
|
}
|
|
}
|
|
}
|
|
|
|
// 2. Per-user runtimes — restart this one connector for each live user who runs it.
|
|
for ctx in self.rt_user_contexts().all_live().await {
|
|
let rows = crate::db::mcp_user_servers::all_startable(&ctx.pool).await.unwrap_or_default();
|
|
let Some(row) = rows.into_iter().find(|r| r.catalog_name.as_deref() == Some(catalog_name)) else {
|
|
continue;
|
|
};
|
|
let container = crate::container::container_name(&ctx.user_id);
|
|
crate::mcp::prepare_local_connector(self.db(), &ctx.user_id, &container, &row).await;
|
|
let spec = crate::mcp::user_row_spec_resolved(&row, &container, self.db()).await;
|
|
if let Err(e) = ctx.user_mcp.start_server(spec).await {
|
|
tracing::warn!(user = %ctx.user_id, connector = %catalog_name, error = %e, "reinstall refresh: failed to restart per-user connector");
|
|
}
|
|
}
|
|
|
|
// 3. Rebuild the frozen prompt prefix, for everyone — a reinstall changes the
|
|
// connector's `llm_short_description`, which the model reads from the
|
|
// `## MCP servers` table inside `base` rather than from the runtime it just
|
|
// reconnected to. Restarting the servers alone left the prompt describing the
|
|
// old version for up to twenty idle minutes.
|
|
//
|
|
// **Last, deliberately.** `render_mcp_list` renders the live runtime's in-RAM
|
|
// state, so a prefix rebuilt before the restarts above would be repopulated
|
|
// from the descriptions we are in the middle of replacing — and nothing would
|
|
// invalidate it a second time. That the global dependency install can take
|
|
// minutes is not a reason to move this earlier: those users were already
|
|
// reading a stale table, and rebuilding it early would only freeze the stale
|
|
// one in place.
|
|
//
|
|
// Everyone, not just the users who run this connector per-user: an enabled
|
|
// global connector is in every granted user's table.
|
|
for ctx in self.rt_user_contexts().all_live().await {
|
|
ctx.sessions.loop_runtime().invalidate_prefixes();
|
|
}
|
|
}
|
|
pub fn sessions(&self) -> &Arc<crate::auth::SessionStore> { &self.rt.sessions }
|
|
pub fn config(&self) -> &Arc<GlobalConfigManager> { &self.rt.config }
|
|
pub fn config_properties(&self) -> &[core_api::ConfigSet] { &self.rt.config_properties }
|
|
pub fn system_bus(&self) -> &Arc<SystemEventBus> { &self.rt.system_bus }
|
|
pub fn event_bus(&self) -> &Arc<ChatEventBus> { &self.rt.event_bus }
|
|
pub fn shutdown_token(&self) -> &CancellationToken { &self.rt.shutdown_token }
|
|
|
|
// Models
|
|
pub fn provider_registry(&self) -> &Arc<ProviderRegistry> { &self.models.provider_registry }
|
|
pub fn llm_manager(&self) -> &Arc<LlmManager> { &self.models.llm_manager }
|
|
pub fn secrets(&self) -> &Arc<SecretsStore> { &self.models.secrets }
|
|
pub fn memory_manager(&self) -> &Arc<MemoryManager> { &self.models.memory_manager }
|
|
|
|
// Media
|
|
pub fn image_generator_manager(&self) -> &Arc<ImageGeneratorManager> { &self.media.image_generator_manager }
|
|
pub fn transcribe_manager(&self) -> &Arc<TranscribeManager> { &self.media.transcribe_manager }
|
|
pub fn tts_manager(&self) -> &Arc<TtsManager> { &self.media.tts_manager }
|
|
|
|
// Tools
|
|
pub fn tools(&self) -> &Arc<ToolRegistry> { &self.tools.tools }
|
|
pub fn catalog(&self) -> &ToolCatalog { &self.tools.catalog }
|
|
pub fn command_manager(&self) -> &Arc<LlmCommandManager> { &self.tools.command_manager }
|
|
|
|
// Integrations
|
|
pub fn mcp(&self) -> &Arc<McpManager> { &self.integrations.mcp }
|
|
pub fn plugin_manager(&self) -> &Arc<PluginManager> { &self.integrations.plugin_manager }
|
|
|
|
// Tasks
|
|
pub fn cron(&self) -> &Arc<TaskManager> { &self.tasks.cron }
|
|
|
|
// Conversation
|
|
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 }
|
|
|
|
// Interaction
|
|
pub fn approval(&self) -> &Arc<ApprovalManager> { &self.interaction.approval }
|
|
pub fn inbox(&self) -> &Inbox { &self.interaction.inbox }
|
|
pub fn clarification(&self) -> &Arc<ClarificationManager> { &self.interaction.clarification }
|
|
pub fn elicitation(&self) -> &Arc<ElicitationManager> { &self.interaction.elicitation }
|
|
|
|
// Infra
|
|
pub fn latex_compiler(&self) -> &LatexCompiler { &self.infra.latex_compiler }
|
|
pub fn location_manager(&self) -> &Arc<LocationManager> { &self.infra.location_manager }
|
|
pub fn remote(&self) -> &Arc<RwLock<Option<Arc<dyn RemoteAccess>>>> { &self.infra.remote }
|
|
|
|
// System agents
|
|
pub fn system_agents(&self) -> &Arc<SystemAgents> { &self.system_agents }
|
|
|
|
/// Start one system-agent pass **now**, for `user_id`, because a human asked.
|
|
///
|
|
/// The schedule answers *when* a pass runs, and the button is a person saying
|
|
/// "now" — so due-ness is skipped, exactly as manual `/compact` skips the
|
|
/// compactor's token threshold. The instance-wide **Enabled** switch is a
|
|
/// different kind of setting and is honoured: it says *whether* the agent runs
|
|
/// at all, and that is the admin's answer, not the caller's.
|
|
///
|
|
/// The pass always runs **as the caller** — their pool, their sessions, their
|
|
/// hub — so a member triggering the shared-memory lint gets their own report
|
|
/// over the shared store, and their own run row. One consequence worth naming:
|
|
/// the attempt is marked in the file the pass ran in, so a member's manual run
|
|
/// of an instance-wide agent does not move the admin's scheduled clock. The two
|
|
/// clocks were always per file; this only makes it visible.
|
|
///
|
|
/// Returns as soon as the work is **scheduled**, not when it finishes: a pass is
|
|
/// an LLM turn and no HTTP request should be held open for it. The run log is
|
|
/// the progress surface — the `running` row exists before this returns to the
|
|
/// browser. The one thing answered synchronously is
|
|
/// [`SystemAgent::has_work`], which is cheap by contract and whose `false`
|
|
/// leaves no row at all: without it the button would report "started" and the
|
|
/// log would stay empty forever.
|
|
pub async fn run_system_agent_now(
|
|
self: &Arc<Self>,
|
|
agent_id: &str,
|
|
user_id: &str,
|
|
) -> Result<ManualRun, ManualRunError> {
|
|
let agent = self.system_agents.get(agent_id).ok_or(ManualRunError::UnknownAgent)?.clone();
|
|
|
|
// A per-subject pass is about somebody else and picks its own subjects;
|
|
// "run it for me" has no meaning for it.
|
|
if agent.scope() == AgentScope::PerSubject {
|
|
return Err(ManualRunError::Unsupported);
|
|
}
|
|
if !agent.is_enabled().await {
|
|
return Err(ManualRunError::Disabled);
|
|
}
|
|
|
|
let ctx = self.user_context(user_id).await.ok_or(ManualRunError::Locked)?;
|
|
|
|
// Taken before `has_work` so that two quick clicks cannot both look, both
|
|
// find work, and both start.
|
|
let claim = self
|
|
.system_agents
|
|
.claim(agent.id(), &SystemAgents::target_of(agent.as_ref(), user_id))
|
|
.ok_or(ManualRunError::AlreadyRunning)?;
|
|
|
|
let run_ctx = AgentRunCtx {
|
|
user_id,
|
|
pool: &ctx.pool,
|
|
sessions: &ctx.sessions,
|
|
hub: &ctx.chat_hub,
|
|
subject: None,
|
|
run_id: None,
|
|
};
|
|
if !agent.has_work(&run_ctx).await.map_err(ManualRunError::Failed)? {
|
|
return Ok(ManualRun::NothingToDo);
|
|
}
|
|
|
|
let user_id = user_id.to_string();
|
|
self.rt.supervisor.spawn("system-agent-manual", async move {
|
|
// Released when the task ends, whichever way it ends.
|
|
let _claim = claim;
|
|
let run_ctx = AgentRunCtx {
|
|
user_id: &user_id,
|
|
pool: &ctx.pool,
|
|
sessions: &ctx.sessions,
|
|
hub: &ctx.chat_hub,
|
|
subject: None,
|
|
run_id: None,
|
|
};
|
|
if let Err(e) = crate::system_agents::run_and_record(agent.as_ref(), &run_ctx).await {
|
|
// The failure is already recorded on the run row, which is where
|
|
// the person who pressed the button will look for it.
|
|
tracing::warn!(agent = agent.id(), user = %user_id, error = %e,
|
|
"system-agents: manual pass failed");
|
|
}
|
|
});
|
|
|
|
Ok(ManualRun::Started)
|
|
}
|
|
}
|
|
|
|
// ── UserChannelApi ────────────────────────────────────────────────────────────
|
|
|
|
use super::user_context::UserContextHandle;
|
|
|
|
#[async_trait::async_trait]
|
|
impl UserChannelApi for Skald {
|
|
async fn resolve_user(&self, user_id: &str) -> Option<std::sync::Arc<dyn core_api::user_channel::UserChannelHandle>> {
|
|
let ctx = self.user_context(user_id).await?;
|
|
Some(std::sync::Arc::new(UserContextHandle::new(ctx)))
|
|
}
|
|
|
|
async fn plugin_access(&self, plugin_id: &str, user_id: &str) -> bool {
|
|
// Admin short-circuit + grant lookup live in `db::plugin_access`; a
|
|
// lookup error fails closed.
|
|
crate::db::plugin_access::effective_access(self.db(), plugin_id, user_id)
|
|
.await
|
|
.unwrap_or(false)
|
|
}
|
|
|
|
async fn is_admin(&self, user_id: &str) -> bool {
|
|
// Built-in admin role; an unknown user or a lookup error fails closed.
|
|
sqlx::query_as::<_, (String,)>("SELECT role_id FROM users WHERE id = ?")
|
|
.bind(user_id)
|
|
.fetch_optional(self.db().as_ref())
|
|
.await
|
|
.ok()
|
|
.flatten()
|
|
.map(|(r,)| r == crate::db::roles::ADMIN_ROLE_ID)
|
|
.unwrap_or(false)
|
|
}
|
|
|
|
async fn user_for_session(&self, token: &str) -> Option<String> {
|
|
self.sessions().user_of(token)
|
|
}
|
|
}
|