Multuser part 1

This commit is contained in:
2026-07-10 22:05:25 +01:00
parent 088c0c01cb
commit c72f18b0d6
15 changed files with 398 additions and 544 deletions
-1
View File
@@ -109,7 +109,6 @@ impl PluginManager {
tts_provider: Arc::clone(skald.tts_manager()) as _,
api_provider_registry: Arc::clone(skald.provider_registry()) as _,
location: Arc::clone(skald.location_manager()) as _,
event_bus: Arc::clone(skald.event_bus()),
system_bus: Arc::clone(skald.system_bus()),
web_port,
remote_slot: Arc::clone(skald.remote()),
@@ -9,7 +9,7 @@ use crate::chatbot::{LlmTurn, ToolCall};
use crate::db::{chat_history, chat_llm_tools};
use crate::events::ServerEvent;
use crate::tools::{
ExecutionOutcome, SimpleExecution, ToolDescriptionLength, ToolExecution, ToolResult,
ExecutionOutcome, SimpleExecution, ToolContext, ToolDescriptionLength, ToolExecution, ToolResult,
};
use futures::stream::{self, StreamExt};
@@ -426,6 +426,9 @@ impl ChatSessionHandler {
}
// Built-in registry tools (incl. execute_cmd, whose SimpleExecution kills
// the child via kill_on_drop when the work future is dropped on /stop).
self.tools.run(name, args)
// The ToolContext carries this session's id and owner pool so owner-bound
// registry tools (e.g. cron management) act on the caller's own database.
let ctx = ToolContext { session_id: self.session_id, pool: Arc::clone(&self.db) };
self.tools.run(name, &ctx, args)
}
}
+11
View File
@@ -52,6 +52,17 @@ 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 }
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 }
+1 -1
View File
@@ -226,7 +226,7 @@ impl Tools {
Arc::clone(&integrations.mcp), Arc::clone(&integrations.plugin_manager), Arc::clone(&tasks.cron)));
tool_registry.register(crate::tools::register_mcp::RegisterMcp::new(Arc::clone(&integrations.mcp)));
tool_registry.register(crate::tools::register_mcp::DeleteMcp::new(Arc::clone(&integrations.mcp)));
tool_registry.register(crate::tools::cron_jobs::DeleteCronJob(Arc::clone(&tasks.cron)));
tool_registry.register(crate::tools::cron_jobs::DeleteCronJob);
tool_registry.register(crate::tools::set_secret::SetSecret(Arc::clone(&models.secrets)));
tool_registry.register(crate::tools::list_secrets::ListSecrets(Arc::clone(&models.secrets)));
tool_registry.register(crate::tools::configure_plugin::ConfigurePlugin(Arc::clone(&integrations.plugin_manager)));
+14
View File
@@ -23,10 +23,13 @@ mod accessors;
mod bundles;
mod runtime;
mod supervisor;
mod user_context;
mod wiring;
use bundles::{Conversation, Infra, Integrations, Interaction, Media, Models, Tasks, Tools};
use runtime::Runtime;
use user_context::{UserContextFactory, UserContextRegistry};
pub use user_context::UserContext;
use wiring::{spawn_background, wire};
pub struct Skald {
@@ -39,6 +42,10 @@ pub struct Skald {
conversation: Conversation,
interaction: Interaction,
infra: Infra,
/// Per-user owner-bound runtimes (chat/hub/cron/interaction), built lazily on
/// first use after a user's pool is unlocked. The global bundles above still
/// serve deferred subsystems and the not-yet-migrated call sites.
user_contexts: UserContextRegistry,
}
impl Skald {
@@ -68,8 +75,15 @@ impl Skald {
wire(&tasks, &conversation, &integrations, &interaction);
spawn_background(&rt, &tasks, &conversation, &integrations, config);
// Per-user context factory: captures the global capability managers, so a
// per-user chat/hub/cron/interaction stack can be stamped out on demand.
let user_contexts = UserContextRegistry::new(UserContextFactory::new(
&rt, &models, &media, &tools, &integrations, &conversation, config,
));
let skald = Arc::new(Skald {
rt, models, media, tools, integrations, tasks, conversation, interaction, infra,
user_contexts,
});
// Inject the fully-constructed instance into the plugin manager — the one
+238
View File
@@ -0,0 +1,238 @@
//! Per-user runtime context (blueprint §9 / §11 / §5.1).
//!
//! The owner-bound managers — chat sessions, chat hub, cron, and the
//! approval/clarification/elicitation/inbox interaction stack — must operate on
//! *one* user's `{userid}.db` and emit on *one* user's server→client channel, so
//! that no chat content, WS event, job or pending approval crosses between users.
//! They are built **lazily** on first use after the user's pool is unlocked, and
//! live exactly as long as that pool (§9: from first login until restart).
//!
//! `UserManager` stays the pool-lifecycle owner (§11 boundary). This factory sits
//! at the `Skald` layer, where the global *capability* managers (LLM, tools, MCP,
//! memory, providers) are visible, and stamps out the per-user instances against a
//! given pool, wiring their construction cycles and starting the per-user cron
//! loop. Capability managers are shared by reference; only owner-bound state is
//! per-user.
//!
//! Split of pools inside one context: session/history/jobs/hub use the **user
//! pool**; approval *rules* and `known_tools` are instance-wide registry data, so
//! `ApprovalManager` and `ToolDiscovery` read the **registry pool** (`system.db`)
//! while `ApprovalManager` still emits on the user's channel and keeps its own
//! pending map.
use std::collections::HashMap;
use std::sync::Arc;
use anyhow::Result;
use chrono_tz::Tz;
use sqlx::SqlitePool;
use tokio::sync::{broadcast, Mutex};
use tokio_util::sync::CancellationToken;
use core_api::events::GlobalEvent;
use core_api::system_bus::SystemEventBus;
use crate::approval::ApprovalManager;
use crate::chat_event_bus::ChatEventBus;
use crate::chat_hub::ChatHub;
use crate::clarification::ClarificationManager;
use crate::compactor::ContextCompactor;
use crate::config::{CompactionConfig, CoreConfig, DatetimeConfig};
use crate::cron::TaskManager;
use crate::elicitation::ElicitationManager;
use crate::image_generate::ImageGeneratorManager;
use crate::inbox::Inbox;
use crate::llm::LlmManager;
use crate::mcp::McpManager;
use crate::memory::MemoryManager;
use crate::run_context::RunContextManager;
use crate::session::handler::{DEFAULT_MAX_PARALLEL_SUBAGENTS, DEFAULT_MAX_TOOL_ROUNDS};
use crate::session::manager::ChatSessionManager;
use crate::tool_discovery::ToolDiscovery;
use crate::tools::ToolRegistry;
use super::bundles::{Conversation, Integrations, Media, Models, Tools};
use super::runtime::Runtime;
/// One unlocked user's owner-bound runtime. Lifetime = the pool's lifetime.
pub struct UserContext {
pub user_id: String,
pub pool: Arc<SqlitePool>,
pub event_bus: Arc<ChatEventBus>,
pub sessions: Arc<ChatSessionManager>,
pub chat_hub: Arc<ChatHub>,
pub cron: Arc<TaskManager>,
pub approval: Arc<ApprovalManager>,
pub clarification: Arc<ClarificationManager>,
pub elicitation: Arc<ElicitationManager>,
pub inbox: Inbox,
/// Per-user server→client push channel. WS handlers subscribe here (via the
/// hub) so a user's `ServerEvent`s never reach another user's socket.
pub global_tx: broadcast::Sender<GlobalEvent>,
}
/// Captures the global capability managers + resolved config once, and stamps out
/// a [`UserContext`] per unlocked pool.
pub(super) struct UserContextFactory {
registry_pool: Arc<SqlitePool>,
llm_manager: Arc<LlmManager>,
tools: Arc<ToolRegistry>,
mcp: Arc<McpManager>,
memory_manager: Arc<MemoryManager>,
image_generator_manager: Arc<ImageGeneratorManager>,
run_context_manager: Arc<RunContextManager>,
system_bus: Arc<SystemEventBus>,
supervisor: Arc<super::supervisor::TaskSupervisor>,
shutdown_token: CancellationToken,
max_history_messages: usize,
max_tool_rounds: usize,
max_parallel_subagents: usize,
max_tool_result_chars: Option<usize>,
datetime_config: DatetimeConfig,
compaction: Option<CompactionConfig>,
cron_tz: Option<Tz>,
}
impl UserContextFactory {
pub(super) fn new(
rt: &Runtime,
models: &Models,
media: &Media,
tools: &Tools,
integrations: &Integrations,
conversation: &Conversation,
config: &CoreConfig,
) -> Self {
let cron_tz = config.timezone.as_deref().and_then(|s| s.parse::<Tz>().ok());
Self {
registry_pool: Arc::clone(&rt.db),
llm_manager: Arc::clone(&models.llm_manager),
tools: Arc::clone(&tools.tools),
mcp: Arc::clone(&integrations.mcp),
memory_manager: Arc::clone(&models.memory_manager),
image_generator_manager: Arc::clone(&media.image_generator_manager),
run_context_manager: Arc::clone(&conversation.run_context_manager),
system_bus: Arc::clone(&rt.system_bus),
supervisor: Arc::clone(&rt.supervisor),
shutdown_token: rt.shutdown_token.clone(),
max_history_messages: config.llm.max_history_messages,
max_tool_rounds: config.llm.max_tool_rounds.unwrap_or(DEFAULT_MAX_TOOL_ROUNDS),
max_parallel_subagents: config.llm.max_parallel_subagents.unwrap_or(DEFAULT_MAX_PARALLEL_SUBAGENTS),
max_tool_result_chars: config.llm.max_tool_result_chars,
datetime_config: DatetimeConfig { timezone: config.timezone.clone(), ..config.llm.datetime },
compaction: config.llm.compaction.clone(),
cron_tz,
}
}
async fn build(&self, user_id: &str, pool: SqlitePool) -> Result<Arc<UserContext>> {
let pool = Arc::new(pool);
let event_bus = Arc::new(ChatEventBus::new());
let (global_tx, _) = broadcast::channel::<GlobalEvent>(512);
// Interaction stack, per-user. Approval reads the shared registry rules but
// emits on this user's channel and keeps its own pending map — no cross-user
// collision on request_id / session_id. Rules are seeded once by the global
// ApprovalManager at boot, so no re-seeding here.
let approval = Arc::new(ApprovalManager::new(Arc::clone(&self.registry_pool), global_tx.clone()));
let clarification = ClarificationManager::new(global_tx.clone());
let elicitation = ElicitationManager::new(global_tx.clone());
let inbox = Inbox::new(
Arc::clone(&approval),
Arc::clone(&clarification),
Arc::clone(&elicitation),
Arc::clone(&self.tools),
);
let compactor = self.compaction.as_ref().map(|cfg| {
Arc::new(ContextCompactor::new(
cfg.clone(),
Arc::clone(&self.llm_manager),
Arc::clone(&event_bus),
))
});
let manager = Arc::new(ChatSessionManager::new(
Arc::clone(&pool),
Arc::clone(&self.llm_manager),
self.max_history_messages,
self.max_tool_rounds,
self.max_parallel_subagents,
self.max_tool_result_chars,
self.datetime_config.clone(),
Arc::clone(&self.tools),
Arc::clone(&self.mcp),
Arc::clone(&approval),
Arc::clone(&clarification),
Arc::clone(&event_bus),
Arc::clone(&self.memory_manager),
Arc::clone(&self.image_generator_manager),
compactor,
Arc::clone(&self.run_context_manager),
// known_tools is registry data → discovery writes to the registry pool.
Arc::new(ToolDiscovery::new(Arc::clone(&self.registry_pool))),
));
let chat_hub = ChatHub::new(
Arc::clone(&pool),
Arc::clone(&manager),
Arc::clone(&approval),
global_tx.clone(),
self.shutdown_token.clone(),
);
chat_hub.register("web").await;
chat_hub.register("talk").await;
let cron = TaskManager::new(Arc::clone(&pool), self.cron_tz, Arc::clone(&self.system_bus));
cron.set_session(Arc::clone(&manager));
cron.set_hub(Arc::clone(&chat_hub));
cron.set_self_arc(Arc::clone(&cron));
chat_hub.set_task_mgr(Arc::clone(&cron));
// Per-user cron loop. `start()` observes the shutdown token, so it stops on
// shutdown; adopting it lets the supervisor also join it. The name is leaked
// to satisfy the `&'static str` label — bounded by the (small) user count.
let name: &'static str = Box::leak(format!("cron:{user_id}").into_boxed_str());
self.supervisor.adopt(name, Arc::clone(&cron).start(self.shutdown_token.clone()));
Ok(Arc::new(UserContext {
user_id: user_id.to_string(),
pool,
event_bus,
sessions: manager,
chat_hub,
cron,
approval,
clarification,
elicitation,
inbox,
global_tx,
}))
}
}
/// The live per-user contexts, keyed by user id, plus the factory that builds them.
/// A `tokio::Mutex` serialises the build so a context (and its cron loop) is created
/// at most once per user, even under concurrent first-use.
pub(super) struct UserContextRegistry {
factory: UserContextFactory,
contexts: Mutex<HashMap<String, Arc<UserContext>>>,
}
impl UserContextRegistry {
pub(super) fn new(factory: UserContextFactory) -> Self {
Self { factory, contexts: Mutex::new(HashMap::new()) }
}
/// Returns the user's context, building it from `pool` on first use. Idempotent:
/// once built, the same `Arc<UserContext>` is returned until restart.
pub(super) async fn resolve(&self, user_id: &str, pool: SqlitePool) -> Result<Arc<UserContext>> {
let mut guard = self.contexts.lock().await;
if let Some(ctx) = guard.get(user_id) {
return Ok(Arc::clone(ctx));
}
let ctx = self.factory.build(user_id, pool).await?;
guard.insert(user_id.to_string(), Arc::clone(&ctx));
Ok(ctx)
}
}
+17 -9
View File
@@ -4,7 +4,7 @@ use anyhow::Result;
use serde_json::{Value, json};
use crate::cron::TaskManager;
use crate::tools::{Tool, ToolDescriptionLength};
use crate::tools::{SimpleExecution, Tool, ToolContext, ToolDescriptionLength, ToolExecution, ToolResult};
// ── execute_task ──────────────────────────────────────────────────────────────
//
@@ -134,7 +134,11 @@ pub fn build_execute_task_interface_tool(
// ── delete_cron_job ───────────────────────────────────────────────────────────
pub struct DeleteCronJob(pub Arc<TaskManager>);
/// Deleting a job is a plain DELETE on the owner's `scheduled_jobs`, so this tool
/// acts on `ToolContext::pool` (the caller's own database) rather than capturing a
/// globally-scoped `TaskManager` at registration — a job created in a user's own
/// space is deleted from that same space.
pub struct DeleteCronJob;
impl Tool for DeleteCronJob {
fn name(&self) -> &str { "delete_cron_job" }
@@ -159,12 +163,16 @@ impl Tool for DeleteCronJob {
format!("delete cron job #{id}")
}
fn execute(&self, args: Value) -> Result<String> {
let id = args["id"].as_i64().ok_or_else(|| anyhow::anyhow!("id must be an integer"))?;
if self.0.delete_job(id)? {
Ok(format!("Task {id} deleted."))
} else {
Ok(format!("No task with id {id}."))
}
fn run_with<'a>(&'a self, ctx: &ToolContext, args: Value) -> Box<dyn ToolExecution + 'a> {
let pool = Arc::clone(&ctx.pool);
Box::new(SimpleExecution::new(Box::pin(async move {
let id = args["id"].as_i64().ok_or_else(|| anyhow::anyhow!("id must be an integer"))?;
let msg = if crate::db::scheduled_jobs::delete(&pool, id).await? {
format!("Task {id} deleted.")
} else {
format!("No task with id {id}.")
};
Ok(ToolResult::Text(msg))
})))
}
}
+7 -3
View File
@@ -55,7 +55,7 @@ use anyhow::Result;
use serde_json::Value;
pub use core_api::tool::{
drive_execution, ExecutionOutcome, SimpleExecution, Tool, ToolCategory,
drive_execution, ExecutionOutcome, SimpleExecution, Tool, ToolCategory, ToolContext,
ToolDescriptionLength, ToolExecution, ToolResult, truncate_label,
};
@@ -207,8 +207,12 @@ impl ToolRegistry {
/// Start a [`ToolExecution`] for a registered tool, or `None` if `name` is not
/// in the registry (MCP / interface tools are handled by the caller). The
/// returned handle borrows the registry, which outlives the turn.
pub fn run(&self, name: &str, args: Value) -> Option<Box<dyn ToolExecution + '_>> {
self.tools.get(name).map(|tool| tool.run(args))
///
/// `ctx` carries the caller's session id and owner pool: owner-bound tools
/// (e.g. cron management) act on `ctx.pool` instead of a globally-captured
/// manager. Context-free tools ignore it via the default `run_with`.
pub fn run(&self, name: &str, ctx: &ToolContext, args: Value) -> Option<Box<dyn ToolExecution + '_>> {
self.tools.get(name).map(|tool| tool.run_with(ctx, args))
}
}