Release 0.2.0 #4
@@ -28,14 +28,14 @@ Three global broadcast buses — **never add a fourth without checking these fir
|
||||
| Bus | Cap | Events | File |
|
||||
|-----|-----|--------|------|
|
||||
| `ChatEventBus` | 256 | user message, assistant response, compaction done | `core-api/src/bus.rs` |
|
||||
| `SystemEventBus` | 64 | provider (un)registered, config key updated, job completed, session cancelled, **user created/deleted/active-changed/mounts-changed** | `core-api/src/system_bus.rs` |
|
||||
| `SystemEventBus` | 64 | provider (un)registered, config key updated, job completed, session cancelled, **user created/deleted/active-changed/mounts-changed**, **global connectors changed, connector reinstalled** | `core-api/src/system_bus.rs` |
|
||||
| `GlobalEvent` (per-user) | 512 | all `ServerEvent` variants → WS clients + inbox lifecycle | `core-api/src/events.rs` |
|
||||
|
||||
Plus internal `mpsc` queues: per-source `SourceInbox` (message serialization) and a central `notify` queue (background agents → user).
|
||||
|
||||
**The user-lifecycle reconciler** is the worked example of the rule. Creating a user, deleting one, deactivating one, or changing a shared-folder/project membership all need Docker work (provision, tear down, stop, recreate with new bind mounts). None of the endpoints that make those changes touches `ContainerManager`: each announces `SystemEvent::User{Created,Deleted,ActiveChanged,MountsChanged}` **after** its DB write, and one subscriber — `skald::wiring::spawn_user_lifecycle`, spawned post-construction because it reacts through `Skald`'s own accessors, holding only a `Weak` — does the reacting, sequentially and best-effort. The payoff is that a *future* endpoint granting membership cannot forget to remount, because remounting was never its job. Reactions never block the HTTP response, and a failure settles at the user's next login or at boot reconciliation.
|
||||
**The user-lifecycle reconciler** is the worked example of the rule. Creating a user, deleting one, deactivating one, or changing a shared-folder/project membership all need Docker work (provision, tear down, stop, recreate with new bind mounts); enabling or reinstalling a connector needs live runtimes re-snapshotted. None of the endpoints that make those changes touches `ContainerManager` or the refresh helpers: each announces `SystemEvent::User{Created,Deleted,ActiveChanged,MountsChanged}` / `McpGlobalServersChanged` / `ConnectorReinstalled` **after** its DB write, and one subscriber — `skald::wiring::spawn_user_lifecycle`, spawned post-construction because it reacts through `Skald`'s own accessors, holding only a `Weak` — does the reacting, sequentially and best-effort. Being off the response path matters for `ConnectorReinstalled` in particular: it re-copies files and restarts servers inside every live user's container, seconds of work the admin's install no longer waits on. The payoff is that a *future* endpoint granting membership cannot forget to remount, because remounting was never its job. Reactions never block the HTTP response, and a failure settles at the user's next login or at boot reconciliation.
|
||||
|
||||
**Where the bus stops: reconciliation rides it, authorization does not.** `SystemEventBus` is a lossy 64-slot broadcast whose contract is *"best-effort, settles at the next login"* — right for a stale mount, wrong for a revocation, where "settles later" *is* the failure. So deactivating or deleting a user splits in two: `Skald::revoke_user_runtime` runs **synchronously in the handler, before it responds** (revoke every session → evict + cancel the `UserContext` → `UserManager::lock`, in that order, so nothing is left querying a pool we then close and the DEK leaves RAM per §9), while only the container half — stop or remove — rides the bus. Before this, `active = 0` blocked the *next* login but left live sessions working: `login` checks the flag, `require_auth` only maps token → id. **Never put an access revocation on a bus.**
|
||||
**Where the bus stops: reconciliation rides it, authorization does not.** `SystemEventBus` is a lossy 64-slot broadcast whose contract is *"best-effort, settles at the next login"* — right for a stale mount, wrong for a revocation, where "settles later" *is* the failure. So deactivating or deleting a user splits in two: `Skald::revoke_user_runtime` runs **synchronously in the handler, before it responds** (revoke every session → evict + cancel the `UserContext` → `UserManager::lock`, in that order, so nothing is left querying a pool we then close and the DEK leaves RAM per §9), while only the container half — stop or remove — rides the bus. Before this, `active = 0` blocked the *next* login but left live sessions working: `login` checks the flag, `require_auth` only maps token → id. Same split for security groups (see the picker section) and for connectors, where the test is worth internalising because the call is literally the same function: `Skald::refresh_global_mcp_access` is **announced** (`McpGlobalServersChanged`) when a global connector is enabled or deleted — the first only makes something *appear*, the second is already enforced by `stop_server` — but **called directly** from `global_set_access` and `user_connectors_set`, where `set_access`/`set_for_user` *replace* a grant set and the refresh is what actually revokes. Both sync call-sites carry a `DELIBERATELY SYNCHRONOUS` comment, because they look identical to the announced ones. **Never put an access revocation on a bus.**
|
||||
|
||||
**Before you add a direct function call or a new import between two components, stop and ask:** is one component producing data another needs? If yes, add a variant to an existing bus and spawn a subscriber. Don't call `some_manager.log_thing(...)` from the producer — emit a `ThingHappened` event on `SystemEventBus` and let the manager subscribe.
|
||||
|
||||
|
||||
@@ -87,6 +87,23 @@ pub enum SystemEvent {
|
||||
UserMountsChanged {
|
||||
user_id: String,
|
||||
},
|
||||
|
||||
// ── Connectors (blueprint §7) ─────────────────────────────────────────────
|
||||
/// The set of **global** MCP connectors changed — one was enabled (and started)
|
||||
/// or deleted (and stopped). Every live user re-snapshots their access filter so
|
||||
/// the connector appears in / disappears from `MCP_LIST` without a re-login.
|
||||
///
|
||||
/// Emitted only for changes to the *server set*. Changing **who may use** a
|
||||
/// connector is a grant/revoke and stays synchronous in its handler, for the same
|
||||
/// reason as [`Self::UserActiveChanged`]: this bus promises "eventually", which is
|
||||
/// the wrong promise for taking access away.
|
||||
McpGlobalServersChanged,
|
||||
/// A marketplace connector was (re)installed. Anything already running it — the
|
||||
/// global runtime, each live user's per-user runtime — re-reads its metadata and
|
||||
/// re-copies its files/deps, so the new version lands without a re-login.
|
||||
ConnectorReinstalled {
|
||||
catalog_name: String,
|
||||
},
|
||||
}
|
||||
|
||||
// ── Bus ───────────────────────────────────────────────────────────────────────
|
||||
|
||||
@@ -154,6 +154,15 @@ pub(super) fn spawn_user_lifecycle(skald: &Arc<super::Skald>) {
|
||||
"user-lifecycle: remount failed (settles at next login/boot)");
|
||||
}
|
||||
}
|
||||
// Both are pure appearance/metadata refreshes across live users —
|
||||
// they widen or re-sync what is visible, never narrow it, which is
|
||||
// what makes them safe to hand to a best-effort bus.
|
||||
SystemEvent::McpGlobalServersChanged => {
|
||||
skald.refresh_global_mcp_access().await;
|
||||
}
|
||||
SystemEvent::ConnectorReinstalled { catalog_name } => {
|
||||
skald.refresh_connector_after_reinstall(&catalog_name).await;
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -32,6 +32,7 @@ use axum::extract::{Extension, Path, Query, State};
|
||||
use axum::http::header;
|
||||
use axum::response::{IntoResponse, Response};
|
||||
use axum::Json;
|
||||
use core_api::system_bus::SystemEvent;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::{json, Value};
|
||||
use sha2::{Digest, Sha256};
|
||||
@@ -794,13 +795,18 @@ pub async fn install(
|
||||
)
|
||||
.await?;
|
||||
|
||||
// Push the (re)installed metadata + code into anything already running it, so a
|
||||
// reinstall lands live instead of waiting for each user's next login: enabled
|
||||
// global servers re-snapshot the description and restart; each live user who
|
||||
// activated it gets its files/deps reconciled and the connector restarted with
|
||||
// the fresh `llm_short_description`. A first-time install matches nothing live
|
||||
// and is a cheap no-op.
|
||||
skald.refresh_connector_after_reinstall(&body.id).await;
|
||||
// Announce the (re)install so anything already running it catches up without
|
||||
// waiting for each user's next login: enabled global servers re-snapshot the
|
||||
// description and restart; each live user who activated it gets its files/deps
|
||||
// reconciled and the connector restarted with the fresh `llm_short_description`.
|
||||
// A first-time install matches nothing live and is a cheap no-op.
|
||||
//
|
||||
// On the bus, not awaited: the reaction re-copies files and restarts servers
|
||||
// inside containers, which can take seconds per live user — the admin's install
|
||||
// should not block on it, and nothing in the response below depends on it.
|
||||
skald.system_bus().send(SystemEvent::ConnectorReinstalled {
|
||||
catalog_name: body.id.clone(),
|
||||
});
|
||||
|
||||
Ok(Json(json!({
|
||||
"id": id,
|
||||
|
||||
+17
-8
@@ -14,6 +14,7 @@ use std::sync::Arc;
|
||||
|
||||
use axum::extract::{Extension, Path, State};
|
||||
use axum::Json;
|
||||
use core_api::system_bus::SystemEvent;
|
||||
use serde::Deserialize;
|
||||
use serde_json::{json, Value};
|
||||
|
||||
@@ -500,9 +501,10 @@ pub async fn global_enable(
|
||||
.ok_or_else(|| ApiError::bad_request("global server vanished after upsert"))?;
|
||||
let spec = skald_core::mcp::global_row_spec(&row);
|
||||
let started = skald.mcp().start_server(spec).await;
|
||||
// Now that the server runs in the global runtime, refresh every live user's
|
||||
// access snapshot so the connector shows up in `MCP_LIST` without a restart.
|
||||
skald.refresh_global_mcp_access().await;
|
||||
// The server now runs in the global runtime; announce it so every live user's
|
||||
// access snapshot picks it up and the connector shows in `MCP_LIST` without a
|
||||
// restart. Safe on the bus: this only makes something *appear*.
|
||||
skald.system_bus().send(SystemEvent::McpGlobalServersChanged);
|
||||
match started {
|
||||
Ok(tools) => Ok(Json(json!({ "id": id, "tools": tools, "verify": verify }))),
|
||||
Err(e) => Ok(Json(json!({ "id": id, "error": e.to_string(), "verify": verify }))),
|
||||
@@ -519,8 +521,10 @@ pub async fn global_delete(
|
||||
skald.mcp().stop_server(&row.name);
|
||||
}
|
||||
mcp_global_servers::delete(skald.db(), id).await?;
|
||||
// Drop the now-deleted server from every live user's access snapshot in place.
|
||||
skald.refresh_global_mcp_access().await;
|
||||
// Enforcement already happened above: `stop_server` killed the runtime, so the
|
||||
// tools are gone whether or not anyone re-snapshots. The announcement below only
|
||||
// tidies each live user's access filter — hence the bus rather than a direct call.
|
||||
skald.system_bus().send(SystemEvent::McpGlobalServersChanged);
|
||||
Ok(Json(json!({ "ok": true })))
|
||||
}
|
||||
|
||||
@@ -639,7 +643,11 @@ pub async fn global_set_access(
|
||||
) -> Result<Json<Value>, ApiError> {
|
||||
require_cap(&skald, &auth.user_id, role_capabilities::MANAGE_CATALOG).await?;
|
||||
mcp_global_access::set_access(skald.db(), id, &body.user_ids).await?;
|
||||
// Reflect the new grant set in every live user's access snapshot at once.
|
||||
// DELIBERATELY SYNCHRONOUS — do not "clean this up" onto `McpGlobalServersChanged`.
|
||||
// `set_access` *replaces* the grant set, so anyone dropped from `user_ids` is
|
||||
// being revoked, and the snapshot refresh is what enforces it. A best-effort
|
||||
// broadcast would leave a revoked user holding the connector until their next
|
||||
// login. Announcing a server appearing is reconciliation; taking one away is not.
|
||||
skald.refresh_global_mcp_access().await;
|
||||
Ok(Json(json!({ "ok": true })))
|
||||
}
|
||||
@@ -747,8 +755,9 @@ pub async fn user_connectors_set(
|
||||
}
|
||||
}
|
||||
|
||||
// Refresh live access snapshots so the target user's global grants take effect
|
||||
// without a restart (the catalog side is handled by the live-revoke block above).
|
||||
// DELIBERATELY SYNCHRONOUS — same reason as `global_set_access`: `set_for_user`
|
||||
// replaces the grant set, so this call is what revokes the globals the admin just
|
||||
// removed. The catalog side is already enforced by the live-revoke block above.
|
||||
skald.refresh_global_mcp_access().await;
|
||||
Ok(Json(json!({ "ok": true })))
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user