diff --git a/docs/content/docs/(configuration)/config.mdx b/docs/content/docs/(configuration)/config.mdx index b8205ff92..7314cead3 100644 --- a/docs/content/docs/(configuration)/config.mdx +++ b/docs/content/docs/(configuration)/config.mdx @@ -654,6 +654,7 @@ Channel/worker temporal context timezone precedence is: | `token` | string | None | Bot token (or `env:VAR_NAME`) | | `instances` | table[] | [] | Optional named Discord bot instances | | `dm_allowed_users` | string[] | [] | User IDs allowed to DM the bot | +| `authority` | string[] | [] | User IDs allowed to run authority-gated slash commands (`/quiet`, `/active`, `/mention-only`). Empty = open | ### `[[messaging.discord.instances]]` @@ -663,6 +664,7 @@ Channel/worker temporal context timezone precedence is: | `enabled` | bool | true | Enable this named instance | | `token` | string | **required** | Bot token (or `env:VAR_NAME`) | | `dm_allowed_users` | string[] | [] | User IDs allowed to DM this instance | +| `authority` | string[] | [] | Authority list for this instance's slash commands. Empty = open | | `allow_bot_messages` | bool | false | Whether this instance accepts bot-authored messages | ### `[messaging.slack]` @@ -674,6 +676,7 @@ Channel/worker temporal context timezone precedence is: | `app_token` | string | None | App-level token (or `env:VAR_NAME`) | | `instances` | table[] | [] | Optional named Slack app instances | | `dm_allowed_users` | string[] | [] | Slack user IDs allowed to DM the bot | +| `authority` | string[] | [] | User IDs allowed to run authority-gated slash commands. Empty = open | ### `[[messaging.slack.instances]]` @@ -684,6 +687,7 @@ Channel/worker temporal context timezone precedence is: | `bot_token` | string | **required** | Bot token (or `env:VAR_NAME`) | | `app_token` | string | **required** | App-level token (or `env:VAR_NAME`) | | `dm_allowed_users` | string[] | [] | Slack user IDs allowed to DM this instance | +| `authority` | string[] | [] | Authority list for this instance's slash commands. Empty = open | ### `[messaging.telegram]` @@ -693,6 +697,7 @@ Channel/worker temporal context timezone precedence is: | `token` | string | None | Bot token from @BotFather (or `env:VAR_NAME`). Falls back to `TELEGRAM_BOT_TOKEN` env var | | `instances` | table[] | [] | Optional named Telegram bot instances | | `dm_allowed_users` | string[] | [] | User IDs allowed to DM the bot. Empty = DMs from anyone accepted | +| `authority` | string[] | [] | User IDs allowed to run authority-gated slash commands. Empty = open | ### `[[messaging.telegram.instances]]` @@ -702,6 +707,7 @@ Channel/worker temporal context timezone precedence is: | `enabled` | bool | true | Enable this named instance | | `token` | string | **required** | Bot token (or `env:VAR_NAME`) | | `dm_allowed_users` | string[] | [] | User IDs allowed to DM this instance | +| `authority` | string[] | [] | Authority list for this instance's slash commands. Empty = open | ### `[messaging.twitch]` @@ -713,6 +719,7 @@ Channel/worker temporal context timezone precedence is: | `instances` | table[] | [] | Optional named Twitch bot instances | | `channels` | string[] | [] | Channels to join | | `trigger_prefix` | string | None | Optional prefix required to trigger replies | +| `authority` | string[] | [] | User IDs or logins (matched case-insensitively) allowed to run authority-gated slash commands. Empty = open | ### `[[messaging.twitch.instances]]` @@ -724,6 +731,7 @@ Channel/worker temporal context timezone precedence is: | `oauth_token` | string | **required** | OAuth token (`oauth:...` or plain token) | | `channels` | string[] | [] | Channels to join for this instance | | `trigger_prefix` | string | None | Optional prefix required to trigger replies | +| `authority` | string[] | [] | Authority list for this instance's slash commands. Empty = open | ### `[messaging.email]` @@ -745,6 +753,7 @@ Channel/worker temporal context timezone precedence is: | `poll_interval_secs` | integer | 30 | How often to check for new email | | `folders` | string[] | `["INBOX"]` | IMAP folders to poll | | `allowed_senders` | string[] | `[]` | Optional allowlist for inbound senders (empty = all) | +| `authority` | string[] | [] | Sender addresses allowed to run authority-gated slash commands. Empty = open | | `max_body_bytes` | integer | 262144 | Max inbound body bytes before truncation | | `max_attachment_bytes` | integer | 10485760 | Max attachment bytes to process metadata for | @@ -769,9 +778,64 @@ Channel/worker temporal context timezone precedence is: | `poll_interval_secs` | integer | 30 | How often to check for new email | | `folders` | string[] | `["INBOX"]` | IMAP folders to poll | | `allowed_senders` | string[] | `[]` | Optional allowlist (empty = all) | +| `authority` | string[] | [] | Authority list for this instance's slash commands. Empty = open | | `max_body_bytes` | integer | 262144 | Max inbound body bytes | | `max_attachment_bytes` | integer | 10485760 | Max attachment bytes | +### `[messaging.signal]` + +| Key | Type | Default | Description | +|-----|------|---------|-------------| +| `enabled` | bool | false | Enable Signal adapter (via signal-cli HTTP endpoint) | +| `http_url` | string | None | signal-cli JSON-RPC HTTP endpoint URL | +| `account` | string | None | Signal account (phone number) to receive as | +| `instances` | table[] | [] | Optional named Signal instances | +| `dm_allowed_users` | string[] | [] | Numbers allowed to DM the bot | +| `authority` | string[] | [] | Numbers allowed to run authority-gated slash commands. Empty = open | +| `group_ids` | string[] | [] | Group IDs to listen in | +| `group_allowed_users` | string[] | [] | Numbers allowed to trigger the bot in groups | +| `ignore_stories` | bool | true | Ignore story messages | + +### `[[messaging.signal.instances]]` + +| Key | Type | Default | Description | +|-----|------|---------|-------------| +| `name` | string | **required** | Instance selector used by bindings (`adapter = "name"`) | +| `enabled` | bool | true | Enable this named instance | +| `http_url` | string | **required** | signal-cli JSON-RPC HTTP endpoint URL | +| `account` | string | **required** | Signal account (phone number) to receive as | +| `dm_allowed_users` | string[] | [] | Numbers allowed to DM this instance | +| `authority` | string[] | [] | Authority list for this instance's slash commands. Empty = open | +| `group_ids` | string[] | [] | Group IDs to listen in | +| `group_allowed_users` | string[] | [] | Numbers allowed to trigger the bot in groups | +| `ignore_stories` | bool | true | Ignore story messages | + +### `[messaging.mattermost]` + +| Key | Type | Default | Description | +|-----|------|---------|-------------| +| `enabled` | bool | false | Enable Mattermost adapter | +| `base_url` | string | None | Mattermost server base URL | +| `token` | string | None | Bot token (or `env:VAR_NAME`) | +| `team_id` | string | None | Team ID to scope the bot to | +| `instances` | table[] | [] | Optional named Mattermost instances | +| `dm_allowed_users` | string[] | [] | User IDs allowed to DM the bot | +| `authority` | string[] | [] | User IDs allowed to run authority-gated slash commands. Empty = open | +| `max_attachment_bytes` | integer | 10485760 | Max attachment bytes to process | + +### `[[messaging.mattermost.instances]]` + +| Key | Type | Default | Description | +|-----|------|---------|-------------| +| `name` | string | **required** | Instance selector used by bindings (`adapter = "name"`) | +| `enabled` | bool | true | Enable this named instance | +| `base_url` | string | **required** | Mattermost server base URL | +| `token` | string | **required** | Bot token (or `env:VAR_NAME`) | +| `team_id` | string | None | Team ID to scope this instance to | +| `dm_allowed_users` | string[] | [] | User IDs allowed to DM this instance | +| `authority` | string[] | [] | Authority list for this instance's slash commands. Empty = open | +| `max_attachment_bytes` | integer | 10485760 | Max attachment bytes to process | + ### `[messaging.webhook]` | Key | Type | Default | Description | @@ -787,8 +851,9 @@ Routes platform conversations to agents. Checked in order; first match wins. Unm | Key | Type | Default | Description | |-----|------|---------|-------------| | `agent_id` | string | **required** | Which agent handles matched messages | -| `channel` | string | **required** | Platform name (`discord`, `slack`, `telegram`, `twitch`, `email`, `webhook`) | +| `channel` | string | **required** | Platform name (`discord`, `slack`, `telegram`, `twitch`, `signal`, `mattermost`, `email`, `webhook`) | | `adapter` | string | None | Optional named adapter selector (e.g. `ops` => `discord:ops`) | | `guild_id` | string | None | Discord guild filter | | `chat_id` | string | None | Telegram chat filter | | `channel_ids` | string[] | [] | Discord channel ID filter (includes threads in those channels) | +| `authority` | string[] | None | User IDs allowed to run authority-gated slash commands in this scope. Omitted falls back to the adapter's `authority` default; an explicit `[]` opens commands to everyone the binding admits | diff --git a/docs/content/docs/(messaging)/discord-setup.mdx b/docs/content/docs/(messaging)/discord-setup.mdx index f817b8350..6ef2ac462 100644 --- a/docs/content/docs/(messaging)/discord-setup.mdx +++ b/docs/content/docs/(messaging)/discord-setup.mdx @@ -31,7 +31,7 @@ In your application settings, go to **Bot**: In your application settings, go to **OAuth2** → **URL Generator**: -- Scopes: `bot` +- Scopes: `bot`, `applications.commands` (required for native slash commands) - Bot Permissions: - Send Messages - Send Messages in Threads diff --git a/docs/content/docs/(messaging)/slack-setup.mdx b/docs/content/docs/(messaging)/slack-setup.mdx index dcae2c6bf..8113b5181 100644 --- a/docs/content/docs/(messaging)/slack-setup.mdx +++ b/docs/content/docs/(messaging)/slack-setup.mdx @@ -59,6 +59,20 @@ In your Slack app settings, go to **Event Subscriptions** and make sure it's tog These tell Slack which message events to send to Spacebot. Without them, the bot won't see messages in the corresponding channel types. +## Step 4b: Add the Slash Command (optional) + +Spacebot exposes its whole command set through one umbrella command: `/spacebot status`, `/spacebot quiet`, `/spacebot help`, and so on. Typing `/spacebot` alone (or an unknown subcommand) replies with the available subcommands. + +In your Slack app settings, go to **Slash Commands** → **Create New Command**: + +- Command: `/spacebot` +- Short description: `Spacebot commands` +- Usage hint: `[command] [args]` + +Socket Mode carries the command over the existing WebSocket, so no Request URL is needed. Slash commands are declared per app in the Slack UI — repeat this step for each app instance. + +Typed commands (`/status` as a plain message) work in every channel without this step; the native command only adds Slack's autocomplete UI. + ## Step 5: Install to Workspace In your Slack app settings, go to **Install App** → **Install to Workspace** → Authorize. diff --git a/interface/src/api/schema.d.ts b/interface/src/api/schema.d.ts index 0421dd15b..92ff82ee0 100644 --- a/interface/src/api/schema.d.ts +++ b/interface/src/api/schema.d.ts @@ -58,6 +58,28 @@ export interface paths { patch?: never; trace?: never; }; + "/agents/autonomy/ceiling": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get?: never; + /** + * Set the instance-wide autonomy ceiling. + * @description Persists to the top-level `[autonomy]` table in config.toml, then stores + * the new level into the shared ArcSwap so every agent picks it up + * immediately. Returns the resulting fleet snapshot. + */ + put: operations["update_autonomy_ceiling"]; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/agents/autonomy/fleet": { parameters: { query?: never; @@ -737,6 +759,62 @@ export interface paths { patch?: never; trace?: never; }; + "/agents/wakes": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + /** List wake definitions for an agent, virtual interval survey first. */ + get: operations["list_wakes"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; + "/agents/wakes/{wake_id}": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get?: never; + /** + * Tune a wake definition. Built-in rows accept enabled, instructions, and + * min_level but keep their name; the virtual interval survey rejects all + * writes. + */ + put: operations["update_wake"]; + post?: never; + /** Delete a user-owned wake definition. */ + delete: operations["delete_wake"]; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; + "/agents/wakes/{wake_id}/fire": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get?: never; + put?: never; + /** Manually test-fire a wake through the authenticated API. */ + post: operations["fire_wake"]; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/agents/warmup": { parameters: { query?: never; @@ -2813,11 +2891,16 @@ export interface components { */ task_number?: number | null; }; + AutonomyCeilingUpdateRequest: { + ceiling: components["schemas"]["AutonomyLevel"]; + }; AutonomyCurrentRun: { started_at: string; }; AutonomyFleetResponse: { agents: components["schemas"]["AutonomyStatusResponse"][]; + /** @description Instance-wide autonomy ceiling applied to every agent. */ + ceiling: components["schemas"]["AutonomyLevel"]; }; /** * @description How much the autonomy channel may do without a user present. @@ -2877,6 +2960,11 @@ export interface components { ] | null; agent_id: string; current_run?: null | components["schemas"]["AutonomyCurrentRun"]; + /** + * @description The agent's dial capped by the instance ceiling — what the agent + * actually runs at. + */ + effective_level: components["schemas"]["AutonomyLevel"]; /** Format: int64 */ interval_secs: number; /** @description When the most recent finished run started. */ @@ -2924,6 +3012,11 @@ export interface components { BindingResponse: { adapter?: string | null; agent_id: string; + /** + * @description Omitted key and explicit empty list are distinct: `None` defers to + * the adapter default, `[]` opens the scope. + */ + authority?: string[] | null; channel: string; channel_ids: string[]; chat_id?: string | null; @@ -3183,6 +3276,12 @@ export interface components { CreateBindingRequest: { adapter?: string | null; agent_id: string; + /** + * @description Omitted and explicit `[]` are distinct: omitted writes no TOML key + * (adapter default applies), `[]` writes an explicit empty list that + * opens the scope. + */ + authority?: string[] | null; channel: string; channel_ids?: string[]; chat_id?: string | null; @@ -4578,6 +4677,11 @@ export interface components { UpdateBindingRequest: { adapter?: string | null; agent_id: string; + /** + * @description Omitted removes the TOML key (adapter default applies); `[]` writes + * an explicit empty list that opens the scope. + */ + authority?: string[] | null; channel: string; channel_ids?: string[]; chat_id?: string | null; @@ -4761,6 +4865,35 @@ export interface components { fired: boolean; message: string; }; + WakeItem: { + builtin: boolean; + enabled: boolean; + id: string; + instructions: string; + last_fired_at?: string | null; + min_level: components["schemas"]["AutonomyLevel"]; + name: string; + /** @description "schedule", "webhook", or "event". */ + trigger_kind: string; + /** + * @description Human-readable trigger summary: the cron expression, "every 15m", + * the event name, or "webhook". + */ + trigger_label: string; + /** @description Derived from live config rather than a `wake_defs` row; read-only. */ + virtual: boolean; + /** @description Public ingress path, webhook wakes only. */ + webhook_url?: string | null; + }; + WakeUpdateRequest: { + enabled?: boolean | null; + instructions?: string | null; + min_level?: null | components["schemas"]["AutonomyLevel"]; + name?: string | null; + }; + WakesResponse: { + wakes: components["schemas"]["WakeItem"][]; + }; WarmupSection: { eager_embedding_load: boolean; enabled: boolean; @@ -5168,6 +5301,43 @@ export interface operations { }; }; }; + update_autonomy_ceiling: { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + requestBody: { + content: { + "application/json": components["schemas"]["AutonomyCeilingUpdateRequest"]; + }; + }; + responses: { + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["AutonomyFleetResponse"]; + }; + }; + /** @description Invalid request */ + 400: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + /** @description Internal server error */ + 500: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + }; + }; autonomy_fleet: { parameters: { query?: never; @@ -6858,6 +7028,185 @@ export interface operations { }; }; }; + list_wakes: { + parameters: { + query: { + agent_id: string; + }; + header?: never; + path?: never; + cookie?: never; + }; + requestBody?: never; + responses: { + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["WakesResponse"]; + }; + }; + /** @description Agent not found */ + 404: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + /** @description Internal server error */ + 500: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + }; + }; + update_wake: { + parameters: { + query: { + agent_id: string; + }; + header?: never; + path: { + /** @description Wake definition id */ + wake_id: string; + }; + cookie?: never; + }; + requestBody: { + content: { + "application/json": components["schemas"]["WakeUpdateRequest"]; + }; + }; + responses: { + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["WakeItem"]; + }; + }; + /** @description Virtual wake, or a rename of a built-in wake */ + 400: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + /** @description Agent or wake not found */ + 404: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + /** @description Internal server error */ + 500: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + }; + }; + delete_wake: { + parameters: { + query: { + agent_id: string; + }; + header?: never; + path: { + /** @description Wake definition id */ + wake_id: string; + }; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description Wake deleted */ + 200: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + /** @description Virtual or built-in wake */ + 400: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + /** @description Agent or wake not found */ + 404: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + /** @description Config-owned wake */ + 409: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + /** @description Internal server error */ + 500: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + }; + }; + fire_wake: { + parameters: { + query: { + agent_id: string; + }; + header?: never; + path: { + /** @description Wake definition id */ + wake_id: string; + }; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description Wake event queued */ + 202: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + /** @description Virtual wake */ + 400: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + /** @description Agent or wake not found */ + 404: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + /** @description Internal server error */ + 500: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + }; + }; get_warmup_status: { parameters: { query?: { @@ -8120,6 +8469,13 @@ export interface operations { }; content?: never; }; + /** @description Internal server error */ + 500: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; /** @description Goal store not initialized */ 503: { headers: { @@ -8203,6 +8559,13 @@ export interface operations { }; content?: never; }; + /** @description Internal server error */ + 500: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; /** @description Goal store not initialized */ 503: { headers: { diff --git a/src/agent/channel.rs b/src/agent/channel.rs index 8c3356b9c..71b8c2357 100644 --- a/src/agent/channel.rs +++ b/src/agent/channel.rs @@ -110,6 +110,27 @@ fn should_process_event_for_channel(event: &ProcessEvent, channel_id: &ChannelId event_is_for_channel(event, channel_id) } +/// Whether an inbound message is a recognized Control command. Control +/// dispatch never runs a turn, so it skips the coalesce flush instead of +/// waiting behind a buffered batch's LLM turn. +fn is_control_command(message: &InboundMessage) -> bool { + if message.source == "system" { + return false; + } + let text = match &message.content { + crate::MessageContent::Text(text) => text.clone(), + crate::MessageContent::Command { .. } => message.content.to_string(), + _ => return false, + }; + match crate::commands::REGISTRY.parse(&text) { + crate::commands::ParseResult::Command(command) => matches!( + command.def.handler, + crate::commands::CommandHandler::Control(_) + ), + _ => false, + } +} + fn should_flush_coalesce_buffer_for_event(event: &ProcessEvent) -> bool { matches!( event, @@ -403,6 +424,13 @@ pub struct ChannelState { /// When set, the `set_outcome` tool is registered for this channel, /// allowing the LLM to explicitly store a delivery payload. pub cron_outcome: Option, + /// Whether a turn is currently in flight. Read by the inbound router's + /// busy policy without entering the channel's message queue. + pub turn_active: Arc, + /// Live response mode, encoded via `ResponseMode::to_u8`. Shared so the + /// router-side control plane can apply a mode change mid-turn; the + /// channel reads it at every gate instead of its startup snapshot. + pub response_mode: Arc, /// Autonomy run state for the `autonomy_complete` tool. Set only on /// `ChannelKind::Autonomy` channels; the run loop uses it to enforce the /// completion contract before self-exit. @@ -663,6 +691,35 @@ impl ChannelControlHandle { } } + /// Whether the channel currently has a turn in flight. Read by the + /// inbound router's busy policy. + pub fn turn_active(&self) -> bool { + self.inner + .state + .turn_active + .load(std::sync::atomic::Ordering::Acquire) + } + + /// Live response mode for this channel. + pub fn response_mode(&self) -> crate::conversation::settings::ResponseMode { + crate::conversation::settings::ResponseMode::from_u8( + self.inner + .state + .response_mode + .load(std::sync::atomic::Ordering::Acquire), + ) + } + + /// Apply a response-mode change to the running channel. Persistence is + /// the caller's job (the control plane writes the settings store); this + /// updates the live cell every gate reads. + pub fn set_response_mode_live(&self, mode: crate::conversation::settings::ResponseMode) { + self.inner + .state + .response_mode + .store(mode.to_u8(), std::sync::atomic::Ordering::Release); + } + /// Cancel all active workers and branches, emitting WorkerComplete/BranchResult /// for each so the channel can retrigger and synthesize partial results. pub async fn cancel_all_workers_and_branches(&self, reason: &str) { @@ -701,6 +758,24 @@ impl ChannelControlHandle { } } +/// RAII flag for the shared turn-active cell: set on entry to message +/// handling, cleared when the turn ends by any path (early return, error, +/// panic unwind). +struct TurnActiveGuard(Arc); + +impl TurnActiveGuard { + fn engage(flag: &Arc) -> Self { + flag.store(true, std::sync::atomic::Ordering::Release); + Self(flag.clone()) + } +} + +impl Drop for TurnActiveGuard { + fn drop(&mut self) { + self.0.store(false, std::sync::atomic::Ordering::Release); + } +} + impl WeakChannelControlHandle { pub fn dangling() -> Self { Self { inner: Weak::new() } @@ -910,6 +985,10 @@ impl Channel { model_overrides: Arc::new(resolved_settings.clone()), active_participants: Arc::new(RwLock::new(HashMap::new())), cron_outcome, + turn_active: Arc::new(std::sync::atomic::AtomicBool::new(false)), + response_mode: Arc::new(std::sync::atomic::AtomicU8::new( + resolved_settings.response_mode.to_u8(), + )), autonomy_run, }; @@ -1057,40 +1136,45 @@ impl Channel { // Update shared state for branches/workers *self.state.worker_context_settings.write().await = resolved.worker_context.clone(); self.state.model_overrides = std::sync::Arc::new(resolved.clone()); + self.state.response_mode.store( + resolved.response_mode.to_u8(), + std::sync::atomic::Ordering::Release, + ); self.resolved_settings = resolved; } /// Whether the channel is in a non-active response mode (Observe or MentionOnly). fn is_suppressed(&self) -> bool { - !matches!(self.resolved_settings.response_mode, ResponseMode::Active) + !matches!(self.response_mode(), ResponseMode::Active) + } + + /// Live response mode. Reads the shared cell rather than the startup + /// snapshot so router-side `/quiet` (and friends) apply mid-turn. + fn response_mode(&self) -> ResponseMode { + ResponseMode::from_u8( + self.state + .response_mode + .load(std::sync::atomic::Ordering::Acquire), + ) } /// Update the response mode and persist to the channel_settings table. async fn set_response_mode(&mut self, mode: ResponseMode) { self.resolved_settings.response_mode = mode; + self.state + .response_mode + .store(mode.to_u8(), std::sync::atomic::Ordering::Release); - // Persist to channel_settings table — load existing settings first so we - // don't overwrite other fields, then spawn the DB write to avoid blocking. + // Persist on a spawned task to avoid blocking the channel. The + // store's atomic field update leaves the rest of the settings row + // untouched, so this write can't restore stale fields over a + // concurrent settings writer or a router-side mode change. let pool = self.deps.sqlite_pool.clone(); let agent_id = self.deps.agent_id.clone(); let channel_id: String = self.id.as_ref().to_owned(); tokio::spawn(async move { let store = crate::conversation::ChannelSettingsStore::new(pool); - let mut settings = match store.get(&agent_id, &channel_id).await { - Ok(Some(existing)) => existing, - Ok(None) => crate::conversation::ConversationSettings::default(), - Err(error) => { - tracing::warn!( - %error, - %channel_id, - ?mode, - "failed to load existing settings before persisting response_mode" - ); - crate::conversation::ConversationSettings::default() - } - }; - settings.response_mode = mode; - if let Err(error) = store.upsert(&agent_id, &channel_id, &settings).await { + if let Err(error) = store.set_response_mode(&agent_id, &channel_id, mode).await { tracing::warn!( %error, %channel_id, @@ -1249,7 +1333,6 @@ impl Channel { ControlAction::Status => { let temporal_context = TemporalContext::from_runtime(self.deps.runtime_config.as_ref()); - let now_line = temporal_context.current_time_line(); let routing = self.deps.runtime_config.routing.load(); let channel_model = self .resolved_settings @@ -1259,46 +1342,24 @@ impl Channel { .resolved_settings .resolve_model("branch") .unwrap_or_else(|| routing.resolve(ProcessType::Branch, None)); - let mode = match self.resolved_settings.response_mode { - ResponseMode::Active => "active", - ResponseMode::Observe => "observe (learning, never responds)", - ResponseMode::MentionOnly => "mention-only (@mention/reply only)", - }; - let adapter = self.current_adapter().unwrap_or("unknown"); - let body = format!( - "status\n\ - - agent: {}\n\ - - channel: {}\n\ - - adapter: {}\n\ - - mode: {}\n\ - - channel model: {}\n\ - - branch model: {}\n\ - - time: {}", - self.deps.agent_id, - self.id, - adapter, - mode, + let body = crate::commands::control::status_text( + &self.deps.agent_id, + &self.id, + self.current_adapter().unwrap_or("unknown"), + self.response_mode(), channel_model, branch_model, - now_line + &temporal_context.current_time_line(), ); self.send_builtin_text(body, def.name).await; } ControlAction::SetResponseMode(mode) => { self.set_response_mode(mode).await; - let confirmation = match mode { - ResponseMode::Active => { - "active mode enabled. i'll respond normally in this chat." - } - ResponseMode::Observe => { - "observe mode enabled. i'll learn from this conversation but won't respond." - } - ResponseMode::MentionOnly => { - "mention-only mode enabled. i'll only respond when @mentioned or replied to." - } - }; - self.send_builtin_text(confirmation.to_string(), def.name) - .await; + self.send_builtin_text( + crate::commands::control::mode_confirmation(mode).to_string(), + def.name, + ) + .await; } ControlAction::Help => { self.send_builtin_text(crate::commands::REGISTRY.help_text(), def.name) @@ -1416,8 +1477,14 @@ impl Channel { self.coalesce_buffer.push(message); self.update_coalesce_deadline(&config).await; } else { - // Flush any pending buffer before handling this message - if let Err(error) = self.flush_coalesce_buffer().await { + // Control commands dispatch immediately without + // flushing — the buffer keeps its own debounce clock. + // Everything else (including Agent commands, which + // are joining the conversation) flushes first so + // order is preserved. + if !is_control_command(&message) + && let Err(error) = self.flush_coalesce_buffer().await + { tracing::error!(%error, channel_id = %self.id, "error flushing coalesce buffer"); } if let Err(error) = self.handle_message(message).await { @@ -1610,6 +1677,7 @@ impl Channel { #[tracing::instrument(skip(self, messages), fields(channel_id = %self.id, agent_id = %self.deps.agent_id, message_count = messages.len()))] async fn handle_message_batch(&mut self, messages: Vec) -> Result<()> { // Apply runtime-config updates immediately without requiring a restart. + let _turn_guard = TurnActiveGuard::engage(&self.state.turn_active); let message_count = messages.len(); let batch_start_timestamp = messages @@ -1815,7 +1883,7 @@ impl Channel { // Observe mode: always suppress (even with mentions in batch). // MentionOnly mode: suppress only when no invocations in the batch. let should_suppress_batch = !self.is_dm() - && match self.resolved_settings.response_mode { + && match self.response_mode() { ResponseMode::Active => false, ResponseMode::Observe => true, ResponseMode::MentionOnly => !batch_has_invoke, @@ -1825,7 +1893,7 @@ impl Channel { tracing::debug!( channel_id = %self.id, message_count, - response_mode = ?self.resolved_settings.response_mode, + response_mode = ?self.response_mode(), "suppressing unsolicited coalesced batch" ); // Inject batch messages into in-memory history so the agent @@ -2066,6 +2134,7 @@ impl Channel { #[tracing::instrument(skip(self, message), fields(channel_id = %self.id, agent_id = %self.deps.agent_id, message_id = %message.id))] async fn handle_message(&mut self, message: InboundMessage) -> Result<()> { // Apply runtime-config updates immediately without requiring a restart. + let _turn_guard = TurnActiveGuard::engage(&self.state.turn_active); // Track the inbound message that triggered this turn so outbound // responses carry the correct routing metadata (e.g. Slack thread_ts). @@ -2197,7 +2266,7 @@ impl Channel { // Deterministic ping ack for Discord mention-only mentions/replies to avoid // flaky model behavior (e.g. skipping or over-formatting simple liveness checks). // Skipped in Observe mode — the agent never responds in Observe. - if !matches!(self.resolved_settings.response_mode, ResponseMode::Observe) + if !matches!(self.response_mode(), ResponseMode::Observe) && should_send_discord_quiet_mode_ping_ack(&message, &raw_text, self.is_suppressed()) { self.send_builtin_text("yeah i'm here".to_string(), "discord-ping") @@ -2261,25 +2330,24 @@ impl Channel { // Response mode guardrail: // Observe mode: always suppress — agent learns but never responds. // MentionOnly mode: suppress unless explicitly invoked. - if !matches!(self.resolved_settings.response_mode, ResponseMode::Active) + if !matches!(self.response_mode(), ResponseMode::Active) && message.source != "system" && !self.is_dm() { // Observe mode always suppresses; MentionOnly checks for invocation. - let should_suppress = - if matches!(self.resolved_settings.response_mode, ResponseMode::Observe) { - true - } else { - (invoked_by_command, invoked_by_mention, invoked_by_reply) = - self.compute_listen_mode_invocation(&message, &raw_text); - !invoked_by_command && !invoked_by_mention && !invoked_by_reply - }; + let should_suppress = if matches!(self.response_mode(), ResponseMode::Observe) { + true + } else { + (invoked_by_command, invoked_by_mention, invoked_by_reply) = + self.compute_listen_mode_invocation(&message, &raw_text); + !invoked_by_command && !invoked_by_mention && !invoked_by_reply + }; if should_suppress { tracing::debug!( channel_id = %self.id, source = %message.source, - response_mode = ?self.resolved_settings.response_mode, + response_mode = ?self.response_mode(), "suppressing unsolicited reply" ); // In Observe and MentionOnly modes, inject the message into diff --git a/src/agent/process_control.rs b/src/agent/process_control.rs index 587b805e2..400afab8c 100644 --- a/src/agent/process_control.rs +++ b/src/agent/process_control.rs @@ -149,6 +149,23 @@ impl ProcessControlRegistry { .collect() } + /// Live control handle for a channel, when one is running. Prunes a + /// stale registration on the way. + pub async fn channel_handle( + &self, + channel_id: &ChannelId, + ) -> Option { + match self.lookup_channel_handle(channel_id).await { + ChannelLookupResult::Found(handle) => Some(handle), + ChannelLookupResult::Stale(registration_id) => { + self.remove_stale_channel_if_matches(channel_id, registration_id) + .await; + None + } + ChannelLookupResult::Missing => None, + } + } + async fn lookup_channel_handle(&self, channel_id: &ChannelId) -> ChannelLookupResult { let handle_entry = { let channels = self.channels.read().await; diff --git a/src/api/bindings.rs b/src/api/bindings.rs index beab78509..a0140236d 100644 --- a/src/api/bindings.rs +++ b/src/api/bindings.rs @@ -18,6 +18,9 @@ pub struct BindingResponse { pub channel_ids: Vec, pub require_mention: bool, pub dm_allowed_users: Vec, + /// Omitted key and explicit empty list are distinct: `None` defers to + /// the adapter default, `[]` opens the scope. + pub authority: Option>, } #[derive(Serialize, Deserialize, utoipa::ToSchema)] @@ -51,6 +54,11 @@ pub(super) struct CreateBindingRequest { require_mention: bool, #[serde(default)] dm_allowed_users: Vec, + /// Omitted and explicit `[]` are distinct: omitted writes no TOML key + /// (adapter default applies), `[]` writes an explicit empty list that + /// opens the scope. + #[serde(default)] + authority: Option>, /// Optional: set platform credentials if not yet configured. #[serde(default)] platform_credentials: Option, @@ -161,6 +169,10 @@ pub(super) struct UpdateBindingRequest { require_mention: bool, #[serde(default)] dm_allowed_users: Vec, + /// Omitted removes the TOML key (adapter default applies); `[]` writes + /// an explicit empty list that opens the scope. + #[serde(default)] + authority: Option>, } #[derive(Serialize, Deserialize, utoipa::ToSchema)] @@ -210,6 +222,7 @@ pub(super) async fn list_bindings( channel_ids: b.channel_ids, require_mention: b.require_mention, dm_allowed_users: b.dm_allowed_users, + authority: b.authority, }) .collect(); @@ -491,6 +504,13 @@ pub(super) async fn create_binding( } binding_table["dm_allowed_users"] = toml_edit::value(arr); } + if let Some(authority) = &request.authority { + let mut arr = toml_edit::Array::new(); + for id in authority { + arr.push(id.as_str()); + } + binding_table["authority"] = toml_edit::value(arr); + } bindings_array.push(binding_table); tokio::fs::write(&config_path, doc.to_string()) @@ -593,18 +613,11 @@ pub(super) async fn create_binding( } } }; - let slack_commands = new_config - .messaging - .slack - .as_ref() - .map(|s| s.commands.clone()) - .unwrap_or_default(); match crate::messaging::slack::SlackAdapter::new( "slack", &bot_token, &app_token, slack_perms, - slack_commands, ) { Ok(adapter) => { if let Err(error) = manager.register_and_start(adapter).await { @@ -877,6 +890,16 @@ pub(super) async fn update_binding( binding.remove("dm_allowed_users"); } + if let Some(authority) = &request.authority { + let mut arr = toml_edit::Array::new(); + for id in authority { + arr.push(id.as_str()); + } + binding["authority"] = toml_edit::value(arr); + } else { + binding.remove("authority"); + } + tokio::fs::write(&config_path, doc.to_string()) .await .map_err(|error| { diff --git a/src/api/messaging.rs b/src/api/messaging.rs index 1d258a277..3769643c4 100644 --- a/src/api/messaging.rs +++ b/src/api/messaging.rs @@ -1277,7 +1277,6 @@ pub(super) async fn toggle_platform( &slack_config.bot_token, &slack_config.app_token, perms, - slack_config.commands.clone(), ) { Ok(adapter) => { if let Err(error) = manager.register_and_start(adapter).await { @@ -1310,7 +1309,6 @@ pub(super) async fn toggle_platform( &instance.bot_token, &instance.app_token, perms, - instance.commands.clone(), ) { Ok(adapter) => { if let Err(error) = manager.register_and_start(adapter).await { diff --git a/src/cli/binding.rs b/src/cli/binding.rs index ea3b38d7b..2ac9b7ddd 100644 --- a/src/cli/binding.rs +++ b/src/cli/binding.rs @@ -45,6 +45,9 @@ pub enum BindingCommand { /// Allow DMs from a user ID (repeatable) #[arg(long = "dm-user")] dm_allowed_users: Vec, + /// User ID allowed to run authority-gated slash commands (repeatable) + #[arg(long = "authority")] + authority: Vec, }, /// Update a binding. The binding is matched by the positional args and /// original selector flags; the updated binding is written exactly as the @@ -99,6 +102,9 @@ pub enum BindingCommand { /// Allowed DM user ID for the updated binding (repeatable, omit to clear) #[arg(long = "dm-user")] dm_allowed_users: Vec, + /// Authority user ID for the updated binding (repeatable, omit to clear) + #[arg(long = "authority")] + authority: Vec, }, /// Delete a binding matched by agent, channel, and selector flags Delete { @@ -187,6 +193,7 @@ pub async fn run(ctx: &super::Context, binding_cmd: BindingCommand) -> anyhow::R channel_ids, require_mention, dm_allowed_users, + authority, } => { let mut body = serde_json::json!({ "agent_id": agent_id, @@ -194,6 +201,7 @@ pub async fn run(ctx: &super::Context, binding_cmd: BindingCommand) -> anyhow::R "channel_ids": channel_ids, "require_mention": require_mention, "dm_allowed_users": dm_allowed_users, + "authority": authority, }); set_optional(&mut body, "adapter", &adapter); set_optional(&mut body, "guild_id", &guild_id); @@ -234,6 +242,7 @@ pub async fn run(ctx: &super::Context, binding_cmd: BindingCommand) -> anyhow::R channel_ids, require_mention, dm_allowed_users, + authority, } => { let mut body = serde_json::json!({ "original_agent_id": agent_id, @@ -243,6 +252,7 @@ pub async fn run(ctx: &super::Context, binding_cmd: BindingCommand) -> anyhow::R "channel_ids": channel_ids, "require_mention": require_mention, "dm_allowed_users": dm_allowed_users, + "authority": authority, }); set_optional(&mut body, "original_adapter", &adapter); set_optional(&mut body, "original_guild_id", &guild_id); diff --git a/src/commands.rs b/src/commands.rs index 2b5e9132e..f7b0cac79 100644 --- a/src/commands.rs +++ b/src/commands.rs @@ -2,9 +2,14 @@ //! //! Design: `docs/design-docs/slash-commands.md`. +pub mod access; +pub mod control; +pub mod dispatch; +pub mod native; pub mod registry; pub use registry::{ - AgentAction, ArgSpec, COMMANDS, CommandCategory, CommandDef, CommandHandler, CommandRegistry, - ControlAction, ParseResult, ParsedCommand, REGISTRY, + AgentAction, ArgSpec, BusyPolicy, COMMANDS, CommandAccess, CommandAvailability, + CommandCategory, CommandDef, CommandHandler, CommandRegistry, ControlAction, ParseResult, + ParsedCommand, REGISTRY, Surface, }; diff --git a/src/commands/access.rs b/src/commands/access.rs new file mode 100644 index 000000000..4b29dea1a --- /dev/null +++ b/src/commands/access.rs @@ -0,0 +1,313 @@ +//! Authority resolution for slash commands. +//! +//! Access layers *authority* (who may change state) on top of admission (who +//! may talk to the agent at all); it never widens admission. The authority +//! list for a scope resolves binding-first, then the adapter-instance +//! default, and is open when neither is configured — existing configs behave +//! exactly as before. + +use super::registry::{CommandAccess, CommandDef, CommandHandler, Surface}; +use std::collections::HashMap; + +/// Adapter-level default authority lists, keyed by runtime adapter key +/// (`discord`, `telegram:support`, ...). Rebuilt whenever config reloads, +/// alongside the binding swap. +#[derive(Debug, Default, Clone)] +pub struct AdapterAuthorityDefaults { + by_adapter: HashMap>, +} + +impl AdapterAuthorityDefaults { + pub fn from_config(config: &crate::config::Config) -> Self { + let mut by_adapter = HashMap::new(); + let mut insert = |key: String, authority: &Vec| { + if !authority.is_empty() { + by_adapter.insert(key, authority.clone()); + } + }; + + let messaging = &config.messaging; + if let Some(discord) = &messaging.discord { + insert("discord".into(), &discord.authority); + for instance in &discord.instances { + insert(format!("discord:{}", instance.name), &instance.authority); + } + } + if let Some(slack) = &messaging.slack { + insert("slack".into(), &slack.authority); + for instance in &slack.instances { + insert(format!("slack:{}", instance.name), &instance.authority); + } + } + if let Some(telegram) = &messaging.telegram { + insert("telegram".into(), &telegram.authority); + for instance in &telegram.instances { + insert(format!("telegram:{}", instance.name), &instance.authority); + } + } + if let Some(twitch) = &messaging.twitch { + insert("twitch".into(), &twitch.authority); + for instance in &twitch.instances { + insert(format!("twitch:{}", instance.name), &instance.authority); + } + } + if let Some(signal) = &messaging.signal { + insert("signal".into(), &signal.authority); + for instance in &signal.instances { + insert(format!("signal:{}", instance.name), &instance.authority); + } + } + if let Some(mattermost) = &messaging.mattermost { + insert("mattermost".into(), &mattermost.authority); + for instance in &mattermost.instances { + insert(format!("mattermost:{}", instance.name), &instance.authority); + } + } + if let Some(email) = &messaging.email { + insert("email".into(), &email.authority); + for instance in &email.instances { + insert(format!("email:{}", instance.name), &instance.authority); + } + } + + Self { by_adapter } + } + + pub fn for_adapter(&self, adapter_key: &str) -> Option<&[String]> { + self.by_adapter + .get(adapter_key) + .map(|authority| authority.as_slice()) + } +} + +/// The identity and scope facts needed to answer "may this sender run this +/// command here". +pub struct AccessContext<'a> { + /// Authority list from the matched binding, if one matched and set the + /// key. An explicit empty list opens the scope; `None` (key omitted) + /// falls back to the adapter default. + pub binding_authority: Option<&'a [String]>, + /// Adapter-instance default authority for this message's adapter. + pub adapter_default: Option<&'a [String]>, + /// Platform user id (`InboundMessage.sender_id`). + pub sender_id: &'a str, + /// Twitch login, compared case-insensitively — Twitch surfaces logins to + /// users while `sender_id` carries the numeric id. + pub sender_login: Option<&'a str>, +} + +impl AccessContext<'_> { + /// Whether the sender holds authority in this scope. Open (true for + /// everyone) when no authority list is configured at either level. A + /// binding's explicit empty list opens its scope without falling back + /// to the adapter default; an omitted binding list defers to it. + pub fn is_authority(&self) -> bool { + let non_empty = |list: &&[String]| !list.is_empty(); + let list = match self.binding_authority { + Some([]) => return true, + Some(list) => list, + None => match self.adapter_default.filter(non_empty) { + Some(list) => list, + None => return true, + }, + }; + + list.iter().any(|entry| { + entry == "*" + || entry == self.sender_id + || self + .sender_login + .is_some_and(|login| entry.eq_ignore_ascii_case(login)) + }) + } + + pub fn allows(&self, def: &CommandDef) -> bool { + match def.access { + CommandAccess::Everyone => true, + CommandAccess::Authority => self.is_authority(), + } + } +} + +/// Denial reply for an Authority command the sender may not run: names the +/// commands available to them instead of a bare "no". +pub fn denial_text(def: &CommandDef, surface: Surface) -> String { + let available: Vec = super::REGISTRY + .defs() + .iter() + .filter(|candidate| { + candidate.access == CommandAccess::Everyone && candidate.availability.on(surface) + }) + .map(|candidate| format!("/{}", candidate.name)) + .collect(); + format!( + "/{} requires authority in this channel. available to you: {}", + def.name, + available.join(", ") + ) +} + +/// Reply for an Agent command with `BusyPolicy::Reject` arriving mid-turn. +pub fn busy_reject_text(def: &CommandDef) -> String { + format!( + "/{} can't run while a turn is in flight — /stop cancels the current work", + def.name + ) +} + +/// Acknowledgment for an Agent command queued behind an in-flight turn. A +/// recognized command is never silently swallowed. +pub fn busy_queued_text(def: &CommandDef) -> String { + debug_assert!(matches!(def.handler, CommandHandler::Agent(_))); + format!( + "/{} queued — it runs after the current turn finishes", + def.name + ) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::commands::REGISTRY; + + fn def(name: &str) -> &'static CommandDef { + REGISTRY.resolve(name).expect("command def") + } + + fn owned(entries: &[&str]) -> Vec { + entries.iter().map(|entry| entry.to_string()).collect() + } + + #[test] + fn absent_authority_lists_leave_commands_open() { + let context = AccessContext { + binding_authority: None, + adapter_default: None, + sender_id: "12345", + sender_login: None, + }; + assert!(context.is_authority()); + assert!(context.allows(def("quiet"))); + } + + #[test] + fn empty_adapter_default_is_treated_as_absent() { + let empty: Vec = Vec::new(); + let context = AccessContext { + binding_authority: None, + adapter_default: Some(&empty), + sender_id: "12345", + sender_login: None, + }; + assert!(context.is_authority()); + } + + #[test] + fn explicit_empty_binding_list_opens_scope_despite_restricted_adapter_default() { + let empty: Vec = Vec::new(); + let adapter = owned(&["999"]); + let context = AccessContext { + binding_authority: Some(&empty), + adapter_default: Some(&adapter), + sender_id: "12345", + sender_login: None, + }; + assert!(context.is_authority()); + assert!(context.allows(def("quiet"))); + } + + #[test] + fn omitted_binding_list_defers_to_restricted_adapter_default() { + let adapter = owned(&["999"]); + let context = AccessContext { + binding_authority: None, + adapter_default: Some(&adapter), + sender_id: "12345", + sender_login: None, + }; + assert!(!context.is_authority()); + assert!(!context.allows(def("quiet"))); + } + + #[test] + fn binding_authority_gates_authority_commands_only() { + let authority = owned(&["999"]); + let context = AccessContext { + binding_authority: Some(&authority), + adapter_default: None, + sender_id: "12345", + sender_login: None, + }; + assert!(!context.is_authority()); + assert!(!context.allows(def("quiet"))); + assert!(context.allows(def("status"))); + assert!(context.allows(def("help"))); + assert!(context.allows(def("tasks"))); + } + + #[test] + fn binding_list_takes_precedence_over_adapter_default() { + let binding = owned(&["12345"]); + let adapter = owned(&["999"]); + let context = AccessContext { + binding_authority: Some(&binding), + adapter_default: Some(&adapter), + sender_id: "12345", + sender_login: None, + }; + assert!(context.is_authority()); + + let excluded = AccessContext { + binding_authority: Some(&adapter), + adapter_default: Some(&binding), + sender_id: "12345", + sender_login: None, + }; + assert!(!excluded.is_authority()); + } + + #[test] + fn adapter_default_applies_when_binding_has_no_list() { + let adapter = owned(&["12345"]); + let context = AccessContext { + binding_authority: None, + adapter_default: Some(&adapter), + sender_id: "12345", + sender_login: None, + }; + assert!(context.is_authority()); + } + + #[test] + fn wildcard_grants_everyone_authority() { + let authority = owned(&["*"]); + let context = AccessContext { + binding_authority: Some(&authority), + adapter_default: None, + sender_id: "anyone", + sender_login: None, + }; + assert!(context.is_authority()); + } + + #[test] + fn twitch_logins_match_case_insensitively() { + let authority = owned(&["StreamerName"]); + let context = AccessContext { + binding_authority: Some(&authority), + adapter_default: None, + sender_id: "44556677", + sender_login: Some("streamername"), + }; + assert!(context.is_authority()); + } + + #[test] + fn denial_names_available_commands() { + let text = denial_text(def("quiet"), Surface::Discord); + assert!(text.starts_with("/quiet requires authority")); + assert!(text.contains("/status")); + assert!(text.contains("/help")); + assert!(!text.contains("/active")); + } +} diff --git a/src/commands/control.rs b/src/commands/control.rs new file mode 100644 index 000000000..3d369ed0e --- /dev/null +++ b/src/commands/control.rs @@ -0,0 +1,411 @@ +//! Router-side control plane for slash commands. +//! +//! Control commands execute deterministically against channel state — the +//! settings stores, process control, and runtime config — without creating +//! an inbound message or consuming an agent turn. Dispatching here, before +//! the channel's message queue, is what makes them busy-immune: a `/quiet` +//! lands even while the channel is mid-turn. +//! +//! The reply-text builders are shared with the channel's in-queue dispatch +//! path so both surfaces produce identical output. + +use super::registry::{ControlAction, Surface}; +use crate::ProcessType; +use crate::agent::channel_prompt::TemporalContext; +use crate::conversation::settings::{ResolvedConversationSettings, ResponseMode}; + +/// Confirmation reply for a response-mode change. +pub fn mode_confirmation(mode: ResponseMode) -> &'static str { + match mode { + ResponseMode::Active => "active mode enabled. i'll respond normally in this chat.", + ResponseMode::Observe => { + "observe mode enabled. i'll learn from this conversation but won't respond." + } + ResponseMode::MentionOnly => { + "mention-only mode enabled. i'll only respond when @mentioned or replied to." + } + } +} + +/// Short mode name for failure replies. +fn mode_name(mode: ResponseMode) -> &'static str { + match mode { + ResponseMode::Active => "active", + ResponseMode::Observe => "observe", + ResponseMode::MentionOnly => "mention-only", + } +} + +/// Human-readable label for a response mode in `/status` output. +pub fn mode_label(mode: ResponseMode) -> &'static str { + match mode { + ResponseMode::Active => "active", + ResponseMode::Observe => "observe (learning, never responds)", + ResponseMode::MentionOnly => "mention-only (@mention/reply only)", + } +} + +/// `/status` reply body. +#[allow(clippy::too_many_arguments)] +pub fn status_text( + agent_id: &str, + channel_id: &str, + adapter: &str, + mode: ResponseMode, + channel_model: &str, + branch_model: &str, + now_line: &str, +) -> String { + format!( + "status\n\ + - agent: {agent_id}\n\ + - channel: {channel_id}\n\ + - adapter: {adapter}\n\ + - mode: {}\n\ + - channel model: {channel_model}\n\ + - branch model: {branch_model}\n\ + - time: {now_line}", + mode_label(mode) + ) +} + +/// Everything a control command needs, owned so execution can run in a +/// spawned task without borrowing the router loop. +pub struct ControlPlane { + pub deps: crate::AgentDeps, + pub conversation_id: String, + /// Runtime adapter key of the surface the command arrived on. + pub adapter: String, + /// Settings defaults from the matched binding, if any. + pub binding_settings: Option, + pub surface: Surface, + pub is_authority: bool, + pub is_portal: bool, +} + +impl ControlPlane { + /// Execute a control action and return the reply text. + pub async fn execute(&self, action: ControlAction) -> String { + match action { + ControlAction::Status => self.status().await, + ControlAction::SetResponseMode(mode) => self.set_response_mode(mode).await, + ControlAction::Help => { + crate::commands::REGISTRY.help_text_for(Some(self.surface), self.is_authority) + } + ControlAction::AgentId => self.deps.agent_id.to_string(), + } + } + + /// Resolve the conversation's settings the same way channel creation + /// does: per-conversation DB override > binding defaults > defaults. + async fn resolved_settings(&self) -> ResolvedConversationSettings { + let db_settings = if self.is_portal { + let store = + crate::conversation::PortalConversationStore::new(self.deps.sqlite_pool.clone()); + match store.get(&self.deps.agent_id, &self.conversation_id).await { + Ok(conversation) => conversation.and_then(|conversation| conversation.settings), + Err(error) => { + tracing::warn!( + %error, + conversation_id = %self.conversation_id, + "failed to load portal conversation settings for control command" + ); + None + } + } + } else { + let store = + crate::conversation::ChannelSettingsStore::new(self.deps.sqlite_pool.clone()); + match store.get(&self.deps.agent_id, &self.conversation_id).await { + Ok(settings) => settings, + Err(error) => { + tracing::warn!( + %error, + conversation_id = %self.conversation_id, + "failed to load channel settings for control command" + ); + None + } + } + }; + ResolvedConversationSettings::resolve( + db_settings.as_ref(), + self.binding_settings.as_ref(), + None, + ) + } + + /// The live channel's control handle, when one is running for this + /// conversation. + async fn live_channel(&self) -> Option { + let channel_id: crate::ChannelId = std::sync::Arc::from(self.conversation_id.as_str()); + self.deps + .process_control_registry + .channel_handle(&channel_id) + .await + } + + async fn status(&self) -> String { + let temporal_context = TemporalContext::from_runtime(self.deps.runtime_config.as_ref()); + let routing = self.deps.runtime_config.routing.load(); + let resolved = self.resolved_settings().await; + let channel_model = resolved + .resolve_model("channel") + .unwrap_or_else(|| routing.resolve(ProcessType::Channel, None)); + let branch_model = resolved + .resolve_model("branch") + .unwrap_or_else(|| routing.resolve(ProcessType::Branch, None)); + // A live channel's in-memory mode wins over the store: mode changes + // persist asynchronously, so the cell is ahead of the DB briefly. + let mode = match self.live_channel().await { + Some(handle) => handle.response_mode(), + None => resolved.response_mode, + }; + status_text( + &self.deps.agent_id, + &self.conversation_id, + &self.adapter, + mode, + channel_model, + branch_model, + &temporal_context.current_time_line(), + ) + } + + async fn set_response_mode(&self, mode: ResponseMode) -> String { + // On persistence failure the change is abandoned — no live update, + // and the reply reports the failure instead of confirming a mode + // that never persisted. + if let Err(error) = persist_response_mode( + &self.deps.sqlite_pool, + &self.deps.agent_id, + &self.conversation_id, + self.is_portal, + mode, + ) + .await + { + tracing::warn!( + %error, + conversation_id = %self.conversation_id, + ?mode, + "failed to persist response mode" + ); + return format!( + "couldn't switch to {} mode — settings persistence failed, mode is unchanged", + mode_name(mode) + ); + } + + // Poke the live channel so the change applies mid-turn, not on the + // next channel restart. + if let Some(handle) = self.live_channel().await { + handle.set_response_mode_live(mode); + } + + mode_confirmation(mode).to_string() + } +} + +/// Persist the response mode through the stores' atomic field updates, +/// leaving every other settings field untouched. No settings read happens +/// here, so concurrent whole-row writers can't be clobbered with stale +/// fields, and an unreadable record can't be replaced with defaults. +pub(crate) async fn persist_response_mode( + pool: &sqlx::SqlitePool, + agent_id: &str, + conversation_id: &str, + is_portal: bool, + mode: ResponseMode, +) -> anyhow::Result<()> { + if is_portal { + let store = crate::conversation::PortalConversationStore::new(pool.clone()); + let updated = store + .set_response_mode(agent_id, conversation_id, mode) + .await?; + if !updated { + anyhow::bail!("portal conversation {conversation_id} not found"); + } + } else { + let store = crate::conversation::ChannelSettingsStore::new(pool.clone()); + store + .set_response_mode(agent_id, conversation_id, mode) + .await?; + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use sqlx::sqlite::SqlitePoolOptions; + + async fn memory_pool() -> sqlx::SqlitePool { + SqlitePoolOptions::new() + .max_connections(1) + .connect("sqlite::memory:") + .await + .expect("in-memory sqlite should connect") + } + + async fn create_channel_settings_table(pool: &sqlx::SqlitePool) { + sqlx::query( + "CREATE TABLE channel_settings ( + agent_id TEXT NOT NULL, + conversation_id TEXT NOT NULL, + settings TEXT NOT NULL DEFAULT '{}', + updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (agent_id, conversation_id) + )", + ) + .execute(pool) + .await + .expect("channel_settings table should create"); + } + + async fn create_portal_conversations_table(pool: &sqlx::SqlitePool) { + sqlx::query( + "CREATE TABLE portal_conversations ( + id TEXT PRIMARY KEY, + agent_id TEXT NOT NULL, + title TEXT NOT NULL, + title_source TEXT NOT NULL DEFAULT 'system', + archived INTEGER NOT NULL DEFAULT 0, + settings TEXT, + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP + )", + ) + .execute(pool) + .await + .expect("portal_conversations table should create"); + } + + #[tokio::test] + async fn channel_mode_change_preserves_other_settings() { + let pool = memory_pool().await; + create_channel_settings_table(&pool).await; + + let store = crate::conversation::ChannelSettingsStore::new(pool.clone()); + let existing = crate::conversation::ConversationSettings { + model: Some("special-model".into()), + ..Default::default() + }; + store.upsert("agent", "conv", &existing).await.unwrap(); + + persist_response_mode(&pool, "agent", "conv", false, ResponseMode::Observe) + .await + .unwrap(); + + let loaded = store.get("agent", "conv").await.unwrap().unwrap(); + assert_eq!(loaded.response_mode, ResponseMode::Observe); + assert_eq!( + loaded.model.as_deref(), + Some("special-model"), + "a mode change must not clobber other persisted settings" + ); + } + + #[tokio::test] + async fn channel_store_failure_propagates_instead_of_writing_defaults() { + // No channel_settings table: the settings read fails, and the + // failure must surface instead of turning into a default upsert + // and a false confirmation. + let pool = memory_pool().await; + let result = + persist_response_mode(&pool, "agent", "conv", false, ResponseMode::Observe).await; + assert!(result.is_err()); + } + + #[tokio::test] + async fn portal_mode_change_preserves_other_settings() { + let pool = memory_pool().await; + create_portal_conversations_table(&pool).await; + + let store = crate::conversation::PortalConversationStore::new(pool.clone()); + store.ensure("agent", "session").await.unwrap(); + let existing = crate::conversation::ConversationSettings { + model: Some("special-model".into()), + ..Default::default() + }; + store + .update("agent", "session", None, None, Some(existing)) + .await + .unwrap(); + + persist_response_mode(&pool, "agent", "session", true, ResponseMode::MentionOnly) + .await + .unwrap(); + + let loaded = store + .get("agent", "session") + .await + .unwrap() + .unwrap() + .settings + .unwrap(); + assert_eq!(loaded.response_mode, ResponseMode::MentionOnly); + assert_eq!(loaded.model.as_deref(), Some("special-model")); + } + + #[tokio::test] + async fn portal_mode_change_initializes_null_settings() { + let pool = memory_pool().await; + create_portal_conversations_table(&pool).await; + + let store = crate::conversation::PortalConversationStore::new(pool.clone()); + store.ensure("agent", "session").await.unwrap(); + + persist_response_mode(&pool, "agent", "session", true, ResponseMode::Observe) + .await + .unwrap(); + + let loaded = store + .get("agent", "session") + .await + .unwrap() + .unwrap() + .settings + .unwrap(); + assert_eq!(loaded.response_mode, ResponseMode::Observe); + } + + #[tokio::test] + async fn channel_mode_change_handles_empty_string_settings() { + let pool = memory_pool().await; + create_channel_settings_table(&pool).await; + + sqlx::query( + "INSERT INTO channel_settings (agent_id, conversation_id, settings) VALUES (?, ?, '')", + ) + .bind("agent") + .bind("conv") + .execute(&pool) + .await + .unwrap(); + + persist_response_mode(&pool, "agent", "conv", false, ResponseMode::MentionOnly) + .await + .unwrap(); + + let store = crate::conversation::ChannelSettingsStore::new(pool.clone()); + let loaded = store.get("agent", "conv").await.unwrap().unwrap(); + assert_eq!(loaded.response_mode, ResponseMode::MentionOnly); + } + + #[tokio::test] + async fn portal_missing_conversation_fails_without_writing() { + let pool = memory_pool().await; + create_portal_conversations_table(&pool).await; + + let result = + persist_response_mode(&pool, "agent", "missing", true, ResponseMode::Observe).await; + assert!(result.is_err()); + + let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM portal_conversations") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(count, 0, "a failed mode change must not create records"); + } +} diff --git a/src/commands/dispatch.rs b/src/commands/dispatch.rs new file mode 100644 index 000000000..b08a5b94f --- /dev/null +++ b/src/commands/dispatch.rs @@ -0,0 +1,218 @@ +//! Inbound-router command dispatch. +//! +//! Runs in the messaging layer, before a message reaches a channel's queue: +//! +//! 1. parse — unknown `/words` flow to the model as ordinary text +//! 2. usage errors reply immediately +//! 3. access check against the scope's authority list +//! 4. Control commands execute on the control plane — no inbound message is +//! created, so they work even while the channel is mid-turn +//! 5. Agent commands are rewritten to `MessageContent::Command` and injected; +//! busy policy applies, and a queued command is always acknowledged +//! +//! Replies (usage, denials, control output, busy acks) are ephemeral where +//! the platform supports it and degrade to plain messages elsewhere. + +use super::access::{self, AccessContext}; +use super::control::ControlPlane; +use super::registry::{BusyPolicy, CommandHandler, ControlAction, ParseResult, Surface}; +use crate::messaging::MessagingManager; +use crate::{InboundMessage, MessageContent, OutboundResponse}; +use std::sync::Arc; + +/// Outcome of router-side dispatch for one inbound message. +pub enum Dispatch { + /// Not a command: forward on the normal message path, untouched. + Forward, + /// Fully handled here (control executed, usage/denial/reject replied). + /// The message must not be forwarded. + Handled, + /// Agent command: the message content has been rewritten to + /// `MessageContent::Command`; forward it to the channel. + ForwardCommand, +} + +/// Facts about the matched scope the router already holds. +pub struct DispatchScope<'a> { + /// Authority list from the matched binding, when one matched and set + /// the key. An explicit empty list opens the scope; `None` falls back + /// to the adapter default. + pub binding_authority: Option<&'a [String]>, + /// Adapter-instance default authority lists. + pub adapter_defaults: &'a access::AdapterAuthorityDefaults, + /// Settings defaults from the matched binding. + pub binding_settings: Option<&'a crate::conversation::ConversationSettings>, + /// Whether the target channel currently has a turn in flight. + pub turn_active: bool, +} + +/// Classify and, where possible, fully handle a slash command. Replies are +/// sent on spawned tasks so the router loop never blocks on an adapter. +pub async fn dispatch_inbound( + message: &mut InboundMessage, + scope: DispatchScope<'_>, + deps: &crate::AgentDeps, + messaging: &Arc, +) -> Dispatch { + if message.source == "system" { + return Dispatch::Forward; + } + let text = match &message.content { + MessageContent::Text(text) => text.clone(), + // Command content arrives from surfaces that parse client-side + // (Discord interactions, Slack subcommands, the portal palette); it + // renders as "/name args" so the shared parser revalidates it. + MessageContent::Command { .. } => message.content.to_string(), + _ => return Dispatch::Forward, + }; + + let bot_username = message + .metadata + .get("telegram_bot_username") + .and_then(|value| value.as_str()); + let parsed = match super::REGISTRY.parse_addressed(&text, bot_username) { + ParseResult::NotACommand => return Dispatch::Forward, + ParseResult::Usage(_, usage) => { + reply_ephemeral(messaging, deps, message, usage); + return Dispatch::Handled; + } + ParseResult::Command(parsed) => parsed, + }; + + let surface = Surface::from_source(&message.source); + let context = AccessContext { + binding_authority: scope.binding_authority, + adapter_default: scope.adapter_defaults.for_adapter(message.adapter_key()), + sender_id: &message.sender_id, + sender_login: message + .metadata + .get("twitch_user_login") + .and_then(|value| value.as_str()), + }; + let is_authority = context.is_authority(); + if !context.allows(parsed.def) { + reply_ephemeral( + messaging, + deps, + message, + access::denial_text(parsed.def, surface), + ); + return Dispatch::Handled; + } + + match parsed.def.handler { + CommandHandler::Control(action) => { + let plane = ControlPlane { + deps: deps.clone(), + conversation_id: message.conversation_id.clone(), + adapter: message.adapter_key().to_string(), + binding_settings: scope.binding_settings.cloned(), + surface, + is_authority, + is_portal: message.adapter.as_deref() == Some("portal"), + }; + match action { + // State-mutating control commands execute inline. The router + // handles inbound messages sequentially, so awaiting here + // serializes mode changes in delivery order — an earlier + // /quiet can't land its store write and live update after a + // later /active. The work is a local SQLite roundtrip plus + // an in-memory cell update; only the reply delivery goes + // over the network, and that stays on a spawned task. + ControlAction::SetResponseMode(_) => { + let reply = plane.execute(action).await; + reply_ephemeral(messaging, deps, message, reply); + } + // Read-only control commands stay off the router's critical + // path entirely. + ControlAction::Status | ControlAction::Help | ControlAction::AgentId => { + let messaging = messaging.clone(); + let deps = deps.clone(); + let target = message.clone(); + tokio::spawn(async move { + let reply = plane.execute(action).await; + send_ephemeral(&messaging, &deps, &target, reply).await; + }); + } + } + Dispatch::Handled + } + CommandHandler::Agent(_) => { + if scope.turn_active { + match parsed.def.busy { + BusyPolicy::Reject => { + reply_ephemeral( + messaging, + deps, + message, + access::busy_reject_text(parsed.def), + ); + return Dispatch::Handled; + } + BusyPolicy::Queue => { + reply_ephemeral( + messaging, + deps, + message, + access::busy_queued_text(parsed.def), + ); + } + } + } + message.content = MessageContent::Command { + name: parsed.def.name.to_string(), + args: parsed.args, + }; + Dispatch::ForwardCommand + } + } +} + +fn reply_ephemeral( + messaging: &Arc, + deps: &crate::AgentDeps, + target: &InboundMessage, + text: String, +) { + let messaging = messaging.clone(); + let deps = deps.clone(); + let target = target.clone(); + tokio::spawn(async move { + send_ephemeral(&messaging, &deps, &target, text).await; + }); +} + +async fn send_ephemeral( + messaging: &Arc, + deps: &crate::AgentDeps, + target: &InboundMessage, + text: String, +) { + // Portal delivery rides the SSE event bus, not a messaging adapter — + // the portal adapter's respond() is deliberately a no-op. + if target.adapter_key() == "portal" { + if let Some(api_state) = deps.api_state.as_deref() { + api_state + .event_tx + .send(crate::api::ApiEvent::OutboundMessage { + agent_id: deps.agent_id.to_string(), + channel_id: target.conversation_id.clone(), + text, + }) + .ok(); + } + return; + } + + let response = OutboundResponse::Ephemeral { + text, + user_id: target.sender_id.clone(), + }; + if let Err(error) = messaging.respond(target, response).await { + tracing::warn!( + %error, + conversation_id = %target.conversation_id, + "failed to deliver command reply" + ); + } +} diff --git a/src/commands/native.rs b/src/commands/native.rs new file mode 100644 index 000000000..5d10c421b --- /dev/null +++ b/src/commands/native.rs @@ -0,0 +1,255 @@ +//! Platform-neutral specs for native command registration. +//! +//! Adapters map these to their platform types (serenity `CreateCommand`, +//! teloxide `BotCommand`, Slack subcommand listings); generating the specs +//! here keeps the registry the single source of truth and gives the parity +//! test one surface to walk per platform. + +use super::registry::{ArgSpec, CommandDef, REGISTRY, Surface}; + +/// Discord's application-command cap. Registration is cap-aware: commands +/// register in table order and overflow is dropped with one log line. +pub const DISCORD_COMMAND_CAP: usize = 100; + +/// Telegram menu limits (Bot API): command names 1–32 chars of +/// `[a-z0-9_]`, descriptions 3–256 chars, at most 100 entries. +pub const TELEGRAM_COMMAND_CAP: usize = 100; +const TELEGRAM_NAME_MAX: usize = 32; +const TELEGRAM_DESCRIPTION_MAX: usize = 256; + +/// A native command argument, mapped from [`ArgSpec`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum NativeArg { + /// Free-text option; `required` distinguishes `Required` from `Optional`. + Text { hint: String, required: bool }, + /// Closed choice set. + Choice { options: Vec }, +} + +/// A platform-neutral command spec for native registration. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct NativeCommandSpec { + pub name: String, + pub description: String, + pub arg: Option, +} + +fn spec_for(def: &CommandDef) -> NativeCommandSpec { + let arg = match def.args { + ArgSpec::None => None, + ArgSpec::Optional(hint) => Some(NativeArg::Text { + hint: hint.to_string(), + required: false, + }), + ArgSpec::Required(hint) => Some(NativeArg::Text { + hint: hint.to_string(), + required: true, + }), + ArgSpec::Choice(options) => Some(NativeArg::Choice { + options: options.iter().map(|option| option.to_string()).collect(), + }), + }; + NativeCommandSpec { + name: def.name.to_string(), + description: def.description.to_string(), + arg, + } +} + +/// Command specs to register as Discord application commands, in table +/// order, truncated to the cap. Returns the specs and the number dropped. +pub fn discord_commands() -> (Vec, usize) { + let available: Vec = REGISTRY + .defs() + .iter() + .filter(|def| def.availability.on(Surface::Discord)) + .map(spec_for) + .collect(); + let dropped = available.len().saturating_sub(DISCORD_COMMAND_CAP); + let mut specs = available; + specs.truncate(DISCORD_COMMAND_CAP); + (specs, dropped) +} + +/// A Telegram menu entry. `command` is the mangled form (hyphens become +/// underscores — Telegram rejects hyphens); the parser folds underscores +/// back to hyphens on resolve. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TelegramMenuEntry { + pub command: String, + pub description: String, +} + +/// The Telegram bot menu, generated from the registry. Choice values are +/// appended to the description as hints; entries that cannot be expressed +/// within Telegram's naming rules are skipped (none exist today — the +/// parity test enforces that skips stay deliberate). +pub fn telegram_menu() -> Vec { + REGISTRY + .defs() + .iter() + .filter(|def| def.availability.on(Surface::Telegram)) + .filter_map(|def| { + let command = def.name.replace('-', "_").to_ascii_lowercase(); + let valid = !command.is_empty() + && command.len() <= TELEGRAM_NAME_MAX + && command + .chars() + .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_'); + if !valid { + tracing::warn!( + command = def.name, + "command name cannot be registered in the telegram menu, skipping" + ); + return None; + } + let mut description = def.description.to_string(); + if let Some(hint) = def.args.hint() { + description.push_str(&format!(" {hint}")); + } + let mut description: String = + description.chars().take(TELEGRAM_DESCRIPTION_MAX).collect(); + if description.chars().count() < 3 { + description = format!("/{command}"); + } + Some(TelegramMenuEntry { + command, + description, + }) + }) + .take(TELEGRAM_COMMAND_CAP) + .collect() +} + +/// Slack `/spacebot` subcommands: every command available on Slack. The +/// umbrella command's help/usage listing derives from this. +pub fn slack_subcommands() -> Vec { + REGISTRY + .defs() + .iter() + .filter(|def| def.availability.on(Surface::Slack)) + .map(spec_for) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::commands::registry::{CommandHandler, Surface}; + + #[test] + fn discord_specs_cover_every_available_command_within_cap() { + let (specs, dropped) = discord_commands(); + assert_eq!(dropped, 0, "registry exceeds the discord cap"); + let expected: Vec<&str> = REGISTRY + .defs() + .iter() + .filter(|def| def.availability.on(Surface::Discord)) + .map(|def| def.name) + .collect(); + let actual: Vec<&str> = specs.iter().map(|spec| spec.name.as_str()).collect(); + assert_eq!(actual, expected); + } + + #[test] + fn telegram_menu_covers_every_available_command() { + let menu = telegram_menu(); + let expected: Vec = REGISTRY + .defs() + .iter() + .filter(|def| def.availability.on(Surface::Telegram)) + .map(|def| def.name.replace('-', "_")) + .collect(); + let actual: Vec<&str> = menu.iter().map(|entry| entry.command.as_str()).collect(); + assert_eq!(actual, expected, "telegram menu drops a command silently"); + } + + #[test] + fn telegram_menu_entries_satisfy_platform_limits() { + for entry in telegram_menu() { + assert!( + (1..=32).contains(&entry.command.len()), + "telegram command name length: {}", + entry.command + ); + assert!( + entry + .command + .chars() + .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_'), + "invalid telegram command name: {}", + entry.command + ); + assert!( + (3..=256).contains(&entry.description.chars().count()), + "telegram description length for /{}", + entry.command + ); + } + } + + #[test] + fn mangled_telegram_names_resolve_to_canonical_defs() { + for entry in telegram_menu() { + let def = REGISTRY + .resolve(&entry.command) + .unwrap_or_else(|| panic!("/{} does not resolve back", entry.command)); + assert_eq!(def.name.replace('-', "_"), entry.command); + } + } + + #[test] + fn every_command_is_exposed_on_every_available_surface() { + // Parity walk: a command available on a platform must be reachable + // through that platform's native surface (or the shared text + // parser for text adapters and portal, which accept everything). + // A command missing from a native listing needs an explicit + // availability opt-out in the registry, not a silent drop. + let (discord, _) = discord_commands(); + let telegram = telegram_menu(); + let slack = slack_subcommands(); + for def in REGISTRY.defs() { + if def.availability.on(Surface::Discord) { + assert!( + discord.iter().any(|spec| spec.name == def.name), + "/{} available on discord but not natively registered", + def.name + ); + } + if def.availability.on(Surface::Telegram) { + assert!( + telegram + .iter() + .any(|entry| entry.command == def.name.replace('-', "_")), + "/{} available on telegram but missing from the menu", + def.name + ); + } + if def.availability.on(Surface::Slack) { + assert!( + slack.iter().any(|spec| spec.name == def.name), + "/{} available on slack but missing from /spacebot subcommands", + def.name + ); + } + } + } + + #[test] + fn control_and_agent_commands_both_appear_in_native_listings() { + let (discord, _) = discord_commands(); + let has_control = discord.iter().any(|spec| { + matches!( + REGISTRY.resolve(&spec.name).unwrap().handler, + CommandHandler::Control(_) + ) + }); + let has_agent = discord.iter().any(|spec| { + matches!( + REGISTRY.resolve(&spec.name).unwrap().handler, + CommandHandler::Agent(_) + ) + }); + assert!(has_control && has_agent); + } +} diff --git a/src/commands/registry.rs b/src/commands/registry.rs index d9d401066..48d2fd948 100644 --- a/src/commands/registry.rs +++ b/src/commands/registry.rs @@ -22,6 +22,91 @@ pub struct CommandDef { pub args: ArgSpec, /// How this command executes. pub handler: CommandHandler, + /// Who may run this command in a scope with an authority list configured. + pub access: CommandAccess, + /// What happens when an Agent command arrives while a turn is in flight. + /// Control commands are busy-immune by construction; this field is only + /// consulted for `CommandHandler::Agent`. + pub busy: BusyPolicy, + /// Which platform surfaces expose this command. + pub availability: CommandAvailability, +} + +/// Who may run a command. Layered on top of admission (bindings and adapter +/// permission snapshots): access never widens who may talk to the agent, it +/// only restricts who may change state. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CommandAccess { + Everyone, + /// Requires the sender to be in the authority list for this scope. With + /// no authority list configured the command is open to everyone the + /// binding already admits. + Authority, +} + +/// What happens when an Agent command arrives while a turn is in flight. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum BusyPolicy { + /// Wait for the current turn, then run as a normal turn. + Queue, + /// Refuse mid-turn with a pointer to /stop. + Reject, +} + +/// Which platform surfaces expose a command. Used by native registration +/// (Discord application commands, Telegram menu, Slack subcommands) and the +/// parity test; the shared text parser accepts every command everywhere. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct CommandAvailability { + pub portal: bool, + pub discord: bool, + pub slack: bool, + pub telegram: bool, + /// Signal, Mattermost, Twitch, Email, Webhook. + pub text_adapters: bool, +} + +impl CommandAvailability { + pub const ALL: CommandAvailability = CommandAvailability { + portal: true, + discord: true, + slack: true, + telegram: true, + text_adapters: true, + }; + + pub fn on(&self, surface: Surface) -> bool { + match surface { + Surface::Portal => self.portal, + Surface::Discord => self.discord, + Surface::Slack => self.slack, + Surface::Telegram => self.telegram, + Surface::TextAdapters => self.text_adapters, + } + } +} + +/// A platform surface, for availability checks. Maps from +/// `InboundMessage.source`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Surface { + Portal, + Discord, + Slack, + Telegram, + TextAdapters, +} + +impl Surface { + pub fn from_source(source: &str) -> Surface { + match source { + "portal" => Surface::Portal, + "discord" => Surface::Discord, + "slack" => Surface::Slack, + "telegram" => Surface::Telegram, + _ => Surface::TextAdapters, + } + } } /// Grouping for `/help` display, rendered in [`CATEGORY_ORDER`]. @@ -138,15 +223,20 @@ impl CommandRegistry { } /// Resolve a bare command token (no slash) across names and aliases, - /// case-insensitively. + /// case-insensitively. Underscores fold to hyphens so Telegram's menu + /// form of a hyphenated name (`/mention_only` — Telegram rejects hyphens + /// in registered commands) resolves to the canonical def. pub fn resolve(&self, name: &str) -> Option<&'static CommandDef> { - self.defs.iter().find(|def| { - def.name.eq_ignore_ascii_case(name) - || def - .aliases - .iter() - .any(|alias| alias.eq_ignore_ascii_case(name)) - }) + let folded = |candidate: &str| { + candidate.len() == name.len() + && candidate + .chars() + .zip(name.chars()) + .all(|(a, b)| a.eq_ignore_ascii_case(&b) || (a == '-' && b == '_')) + }; + self.defs + .iter() + .find(|def| folded(def.name) || def.aliases.iter().any(|alias| folded(alias))) } /// Parse message text into a command. @@ -156,6 +246,13 @@ impl CommandRegistry { /// never parse as commands, and un-mangles smart dashes in arguments /// (iOS autocorrects `--` to `—` and `-` to `–`). pub fn parse(&self, text: &str) -> ParseResult { + self.parse_addressed(text, None) + } + + /// [`parse`](Self::parse) with the receiving bot's username, when the + /// surface knows it. A `/cmd@other_bot` addressed to a different bot in a + /// Telegram group is conversation, not a command for us. + pub fn parse_addressed(&self, text: &str, bot_username: Option<&str>) -> ParseResult { let trimmed = text.trim(); let Some(body) = trimmed.strip_prefix('/') else { return ParseResult::NotACommand; @@ -167,7 +264,17 @@ impl CommandRegistry { if token.is_empty() || token.contains('/') { return ParseResult::NotACommand; } - let token = token.split('@').next().unwrap_or(token); + let token = match token.split_once('@') { + Some((bare, addressed)) => { + if let Some(own_name) = bot_username + && !addressed.eq_ignore_ascii_case(own_name) + { + return ParseResult::NotACommand; + } + bare + } + None => token, + }; let Some(def) = self.resolve(token) else { return ParseResult::NotACommand; }; @@ -206,12 +313,25 @@ impl CommandRegistry { /// aliases and argument hints. Generated so it cannot drift from the /// table. pub fn help_text(&self) -> String { + self.help_text_for(None, true) + } + + /// [`help_text`](Self::help_text) filtered to what the caller can see and + /// run: commands available on `surface` (all surfaces when `None`), minus + /// Authority commands when the caller lacks authority. A user never sees + /// a command they can't run. + pub fn help_text_for(&self, surface: Option, is_authority: bool) -> String { + let visible = |def: &&CommandDef| { + surface.is_none_or(|surface| def.availability.on(surface)) + && (is_authority || def.access == CommandAccess::Everyone) + }; let mut out = String::from("commands:"); for (category, label) in CATEGORY_ORDER { let defs: Vec<&CommandDef> = self .defs .iter() .filter(|def| def.category == *category) + .filter(visible) .collect(); if defs.is_empty() { continue; @@ -265,6 +385,9 @@ pub static COMMANDS: &[CommandDef] = &[ aliases: &[], args: ArgSpec::None, handler: CommandHandler::Control(ControlAction::SetResponseMode(ResponseMode::Active)), + access: CommandAccess::Authority, + busy: BusyPolicy::Queue, + availability: CommandAvailability::ALL, }, CommandDef { name: "mention-only", @@ -273,6 +396,9 @@ pub static COMMANDS: &[CommandDef] = &[ aliases: &[], args: ArgSpec::None, handler: CommandHandler::Control(ControlAction::SetResponseMode(ResponseMode::MentionOnly)), + access: CommandAccess::Authority, + busy: BusyPolicy::Queue, + availability: CommandAvailability::ALL, }, CommandDef { name: "quiet", @@ -281,6 +407,9 @@ pub static COMMANDS: &[CommandDef] = &[ aliases: &["observe"], args: ArgSpec::None, handler: CommandHandler::Control(ControlAction::SetResponseMode(ResponseMode::Observe)), + access: CommandAccess::Authority, + busy: BusyPolicy::Queue, + availability: CommandAvailability::ALL, }, CommandDef { name: "tasks", @@ -289,6 +418,9 @@ pub static COMMANDS: &[CommandDef] = &[ aliases: &[], args: ArgSpec::None, handler: CommandHandler::Agent(AgentAction::PromptTemplate("commands/tasks")), + access: CommandAccess::Everyone, + busy: BusyPolicy::Queue, + availability: CommandAvailability::ALL, }, CommandDef { name: "today", @@ -297,6 +429,9 @@ pub static COMMANDS: &[CommandDef] = &[ aliases: &[], args: ArgSpec::None, handler: CommandHandler::Agent(AgentAction::PromptTemplate("commands/today")), + access: CommandAccess::Everyone, + busy: BusyPolicy::Queue, + availability: CommandAvailability::ALL, }, CommandDef { name: "digest", @@ -305,6 +440,9 @@ pub static COMMANDS: &[CommandDef] = &[ aliases: &[], args: ArgSpec::None, handler: CommandHandler::Agent(AgentAction::PromptTemplate("commands/digest")), + access: CommandAccess::Everyone, + busy: BusyPolicy::Queue, + availability: CommandAvailability::ALL, }, CommandDef { name: "status", @@ -313,6 +451,9 @@ pub static COMMANDS: &[CommandDef] = &[ aliases: &[], args: ArgSpec::None, handler: CommandHandler::Control(ControlAction::Status), + access: CommandAccess::Everyone, + busy: BusyPolicy::Queue, + availability: CommandAvailability::ALL, }, CommandDef { name: "help", @@ -321,6 +462,9 @@ pub static COMMANDS: &[CommandDef] = &[ aliases: &["commands"], args: ArgSpec::None, handler: CommandHandler::Control(ControlAction::Help), + access: CommandAccess::Everyone, + busy: BusyPolicy::Queue, + availability: CommandAvailability::ALL, }, CommandDef { name: "agent-id", @@ -329,6 +473,9 @@ pub static COMMANDS: &[CommandDef] = &[ aliases: &[], args: ArgSpec::None, handler: CommandHandler::Control(ControlAction::AgentId), + access: CommandAccess::Everyone, + busy: BusyPolicy::Queue, + availability: CommandAvailability::ALL, }, ]; @@ -411,6 +558,9 @@ mod tests { aliases: &[], args: ArgSpec::Optional("[query]"), handler: CommandHandler::Control(ControlAction::Help), + access: CommandAccess::Everyone, + busy: BusyPolicy::Queue, + availability: CommandAvailability::ALL, }, CommandDef { name: "background", @@ -419,6 +569,9 @@ mod tests { aliases: &[], args: ArgSpec::Required(""), handler: CommandHandler::Control(ControlAction::Help), + access: CommandAccess::Everyone, + busy: BusyPolicy::Queue, + availability: CommandAvailability::ALL, }, CommandDef { name: "voice", @@ -427,6 +580,9 @@ mod tests { aliases: &[], args: ArgSpec::Choice(&["on", "off", "status"]), handler: CommandHandler::Control(ControlAction::Help), + access: CommandAccess::Everyone, + busy: BusyPolicy::Queue, + availability: CommandAvailability::ALL, }, ]; diff --git a/src/config.rs b/src/config.rs index 964142f8d..be0d352f4 100644 --- a/src/config.rs +++ b/src/config.rs @@ -560,18 +560,19 @@ bind = "127.0.0.1" /// Helper to build a minimal `SlackConfig` for permission tests. fn slack_config_with_dm_users(dm_allowed_users: Vec) -> SlackConfig { SlackConfig { + authority: vec![], enabled: true, bot_token: "xoxb-test".into(), app_token: "xapp-test".into(), instances: vec![], dm_allowed_users, - commands: vec![], } } /// Helper to build a Slack binding with optional dm_allowed_users. fn slack_binding(workspace_id: Option<&str>, dm_allowed_users: Vec) -> Binding { Binding { + authority: None, agent_id: "test-agent".into(), channel: "slack".into(), adapter: None, @@ -1473,6 +1474,7 @@ id = "main" #[test] fn binding_runtime_adapter_key_method() { let binding = Binding { + authority: None, agent_id: "main".into(), channel: "telegram".into(), adapter: Some("sales".into()), @@ -1491,6 +1493,7 @@ id = "main" #[test] fn binding_uses_default_adapter() { let binding = Binding { + authority: None, agent_id: "main".into(), channel: "discord".into(), adapter: None, @@ -1524,6 +1527,7 @@ id = "main" #[test] fn adapter_matches_default_binding_default_message() { let binding = Binding { + authority: None, agent_id: "main".into(), channel: "telegram".into(), adapter: None, @@ -1543,6 +1547,7 @@ id = "main" #[test] fn adapter_matches_named_binding_named_message() { let binding = Binding { + authority: None, agent_id: "main".into(), channel: "telegram".into(), adapter: Some("support".into()), @@ -1562,6 +1567,7 @@ id = "main" #[test] fn adapter_mismatch_named_vs_default() { let binding = Binding { + authority: None, agent_id: "main".into(), channel: "telegram".into(), adapter: Some("support".into()), @@ -1581,6 +1587,7 @@ id = "main" #[test] fn adapter_mismatch_default_vs_named() { let binding = Binding { + authority: None, agent_id: "main".into(), channel: "telegram".into(), adapter: None, @@ -1600,6 +1607,7 @@ id = "main" #[test] fn adapter_mismatch_different_names() { let binding = Binding { + authority: None, agent_id: "main".into(), channel: "telegram".into(), adapter: Some("support".into()), @@ -1616,15 +1624,52 @@ id = "main" assert!(!binding_adapter_matches(&binding, &message)); } + #[test] + fn matched_binding_resolves_preassigned_portal_messages() { + let bindings = vec![Binding { + authority: Some(vec!["42".into()]), + agent_id: "orion".into(), + channel: "portal".into(), + adapter: None, + guild_id: None, + workspace_id: None, + chat_id: None, + team_id: None, + channel_ids: vec![], + require_mention: false, + dm_allowed_users: vec![], + settings: None, + }]; + + // Portal sends preassign `agent_id`, so the router skips agent + // resolution — the binding scan must still find this binding. + let mut message = test_inbound_message("portal", Some("portal")); + message.agent_id = Some(std::sync::Arc::from("orion")); + + let matched = + matched_binding(&bindings, &message).expect("portal binding should match on agent_id"); + assert_eq!(matched.agent_id, "orion"); + assert_eq!( + matched_binding_authority(&bindings, &message), + Some(vec!["42".to_string()]) + ); + + let mut other_agent = test_inbound_message("portal", Some("portal")); + other_agent.agent_id = Some(std::sync::Arc::from("someone-else")); + assert!(matched_binding(&bindings, &other_agent).is_none()); + } + #[test] fn validate_named_adapters_valid_config() { let messaging = MessagingConfig { discord: None, slack: None, telegram: Some(TelegramConfig { + authority: vec![], enabled: true, token: "tok".into(), instances: vec![TelegramInstanceConfig { + authority: vec![], name: "support".into(), enabled: true, token: "tok2".into(), @@ -1640,6 +1685,7 @@ id = "main" }; let bindings = vec![ Binding { + authority: None, agent_id: "main".into(), channel: "telegram".into(), adapter: None, @@ -1653,6 +1699,7 @@ id = "main" settings: None, }, Binding { + authority: None, agent_id: "support-agent".into(), channel: "telegram".into(), adapter: Some("support".into()), @@ -1677,6 +1724,7 @@ id = "main" discord: None, slack: None, telegram: Some(TelegramConfig { + authority: vec![], enabled: true, token: "tok".into(), instances: vec![], @@ -1689,6 +1737,7 @@ id = "main" mattermost: None, }; let bindings = vec![Binding { + authority: None, agent_id: "main".into(), channel: "telegram".into(), adapter: Some("nonexistent".into()), @@ -1731,6 +1780,7 @@ id = "main" slack: None, telegram: None, email: Some(EmailConfig { + authority: vec![], enabled: true, imap_host: "imap.test.com".into(), imap_port: 993, @@ -1757,6 +1807,7 @@ id = "main" mattermost: None, }; let bindings = vec![Binding { + authority: None, agent_id: "main".into(), channel: "email".into(), adapter: Some("named".into()), @@ -1783,9 +1834,11 @@ id = "main" discord: None, slack: None, telegram: Some(TelegramConfig { + authority: vec![], enabled: true, token: "".into(), // no default credential instances: vec![TelegramInstanceConfig { + authority: vec![], name: "support".into(), enabled: true, token: "tok".into(), @@ -1801,6 +1854,7 @@ id = "main" }; // Binding targets default adapter, but no default credentials exist let bindings = vec![Binding { + authority: None, agent_id: "main".into(), channel: "telegram".into(), adapter: None, @@ -1827,9 +1881,11 @@ id = "main" discord: None, slack: None, telegram: Some(TelegramConfig { + authority: vec![], enabled: true, token: "tok".into(), instances: vec![TelegramInstanceConfig { + authority: vec![], name: "support".into(), enabled: true, token: "tok2".into(), @@ -1846,6 +1902,7 @@ id = "main" let bindings = vec![ // Valid: default adapter with credentials Binding { + authority: None, agent_id: "agent-a".into(), channel: "telegram".into(), adapter: None, @@ -1860,6 +1917,7 @@ id = "main" }, // Invalid: references a non-existent named adapter Binding { + authority: None, agent_id: "agent-b".into(), channel: "telegram".into(), adapter: Some("ghost".into()), @@ -1874,6 +1932,7 @@ id = "main" }, // Valid: references an existing named adapter Binding { + authority: None, agent_id: "agent-c".into(), channel: "telegram".into(), adapter: Some("support".into()), @@ -1888,6 +1947,7 @@ id = "main" }, // Invalid: no discord config at all Binding { + authority: None, agent_id: "agent-d".into(), channel: "discord".into(), adapter: None, @@ -1925,6 +1985,7 @@ id = "main" mattermost: None, }; let bindings = vec![Binding { + authority: None, agent_id: "main".into(), channel: "telegram".into(), adapter: None, @@ -1958,6 +2019,7 @@ id = "main" mattermost: None, }; let bindings = vec![Binding { + authority: None, agent_id: "main".into(), channel: "telegram".into(), adapter: None, @@ -1983,9 +2045,11 @@ id = "main" discord: None, slack: None, telegram: Some(TelegramConfig { + authority: vec![], enabled: true, token: "tok".into(), instances: vec![TelegramInstanceConfig { + authority: vec![], name: "support".into(), enabled: false, token: "tok2".into(), @@ -2000,6 +2064,7 @@ id = "main" mattermost: None, }; let bindings = vec![Binding { + authority: None, agent_id: "main".into(), channel: "telegram".into(), adapter: Some("support".into()), @@ -2026,6 +2091,7 @@ id = "main" discord: None, slack: None, telegram: Some(TelegramConfig { + authority: vec![], enabled: false, token: "tok".into(), instances: vec![], @@ -2038,6 +2104,7 @@ id = "main" mattermost: None, }; let bindings = vec![Binding { + authority: None, agent_id: "main".into(), channel: "telegram".into(), adapter: None, diff --git a/src/config/load.rs b/src/config/load.rs index 072b9dd3a..5b3d3e787 100644 --- a/src/config/load.rs +++ b/src/config/load.rs @@ -18,9 +18,9 @@ use super::{ MattermostInstanceConfig, McpServerConfig, McpTransport, MemoryJanitorConfig, MemoryPersistenceConfig, MessagingConfig, MetricsConfig, OpenCodeConfig, ParticipantContextConfig, ProjectsConfig, ProviderConfig, ReflectionConfig, SignalConfig, - SignalInstanceConfig, SkillsConfig, SlackCommandConfig, SlackConfig, SlackInstanceConfig, - TelegramConfig, TelegramInstanceConfig, TelemetryConfig, TwitchConfig, TwitchInstanceConfig, - WarmupConfig, WebhookConfig, normalize_adapter, validate_named_messaging_adapters, + SignalInstanceConfig, SkillsConfig, SlackConfig, SlackInstanceConfig, TelegramConfig, + TelegramInstanceConfig, TelemetryConfig, TwitchConfig, TwitchInstanceConfig, WarmupConfig, + WebhookConfig, normalize_adapter, validate_named_messaging_adapters, }; use crate::error::{ConfigError, Result}; @@ -2096,6 +2096,7 @@ impl Config { ); } DiscordInstanceConfig { + authority: instance.authority, name: instance.name, enabled: instance.enabled && token.is_some(), token: token.unwrap_or_default(), @@ -2114,6 +2115,7 @@ impl Config { } Some(DiscordConfig { + authority: d.authority, enabled: d.enabled, token: token.unwrap_or_default(), instances, @@ -2138,20 +2140,12 @@ impl Config { } let has_credentials = bot_token.is_some() && app_token.is_some(); SlackInstanceConfig { + authority: instance.authority, name: instance.name, enabled: instance.enabled && has_credentials, bot_token: bot_token.unwrap_or_default(), app_token: app_token.unwrap_or_default(), dm_allowed_users: instance.dm_allowed_users, - commands: instance - .commands - .into_iter() - .map(|command| SlackCommandConfig { - command: command.command, - agent_id: command.agent_id, - description: command.description, - }) - .collect(), } }) .collect::>(); @@ -2168,20 +2162,12 @@ impl Config { } Some(SlackConfig { + authority: s.authority, enabled: s.enabled, bot_token: bot_token.unwrap_or_default(), app_token: app_token.unwrap_or_default(), instances, dm_allowed_users: s.dm_allowed_users, - commands: s - .commands - .into_iter() - .map(|c| SlackCommandConfig { - command: c.command, - agent_id: c.agent_id, - description: c.description, - }) - .collect(), }) }), telegram: toml.messaging.telegram.and_then(|t| { @@ -2197,6 +2183,7 @@ impl Config { ); } TelegramInstanceConfig { + authority: instance.authority, name: instance.name, enabled: instance.enabled && token.is_some(), token: token.unwrap_or_default(), @@ -2214,6 +2201,7 @@ impl Config { } Some(TelegramConfig { + authority: t.authority, enabled: t.enabled, token: token.unwrap_or_default(), instances, @@ -2267,6 +2255,7 @@ impl Config { instance.from_name.as_deref().and_then(resolve_env_value); EmailInstanceConfig { + authority: instance.authority, name: instance.name, enabled: instance.enabled && has_credentials, imap_host: imap_host.unwrap_or_default(), @@ -2338,6 +2327,7 @@ impl Config { .or_else(|| email.from_name.as_deref().and_then(resolve_env_value)); Some(EmailConfig { + authority: email.authority, enabled: email.enabled, imap_host, imap_port: email.imap_port, @@ -2396,6 +2386,7 @@ impl Config { .as_deref() .and_then(resolve_env_value); TwitchInstanceConfig { + authority: instance.authority, name: instance.name, enabled: instance.enabled && has_credentials, username: username.unwrap_or_default(), @@ -2436,6 +2427,7 @@ impl Config { .and_then(resolve_env_value) .or_else(|| std::env::var("TWITCH_REFRESH_TOKEN").ok()); Some(TwitchConfig { + authority: t.authority, enabled: t.enabled, username: username.unwrap_or_default(), oauth_token: oauth_token.unwrap_or_default(), @@ -2462,6 +2454,7 @@ impl Config { } let has_credentials = http_url.is_some() && account.is_some(); SignalInstanceConfig { + authority: instance.authority, name: instance.name, enabled: instance.enabled && has_credentials, http_url: http_url.unwrap_or_default(), @@ -2486,6 +2479,7 @@ impl Config { } Some(SignalConfig { + authority: s.authority, enabled: s.enabled, http_url: http_url.unwrap_or_default(), account: account.unwrap_or_default(), @@ -2511,6 +2505,7 @@ impl Config { ); } MattermostInstanceConfig { + authority: instance.authority, name: instance.name, enabled: instance.enabled && has_credentials, base_url: base_url.unwrap_or_default(), @@ -2535,6 +2530,7 @@ impl Config { } Some(MattermostConfig { + authority: mm.authority, enabled: mm.enabled, base_url: base_url.unwrap_or_default(), token: token.unwrap_or_default(), @@ -2594,6 +2590,7 @@ impl Config { cs }); Binding { + authority: b.authority, agent_id: b.agent_id, channel: b.channel, adapter: normalize_adapter(b.adapter), diff --git a/src/config/toml_schema.rs b/src/config/toml_schema.rs index 5941bc16f..8cdda4e42 100644 --- a/src/config/toml_schema.rs +++ b/src/config/toml_schema.rs @@ -587,6 +587,8 @@ pub(super) struct TomlDiscordConfig { #[serde(default)] pub(super) dm_allowed_users: Vec, #[serde(default)] + pub(super) authority: Vec, + #[serde(default)] pub(super) allow_bot_messages: bool, } @@ -599,6 +601,8 @@ pub(super) struct TomlDiscordInstanceConfig { #[serde(default)] pub(super) dm_allowed_users: Vec, #[serde(default)] + pub(super) authority: Vec, + #[serde(default)] pub(super) allow_bot_messages: bool, } @@ -613,7 +617,7 @@ pub(super) struct TomlSlackConfig { #[serde(default)] pub(super) dm_allowed_users: Vec, #[serde(default)] - pub(super) commands: Vec, + pub(super) authority: Vec, } #[derive(Deserialize)] @@ -626,14 +630,7 @@ pub(super) struct TomlSlackInstanceConfig { #[serde(default)] pub(super) dm_allowed_users: Vec, #[serde(default)] - pub(super) commands: Vec, -} - -#[derive(Deserialize)] -pub(super) struct TomlSlackCommandConfig { - pub(super) command: String, - pub(super) agent_id: String, - pub(super) description: Option, + pub(super) authority: Vec, } #[derive(Deserialize)] @@ -645,6 +642,8 @@ pub(super) struct TomlTelegramConfig { pub(super) instances: Vec, #[serde(default)] pub(super) dm_allowed_users: Vec, + #[serde(default)] + pub(super) authority: Vec, } #[derive(Deserialize)] @@ -655,6 +654,8 @@ pub(super) struct TomlTelegramInstanceConfig { pub(super) token: Option, #[serde(default)] pub(super) dm_allowed_users: Vec, + #[serde(default)] + pub(super) authority: Vec, } #[derive(Deserialize)] @@ -683,6 +684,8 @@ pub(super) struct TomlEmailConfig { pub(super) folders: Vec, #[serde(default)] pub(super) allowed_senders: Vec, + #[serde(default)] + pub(super) authority: Vec, #[serde(default = "default_email_max_body_bytes")] pub(super) max_body_bytes: usize, #[serde(default = "default_email_max_attachment_bytes")] @@ -718,6 +721,8 @@ pub(super) struct TomlEmailInstanceConfig { pub(super) folders: Vec, #[serde(default)] pub(super) allowed_senders: Vec, + #[serde(default)] + pub(super) authority: Vec, #[serde(default = "default_email_max_body_bytes")] pub(super) max_body_bytes: usize, #[serde(default = "default_email_max_attachment_bytes")] @@ -748,6 +753,8 @@ pub(super) struct TomlTwitchConfig { pub(super) instances: Vec, #[serde(default)] pub(super) channels: Vec, + #[serde(default)] + pub(super) authority: Vec, pub(super) trigger_prefix: Option, } @@ -763,6 +770,8 @@ pub(super) struct TomlTwitchInstanceConfig { pub(super) refresh_token: Option, #[serde(default)] pub(super) channels: Vec, + #[serde(default)] + pub(super) authority: Vec, pub(super) trigger_prefix: Option, } @@ -777,6 +786,8 @@ pub(super) struct TomlSignalConfig { #[serde(default)] pub(super) dm_allowed_users: Vec, #[serde(default)] + pub(super) authority: Vec, + #[serde(default)] pub(super) group_ids: Vec, #[serde(default)] pub(super) group_allowed_users: Vec, @@ -794,6 +805,8 @@ pub(super) struct TomlSignalInstanceConfig { #[serde(default)] pub(super) dm_allowed_users: Vec, #[serde(default)] + pub(super) authority: Vec, + #[serde(default)] pub(super) group_ids: Vec, #[serde(default)] pub(super) group_allowed_users: Vec, @@ -871,6 +884,11 @@ pub(super) struct TomlBinding { pub(super) require_mention: bool, #[serde(default)] pub(super) dm_allowed_users: Vec, + /// `None` when the key is omitted, `Some(vec![])` for an explicit + /// `authority = []` — the distinction decides whether the adapter + /// default applies. + #[serde(default)] + pub(super) authority: Option>, #[serde(default)] pub(super) settings: Option, } @@ -886,6 +904,8 @@ pub(super) struct TomlMattermostConfig { pub(super) instances: Vec, #[serde(default)] pub(super) dm_allowed_users: Vec, + #[serde(default)] + pub(super) authority: Vec, #[serde(default = "default_mattermost_max_attachment_bytes")] pub(super) max_attachment_bytes: usize, } @@ -900,6 +920,8 @@ pub(super) struct TomlMattermostInstanceConfig { pub(super) team_id: Option, #[serde(default)] pub(super) dm_allowed_users: Vec, + #[serde(default)] + pub(super) authority: Vec, #[serde(default = "default_mattermost_max_attachment_bytes")] pub(super) max_attachment_bytes: usize, } diff --git a/src/config/types.rs b/src/config/types.rs index b496f3fbc..82a42d3a6 100644 --- a/src/config/types.rs +++ b/src/config/types.rs @@ -1972,6 +1972,12 @@ pub struct Binding { pub require_mention: bool, /// User IDs allowed to DM the bot through this binding. pub dm_allowed_users: Vec, + /// User IDs allowed to run authority-gated slash commands in scopes + /// matched by this binding. `None` (key omitted) falls back to the + /// adapter-instance default; `Some(vec![])` (explicit `authority = []`) + /// opens commands to everyone this binding admits, even when the + /// adapter default is restricted. + pub authority: Option>, /// Default conversation settings for channels matched by this binding. pub settings: Option, } @@ -2676,6 +2682,28 @@ pub fn resolve_agent_for_message( Some((std::sync::Arc::from(default_agent_id), None)) } +/// The binding that routes `message`, matched on routing criteria only. +/// Uses the same first-match-wins scan as [`resolve_agent_for_message`], so +/// the result is always the binding that actually routed the message. +pub fn matched_binding<'a>( + bindings: &'a [Binding], + message: &crate::InboundMessage, +) -> Option<&'a Binding> { + bindings + .iter() + .find(|binding| binding.matches_route(message)) +} + +/// Authority list of the binding that routes `message`, for slash-command +/// access checks. `None` when no binding matched or the matched binding +/// omits `authority` — both fall back to the adapter default downstream. +pub fn matched_binding_authority( + bindings: &[Binding], + message: &crate::InboundMessage, +) -> Option> { + matched_binding(bindings, message).and_then(|binding| binding.authority.clone()) +} + // --------------------------------------------------------------------------- // Messaging platform configs // --------------------------------------------------------------------------- @@ -2701,6 +2729,8 @@ pub struct DiscordConfig { pub instances: Vec, /// User IDs allowed to DM the bot. If empty, DMs are ignored entirely. pub dm_allowed_users: Vec, + /// Default authority list for slash commands on this adapter; binding-level lists take precedence. + pub authority: Vec, /// Whether to process messages from other bots (self-messages are always ignored). pub allow_bot_messages: bool, } @@ -2712,6 +2742,8 @@ pub struct DiscordInstanceConfig { pub token: String, /// User IDs allowed to DM this bot instance. pub dm_allowed_users: Vec, + /// Default authority list for slash commands on this adapter; binding-level lists take precedence. + pub authority: Vec, /// Whether this bot instance processes messages from other bots. pub allow_bot_messages: bool, } @@ -2723,6 +2755,7 @@ impl std::fmt::Debug for DiscordInstanceConfig { .field("enabled", &self.enabled) .field("token", &"[REDACTED]") .field("dm_allowed_users", &self.dm_allowed_users) + .field("authority", &self.authority) .field("allow_bot_messages", &self.allow_bot_messages) .finish() } @@ -2735,6 +2768,7 @@ impl std::fmt::Debug for DiscordConfig { .field("token", &"[REDACTED]") .field("instances", &self.instances) .field("dm_allowed_users", &self.dm_allowed_users) + .field("authority", &self.authority) .field("allow_bot_messages", &self.allow_bot_messages) .finish() } @@ -2761,20 +2795,6 @@ impl SystemSecrets for DiscordConfig { } } -/// A single slash command definition for the Slack adapter. -/// -/// Maps a Slack slash command (e.g. `/ask`) to a target agent. -/// Commands not listed here are acknowledged but produce a "not configured" reply. -#[derive(Debug, Clone)] -pub struct SlackCommandConfig { - /// The slash command string exactly as Slack sends it, e.g. `"/ask"`. - pub command: String, - /// ID of the agent that should handle this command. - pub agent_id: String, - /// Short description shown in Slack's command autocomplete hint (optional). - pub description: Option, -} - #[derive(Clone)] pub struct SlackConfig { pub enabled: bool, @@ -2784,8 +2804,8 @@ pub struct SlackConfig { pub instances: Vec, /// User IDs allowed to DM the bot. If empty, DMs are ignored entirely. pub dm_allowed_users: Vec, - /// Slash command definitions. If empty, all slash commands are ignored. - pub commands: Vec, + /// Default authority list for slash commands on this adapter; binding-level lists take precedence. + pub authority: Vec, } #[derive(Clone)] @@ -2796,8 +2816,8 @@ pub struct SlackInstanceConfig { pub app_token: String, /// User IDs allowed to DM this app instance. pub dm_allowed_users: Vec, - /// Slash command definitions for this app instance. - pub commands: Vec, + /// Default authority list for slash commands on this adapter; binding-level lists take precedence. + pub authority: Vec, } impl std::fmt::Debug for SlackInstanceConfig { @@ -2808,7 +2828,7 @@ impl std::fmt::Debug for SlackInstanceConfig { .field("bot_token", &"[REDACTED]") .field("app_token", &"[REDACTED]") .field("dm_allowed_users", &self.dm_allowed_users) - .field("commands", &self.commands) + .field("authority", &self.authority) .finish() } } @@ -2821,7 +2841,7 @@ impl std::fmt::Debug for SlackConfig { .field("app_token", &"[REDACTED]") .field("instances", &self.instances) .field("dm_allowed_users", &self.dm_allowed_users) - .field("commands", &self.commands) + .field("authority", &self.authority) .finish() } } @@ -2865,6 +2885,8 @@ pub struct TelegramConfig { pub instances: Vec, /// User IDs allowed to DM the bot. If empty, DMs are ignored entirely. pub dm_allowed_users: Vec, + /// Default authority list for slash commands on this adapter; binding-level lists take precedence. + pub authority: Vec, } #[derive(Clone)] @@ -2874,6 +2896,8 @@ pub struct TelegramInstanceConfig { pub token: String, /// User IDs allowed to DM this bot instance. pub dm_allowed_users: Vec, + /// Default authority list for slash commands on this adapter; binding-level lists take precedence. + pub authority: Vec, } impl std::fmt::Debug for TelegramInstanceConfig { @@ -2883,6 +2907,7 @@ impl std::fmt::Debug for TelegramInstanceConfig { .field("enabled", &self.enabled) .field("token", &"[REDACTED]") .field("dm_allowed_users", &self.dm_allowed_users) + .field("authority", &self.authority) .finish() } } @@ -2894,6 +2919,7 @@ impl std::fmt::Debug for TelegramConfig { .field("token", &"[REDACTED]") .field("instances", &self.instances) .field("dm_allowed_users", &self.dm_allowed_users) + .field("authority", &self.authority) .finish() } } @@ -2937,6 +2963,8 @@ pub struct EmailConfig { pub poll_interval_secs: u64, pub folders: Vec, pub allowed_senders: Vec, + /// Default authority list for slash commands on this adapter; binding-level lists take precedence. + pub authority: Vec, pub max_body_bytes: usize, pub max_attachment_bytes: usize, pub instances: Vec, @@ -2962,6 +2990,8 @@ pub struct EmailInstanceConfig { pub poll_interval_secs: u64, pub folders: Vec, pub allowed_senders: Vec, + /// Default authority list for slash commands on this adapter; binding-level lists take precedence. + pub authority: Vec, pub max_body_bytes: usize, pub max_attachment_bytes: usize, } @@ -2986,6 +3016,7 @@ impl std::fmt::Debug for EmailInstanceConfig { .field("poll_interval_secs", &self.poll_interval_secs) .field("folders", &self.folders) .field("allowed_senders", &"[REDACTED]") + .field("authority", &self.authority) .field("max_body_bytes", &self.max_body_bytes) .field("max_attachment_bytes", &self.max_attachment_bytes) .finish() @@ -3011,6 +3042,7 @@ impl std::fmt::Debug for EmailConfig { .field("poll_interval_secs", &self.poll_interval_secs) .field("folders", &self.folders) .field("allowed_senders", &"[REDACTED]") + .field("authority", &self.authority) .field("max_body_bytes", &self.max_body_bytes) .field("max_attachment_bytes", &self.max_attachment_bytes) .finish() @@ -3076,6 +3108,8 @@ pub struct TwitchConfig { pub instances: Vec, /// Channels to join (without the # prefix). pub channels: Vec, + /// Default authority list for slash commands on this adapter; binding-level lists take precedence. + pub authority: Vec, /// Optional prefix that triggers the bot (e.g. "!ask"). If empty, all messages are processed. pub trigger_prefix: Option, } @@ -3091,6 +3125,8 @@ pub struct TwitchInstanceConfig { pub refresh_token: Option, /// Channels to join (without the # prefix). pub channels: Vec, + /// Default authority list for slash commands on this adapter; binding-level lists take precedence. + pub authority: Vec, /// Optional prefix that triggers the bot for this instance. pub trigger_prefix: Option, } @@ -3112,6 +3148,7 @@ impl std::fmt::Debug for TwitchInstanceConfig { &self.refresh_token.as_ref().map(|_| "[REDACTED]"), ) .field("channels", &self.channels) + .field("authority", &self.authority) .field("trigger_prefix", &self.trigger_prefix) .finish() } @@ -3125,6 +3162,7 @@ impl std::fmt::Debug for TwitchConfig { .field("oauth_token", &"[REDACTED]") .field("instances", &self.instances) .field("channels", &self.channels) + .field("authority", &self.authority) .field("trigger_prefix", &self.trigger_prefix) .finish() } @@ -3201,6 +3239,8 @@ pub struct SignalConfig { pub instances: Vec, /// Phone numbers or UUIDs allowed to DM the bot. If empty, DMs are ignored. pub dm_allowed_users: Vec, + /// Default authority list for slash commands on this adapter; binding-level lists take precedence. + pub authority: Vec, /// Group IDs allowed for this adapter. If empty, all groups are blocked /// (same as `None` in the permission filter — groups are opt-in only). pub group_ids: Vec, @@ -3221,6 +3261,8 @@ pub struct SignalInstanceConfig { pub account: String, /// Phone numbers or UUIDs allowed to DM this instance. pub dm_allowed_users: Vec, + /// Default authority list for slash commands on this adapter; binding-level lists take precedence. + pub authority: Vec, /// Group IDs allowed for this instance. pub group_ids: Vec, /// User IDs allowed to message in Signal groups for this instance. @@ -3237,6 +3279,7 @@ impl std::fmt::Debug for SignalInstanceConfig { .field("http_url", &"[REDACTED]") .field("account", &"[REDACTED]") .field("dm_allowed_users", &"[REDACTED]") + .field("authority", &self.authority) .field("group_ids", &self.group_ids) .field("group_allowed_users", &"[REDACTED]") .field("ignore_stories", &self.ignore_stories) @@ -3252,6 +3295,7 @@ impl std::fmt::Debug for SignalConfig { .field("account", &"[REDACTED]") .field("instances", &self.instances) .field("dm_allowed_users", &"[REDACTED]") + .field("authority", &self.authority) .field("group_ids", &self.group_ids) .field("group_allowed_users", &"[REDACTED]") .field("ignore_stories", &self.ignore_stories) @@ -3292,6 +3336,8 @@ pub struct MattermostConfig { pub team_id: Option, pub instances: Vec, pub dm_allowed_users: Vec, + /// Default authority list for slash commands on this adapter; binding-level lists take precedence. + pub authority: Vec, pub max_attachment_bytes: usize, } @@ -3304,6 +3350,7 @@ impl std::fmt::Debug for MattermostConfig { .field("team_id", &self.team_id) .field("instances", &self.instances) .field("dm_allowed_users", &self.dm_allowed_users) + .field("authority", &self.authority) .field("max_attachment_bytes", &self.max_attachment_bytes) .finish() } @@ -3317,6 +3364,8 @@ pub struct MattermostInstanceConfig { pub token: String, pub team_id: Option, pub dm_allowed_users: Vec, + /// Default authority list for slash commands on this adapter; binding-level lists take precedence. + pub authority: Vec, pub max_attachment_bytes: usize, } @@ -3360,6 +3409,7 @@ impl std::fmt::Debug for MattermostInstanceConfig { .field("token", &"[REDACTED]") .field("team_id", &self.team_id) .field("dm_allowed_users", &self.dm_allowed_users) + .field("authority", &self.authority) .field("max_attachment_bytes", &self.max_attachment_bytes) .finish() } diff --git a/src/config/watcher.rs b/src/config/watcher.rs index da732cdf2..36eec3c97 100644 --- a/src/config/watcher.rs +++ b/src/config/watcher.rs @@ -46,6 +46,7 @@ pub fn spawn_file_watcher( mattermost_permissions: Option>>, signal_permissions: Option>>, bindings: Arc>>, + authority_defaults: Arc>, messaging_manager: Option>, llm_manager: Arc, agent_links: Arc>>, @@ -220,6 +221,10 @@ pub fn spawn_file_watcher( bindings.store(Arc::new(config.bindings.clone())); tracing::info!("bindings reloaded ({} entries)", config.bindings.len()); + authority_defaults.store(Arc::new( + crate::commands::access::AdapterAuthorityDefaults::from_config(config), + )); + match crate::links::AgentLink::from_config(&config.links) { Ok(links) => { agent_links.store(Arc::new(links)); @@ -439,11 +444,10 @@ fn build_desired_configured_adapters( )) }); let fingerprint = format!( - "bot_token={}|app_token={}|dm={:?}|commands={:?}|permissions={}", + "bot_token={}|app_token={}|dm={:?}|permissions={}", secret_fingerprint(&slack_config.bot_token), secret_fingerprint(&slack_config.app_token), sorted_strings(slack_config.dm_allowed_users.clone()), - sorted_slack_commands(slack_config.commands.clone()), slack_permissions_fingerprint(&permissions_snapshot) ); let adapter = crate::messaging::slack::SlackAdapter::new( @@ -451,7 +455,6 @@ fn build_desired_configured_adapters( &slack_config.bot_token, &slack_config.app_token, permissions, - slack_config.commands.clone(), )?; desired.push(crate::messaging::ConfiguredAdapter::new( adapter, @@ -470,11 +473,10 @@ fn build_desired_configured_adapters( let permissions_snapshot = SlackPermissions::from_instance_config(instance, &config.bindings); let fingerprint = format!( - "bot_token={}|app_token={}|dm={:?}|commands={:?}|permissions={}", + "bot_token={}|app_token={}|dm={:?}|permissions={}", secret_fingerprint(&instance.bot_token), secret_fingerprint(&instance.app_token), sorted_strings(instance.dm_allowed_users.clone()), - sorted_slack_commands(instance.commands.clone()), slack_permissions_fingerprint(&permissions_snapshot) ); let adapter = crate::messaging::slack::SlackAdapter::new( @@ -482,7 +484,6 @@ fn build_desired_configured_adapters( &instance.bot_token, &instance.app_token, Arc::new(arc_swap::ArcSwap::from_pointee(permissions_snapshot)), - instance.commands.clone(), )?; desired.push(crate::messaging::ConfiguredAdapter::new( adapter, @@ -963,14 +964,3 @@ fn secret_fingerprint(value: &str) -> String { let digest = Sha256::digest(value.as_bytes()); hex::encode(&digest[..8]) } - -fn sorted_slack_commands( - mut commands: Vec, -) -> Vec<(String, String, Option)> { - let mut normalized = commands - .drain(..) - .map(|command| (command.command, command.agent_id, command.description)) - .collect::>(); - normalized.sort_by(|left, right| left.0.cmp(&right.0).then(left.1.cmp(&right.1))); - normalized -} diff --git a/src/conversation/channel_settings.rs b/src/conversation/channel_settings.rs index 23cadaa00..6d422c1b4 100644 --- a/src/conversation/channel_settings.rs +++ b/src/conversation/channel_settings.rs @@ -42,6 +42,35 @@ impl ChannelSettingsStore { })) } + /// Atomically set only the response mode. A single JSON-patch statement + /// avoids the read-modify-write window in which a concurrent whole-row + /// writer's changes could be overwritten with stale fields. + pub async fn set_response_mode( + &self, + agent_id: &str, + conversation_id: &str, + mode: crate::conversation::settings::ResponseMode, + ) -> crate::error::Result<()> { + let mode_str = mode.as_setting_str()?; + + sqlx::query( + "INSERT INTO channel_settings (agent_id, conversation_id, settings, updated_at) \ + VALUES (?, ?, json_set('{}', '$.response_mode', ?), CURRENT_TIMESTAMP) \ + ON CONFLICT (agent_id, conversation_id) \ + DO UPDATE SET settings = json_set(COALESCE(NULLIF(channel_settings.settings, ''), '{}'), '$.response_mode', ?), \ + updated_at = CURRENT_TIMESTAMP", + ) + .bind(agent_id) + .bind(conversation_id) + .bind(&mode_str) + .bind(&mode_str) + .execute(&self.pool) + .await + .map_err(|error| anyhow::anyhow!(error))?; + + Ok(()) + } + /// Insert or update settings for a channel. pub async fn upsert( &self, diff --git a/src/conversation/portal.rs b/src/conversation/portal.rs index 5bce8ed25..64e1a3857 100644 --- a/src/conversation/portal.rs +++ b/src/conversation/portal.rs @@ -187,6 +187,34 @@ impl PortalConversationStore { self.get(agent_id, session_id).await } + /// Atomically set only the response mode in the conversation's settings. + /// A single JSON-patch statement avoids the read-modify-write window in + /// which a concurrent whole-row writer's changes could be overwritten + /// with stale fields. Returns false when the conversation doesn't exist. + pub async fn set_response_mode( + &self, + agent_id: &str, + session_id: &str, + mode: crate::conversation::settings::ResponseMode, + ) -> crate::error::Result { + let mode_str = mode.as_setting_str()?; + + let result = sqlx::query( + "UPDATE portal_conversations \ + SET settings = json_set(COALESCE(NULLIF(settings, ''), '{}'), '$.response_mode', ?), \ + updated_at = CURRENT_TIMESTAMP \ + WHERE agent_id = ? AND id = ?", + ) + .bind(&mode_str) + .bind(agent_id) + .bind(session_id) + .execute(&self.pool) + .await + .map_err(|error| anyhow::anyhow!(error))?; + + Ok(result.rows_affected() > 0) + } + pub async fn delete(&self, agent_id: &str, session_id: &str) -> crate::error::Result { let mut tx = self .pool diff --git a/src/conversation/settings.rs b/src/conversation/settings.rs index ca15c78df..002df4f45 100644 --- a/src/conversation/settings.rs +++ b/src/conversation/settings.rs @@ -138,6 +138,37 @@ pub enum ResponseMode { MentionOnly, } +impl ResponseMode { + /// Atomic-cell encoding, for sharing the live mode across tasks. Values + /// must round-trip with [`from_u8`](Self::from_u8). + pub fn to_u8(self) -> u8 { + match self { + ResponseMode::Active => 0, + ResponseMode::Observe => 1, + ResponseMode::MentionOnly => 2, + } + } + + pub fn from_u8(value: u8) -> Self { + match value { + 1 => ResponseMode::Observe, + 2 => ResponseMode::MentionOnly, + _ => ResponseMode::Active, + } + } + + /// The serde string form ("active" / "observe" / "mention_only"), used + /// by the settings stores to JSON-patch the persisted `response_mode` + /// field without rewriting the whole settings row. + pub fn as_setting_str(self) -> anyhow::Result { + let value = serde_json::to_value(self)?; + value + .as_str() + .map(str::to_owned) + .ok_or_else(|| anyhow::anyhow!("response mode did not serialize to a string")) + } +} + /// Worker context settings control what context workers receive when spawned. #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] pub struct WorkerContextMode { diff --git a/src/main.rs b/src/main.rs index 638eb5533..ba5270bf1 100644 --- a/src/main.rs +++ b/src/main.rs @@ -175,6 +175,17 @@ fn forward_sse_event( }) .ok(); } + // Portal has no ephemeral surface; command replies render as + // ordinary messages instead of disappearing. + spacebot::OutboundResponse::Ephemeral { text, .. } => { + api_event_tx + .send(spacebot::api::ApiEvent::OutboundMessage { + agent_id: agent_id.to_string(), + channel_id: channel_id.to_string(), + text: text.clone(), + }) + .ok(); + } _ => {} } } @@ -871,6 +882,10 @@ async fn run( let bindings: Arc>> = Arc::new(ArcSwap::from_pointee(config.bindings.clone())); api_state.set_bindings(bindings.clone()).await; + let authority_defaults: Arc> = + Arc::new(ArcSwap::from_pointee( + spacebot::commands::access::AdapterAuthorityDefaults::from_config(&config), + )); let default_agent_id = config.default_agent_id().to_string(); // Set the config path on the API state for config.toml writes @@ -945,6 +960,7 @@ async fn run( mattermost_permissions, signal_permissions, bindings.clone(), + authority_defaults.clone(), Some(messaging_manager.clone()), llm_manager.clone(), agent_links.clone(), @@ -963,6 +979,7 @@ async fn run( None, // mattermost_permissions None, // signal_permissions bindings.clone(), + authority_defaults.clone(), None, llm_manager.clone(), agent_links.clone(), @@ -1307,7 +1324,16 @@ async fn run( tokio::select! { Some(mut message) = inbound_next, if agents_initialized => { let mut binding_settings: Option = None; + let mut binding_authority: Option> = None; let agent_id = if let Some(existing) = message.agent_id.as_ref() { + // Preassigned agent (portal sends set `agent_id` up + // front): the binding scan still runs so binding-level + // settings and authority apply to this scope. + let current_bindings = bindings.load(); + if let Some(binding) = spacebot::config::matched_binding(¤t_bindings, &message) { + binding_settings = binding.settings.clone(); + binding_authority = binding.authority.clone(); + } existing.clone() } else { let current_bindings = bindings.load(); @@ -1320,6 +1346,7 @@ async fn run( continue; }; binding_settings = matched_settings; + binding_authority = spacebot::config::matched_binding_authority(¤t_bindings, &message); message.agent_id = Some(resolved.clone()); resolved }; @@ -1327,6 +1354,40 @@ async fn run( let conversation_id = message.conversation_id.clone(); let channel_key = ActiveChannelKey::new(agent_id.to_string(), conversation_id.clone()); + // Slash-command dispatch, in the messaging layer before the + // channel queue. Control commands execute on the control + // plane without creating an inbound message, so they land + // even while the channel is mid-turn; Agent commands are + // rewritten to structured Command content and forwarded. + if let Some(agent) = agents.get(&agent_id) { + let turn_active = { + let channel_ref: spacebot::ChannelId = Arc::from(conversation_id.as_str()); + match agent.deps.process_control_registry.channel_handle(&channel_ref).await { + Some(handle) => handle.turn_active(), + None => false, + } + }; + let authority_snapshot = authority_defaults.load(); + let scope = spacebot::commands::dispatch::DispatchScope { + binding_authority: binding_authority.as_deref(), + adapter_defaults: &authority_snapshot, + binding_settings: binding_settings.as_ref(), + turn_active, + }; + match spacebot::commands::dispatch::dispatch_inbound( + &mut message, + scope, + &agent.deps, + &messaging_manager, + ) + .await + { + spacebot::commands::dispatch::Dispatch::Handled => continue, + spacebot::commands::dispatch::Dispatch::Forward + | spacebot::commands::dispatch::Dispatch::ForwardCommand => {} + } + } + // Find or create a channel for this conversation if !active_channels.contains_key(&channel_key) { let Some(agent) = agents.get(&agent_id) else { @@ -1744,6 +1805,7 @@ async fn run( new_mattermost_permissions, new_signal_permissions, bindings.clone(), + authority_defaults.clone(), Some(messaging_manager.clone()), new_llm_manager.clone(), agent_links.clone(), @@ -2396,7 +2458,6 @@ async fn initialize_agents( slack_permissions.clone().ok_or_else(|| { anyhow::anyhow!("slack permissions not initialized when slack is enabled") })?, - slack_config.commands.clone(), ) { Ok(adapter) => { new_messaging_manager.register(adapter).await; @@ -2431,7 +2492,6 @@ async fn initialize_agents( &instance.bot_token, &instance.app_token, perms, - instance.commands.clone(), ) { Ok(adapter) => { new_messaging_manager.register(adapter).await; diff --git a/src/messaging/discord.rs b/src/messaging/discord.rs index a8ced37c3..1b26716ea 100644 --- a/src/messaging/discord.rs +++ b/src/messaging/discord.rs @@ -9,17 +9,30 @@ use anyhow::Context as _; use arc_swap::ArcSwap; use async_trait::async_trait; use serenity::all::{ - ButtonStyle, ChannelId, ChannelType, Context, CreateActionRow, CreateAttachment, CreateButton, - CreateEmbed, CreateEmbedAuthor, CreateEmbedFooter, CreateInteractionResponse, - CreateInteractionResponseMessage, CreateMessage, CreatePoll, CreatePollAnswer, - CreateSelectMenu, CreateSelectMenuKind, CreateSelectMenuOption, CreateThread, EditMessage, - EventHandler, GatewayIntents, GetMessages, Http, Interaction, Message, MessageId, ReactionType, - Ready, ShardManager, Timestamp, User, UserId, + ButtonStyle, ChannelId, ChannelType, Command as ApplicationCommand, CommandDataOptionValue, + CommandInteraction, CommandOptionType, Context, CreateActionRow, CreateAttachment, + CreateButton, CreateCommand, CreateCommandOption, CreateEmbed, CreateEmbedAuthor, + CreateEmbedFooter, CreateInteractionResponse, CreateInteractionResponseMessage, CreateMessage, + CreatePoll, CreatePollAnswer, CreateSelectMenu, CreateSelectMenuKind, CreateSelectMenuOption, + CreateThread, EditInteractionResponse, EditMessage, EventHandler, GatewayIntents, GetMessages, + GuildId, Http, Interaction, Message, MessageId, ReactionType, Ready, ShardManager, Timestamp, + User, UserId, }; use std::collections::HashMap; use std::sync::Arc; +use std::time::Instant; use tokio::sync::{RwLock, mpsc}; +/// Discord interaction tokens are valid for 15 minutes; stop trying to +/// resolve deferred responses a minute early. +const INTERACTION_TOKEN_TTL: std::time::Duration = std::time::Duration::from_secs(14 * 60); + +/// A deferred interaction awaiting its real response. +struct InteractionToken { + token: String, + created: Instant, +} + /// Discord adapter state. pub struct DiscordAdapter { runtime_key: String, @@ -32,6 +45,12 @@ pub struct DiscordAdapter { /// Typing handles per message. Typing stops when the handle is dropped. typing_tasks: Arc>>, shard_manager: Arc>>>, + /// Deferred slash-command interactions keyed by InboundMessage.id, so + /// replies can resolve the deferral within the token window. + interaction_tokens: Arc>>, + /// Guards against concurrent application-command syncs when `ready` + /// re-fires on reconnect. + command_sync_active: Arc, } impl DiscordAdapter { @@ -49,6 +68,30 @@ impl DiscordAdapter { active_messages: Arc::new(RwLock::new(HashMap::new())), typing_tasks: Arc::new(RwLock::new(HashMap::new())), shard_manager: Arc::new(RwLock::new(None)), + interaction_tokens: Arc::new(RwLock::new(HashMap::new())), + command_sync_active: Arc::new(std::sync::atomic::AtomicBool::new(false)), + } + } + + /// Take the deferred-interaction token for a message, if still within + /// the token window. Expired entries are pruned on the way. + async fn take_interaction_token(&self, message_id: &str) -> Option { + let mut tokens = self.interaction_tokens.write().await; + tokens.retain(|_, entry| entry.created.elapsed() < INTERACTION_TOKEN_TTL); + tokens.remove(message_id).map(|entry| entry.token) + } + + /// Clear a deferred interaction whose real response is going out through + /// a surface that can't resolve it (rich messages, streaming). Deleting + /// the deferral removes the dangling "thinking" state. + async fn clear_deferred_interaction(&self, http: &Arc, message_id: &str) { + if let Some(token) = self.take_interaction_token(message_id).await { + let http = http.clone(); + tokio::spawn(async move { + if let Err(error) = http.delete_original_interaction_response(&token).await { + tracing::debug!(%error, "failed to delete deferred interaction response"); + } + }); } } @@ -113,6 +156,8 @@ impl Messaging for DiscordAdapter { permissions: self.permissions.clone(), http_slot: self.http.clone(), bot_user_id_slot: self.bot_user_id.clone(), + interaction_tokens: self.interaction_tokens.clone(), + command_sync_active: self.command_sync_active.clone(), }; let intents = GatewayIntents::GUILD_MESSAGES @@ -150,8 +195,29 @@ impl Messaging for DiscordAdapter { OutboundResponse::Text(text) => { self.stop_typing(message).await; let reply_to = Self::extract_reply_message_id(message); + let deferred = self.take_interaction_token(&message.id).await; for (index, chunk) in split_message(&text, 2000).into_iter().enumerate() { + // The first chunk resolves a deferred slash-command + // interaction in place; overflow chunks follow as + // ordinary messages. + if index == 0 + && let Some(token) = deferred.as_deref() + { + let edit = EditInteractionResponse::new().content(chunk.as_str()); + match http + .edit_original_interaction_response(token, &edit, Vec::new()) + .await + { + Ok(_) => continue, + Err(error) => { + tracing::warn!( + %error, + "failed to resolve deferred interaction, falling back to channel send" + ); + } + } + } let mut builder = CreateMessage::new().content(chunk); if index == 0 && let Some(reply_message_id) = reply_to @@ -172,6 +238,7 @@ impl Messaging for DiscordAdapter { .. } => { self.stop_typing(message).await; + self.clear_deferred_interaction(&http, &message.id).await; let reply_to = Self::extract_reply_message_id(message); let parts = prepare_rich_message_parts(text, &cards, &interactive_elements, poll.as_ref()); @@ -316,6 +383,7 @@ impl Messaging for DiscordAdapter { } OutboundResponse::StreamStart => { self.stop_typing(message).await; + self.clear_deferred_interaction(&http, &message.id).await; let placeholder = channel_id .say(&*http, "\u{200B}") @@ -351,14 +419,23 @@ impl Messaging for DiscordAdapter { // Slack-specific variants — graceful fallbacks for Discord OutboundResponse::RemoveReaction(_) => {} // no-op OutboundResponse::Ephemeral { text, .. } => { - // Discord has no ephemeral equivalent here; send as regular text - if let Ok(channel_id) = self.extract_channel_id(message) { - let http = self.get_http().await?; - channel_id - .say(&*http, &text) + // A deferred slash command resolves ephemerally in place; + // outside an interaction Discord has no ephemeral surface, + // so degrade to a regular message. + if let Some(token) = self.take_interaction_token(&message.id).await { + let edit = EditInteractionResponse::new().content(&text); + if http + .edit_original_interaction_response(&token, &edit, Vec::new()) .await - .context("failed to send ephemeral fallback on discord")?; + .is_ok() + { + return Ok(()); + } } + channel_id + .say(&*http, &text) + .await + .context("failed to send ephemeral fallback on discord")?; } OutboundResponse::ScheduledMessage { text, .. } => { // Discord has no native scheduled messages — send immediately @@ -589,6 +666,128 @@ struct Handler { permissions: Arc>, http_slot: Arc>>>, bot_user_id_slot: Arc>>, + interaction_tokens: Arc>>, + command_sync_active: Arc, +} + +impl Handler { + /// Handle a native slash-command interaction: admission checks, defer + /// (ephemeral for Control commands), then inject as structured + /// `MessageContent::Command` for the router's shared dispatch path. + async fn handle_command_interaction(&self, ctx: Context, command: CommandInteraction) { + let permissions = self.permissions.load(); + let user = &command.user; + + if command.guild_id.is_none() + && (permissions.dm_allowed_users.is_empty() + || !permissions.dm_allowed_users.contains(&user.id.get())) + { + return; + } + if let Some(filter) = &permissions.guild_filter + && let Some(guild_id) = command.guild_id + && !filter.contains(&guild_id.get()) + { + return; + } + + // Control replies resolve the deferral ephemerally; Agent commands + // defer publicly and the real response arrives through the normal + // message path. + let is_control = crate::commands::REGISTRY + .resolve(&command.data.name) + .is_some_and(|def| matches!(def.handler, crate::commands::CommandHandler::Control(_))); + let defer = if is_control { + CreateInteractionResponse::Defer( + CreateInteractionResponseMessage::new().ephemeral(true), + ) + } else { + CreateInteractionResponse::Defer(CreateInteractionResponseMessage::new()) + }; + if let Err(error) = command.create_response(&ctx.http, defer).await { + tracing::warn!(%error, command = %command.data.name, "failed to defer slash command"); + return; + } + + let interaction_id = command.id.to_string(); + { + let mut tokens = self.interaction_tokens.write().await; + tokens.retain(|_, entry| entry.created.elapsed() < INTERACTION_TOKEN_TTL); + tokens.insert( + interaction_id.clone(), + InteractionToken { + token: command.token.clone(), + created: Instant::now(), + }, + ); + } + + let args = command + .data + .options + .first() + .and_then(|option| match &option.value { + CommandDataOptionValue::String(value) => Some(value.clone()), + _ => None, + }) + .unwrap_or_default(); + + let base_conversation_id = match command.guild_id { + Some(guild_id) => format!("discord:{}:{}", guild_id, command.channel_id), + None => format!("discord:dm:{}", user.id), + }; + let conversation_id = + apply_runtime_adapter_to_conversation_id(&self.runtime_key, base_conversation_id); + + let mut metadata = HashMap::new(); + metadata.insert( + "discord_channel_id".into(), + serde_json::Value::Number(command.channel_id.get().into()), + ); + if let Some(guild_id) = command.guild_id { + metadata.insert( + "discord_guild_id".into(), + serde_json::Value::Number(guild_id.get().into()), + ); + } + // A command invocation addresses the bot directly. + metadata.insert("discord_mentioned_bot".into(), true.into()); + metadata.insert("discord_reply_to_bot".into(), false.into()); + metadata.insert("discord_mentions_or_replies_to_bot".into(), true.into()); + + let formatted_author = format!("{} (<@{}>)", user.name, user.id); + metadata.insert( + "discord_user_id".into(), + serde_json::Value::Number(user.id.get().into()), + ); + metadata.insert( + "sender_display_name".into(), + serde_json::Value::String(formatted_author.clone()), + ); + + let inbound = InboundMessage { + id: interaction_id, + source: "discord".into(), + adapter: Some(self.runtime_key.clone()), + conversation_id, + sender_id: user.id.to_string(), + agent_id: None, + content: MessageContent::Command { + name: command.data.name.clone(), + args, + }, + timestamp: chrono::Utc::now(), + metadata, + formatted_author: Some(formatted_author), + }; + + if let Err(error) = self.inbound_tx.send(inbound).await { + tracing::warn!( + %error, + "failed to send inbound slash command from Discord (receiver dropped)" + ); + } + } } #[async_trait] @@ -599,6 +798,22 @@ impl EventHandler for Handler { *self.http_slot.write().await = Some(ctx.http.clone()); *self.bot_user_id_slot.write().await = Some(ready.user.id); tracing::info!(guild_count = ready.guilds.len(), "discord guilds available"); + + // Register application commands from the registry. `ready` re-fires + // on reconnect; the sync is diff-only so re-runs are no-ops, and the + // guard just prevents overlapping syncs. + if !self + .command_sync_active + .swap(true, std::sync::atomic::Ordering::AcqRel) + { + let http = ctx.http.clone(); + let permissions = self.permissions.clone(); + let guard = self.command_sync_active.clone(); + tokio::spawn(async move { + sync_application_commands(&http, &permissions).await; + guard.store(false, std::sync::atomic::Ordering::Release); + }); + } } async fn message(&self, ctx: Context, message: Message) { @@ -678,7 +893,11 @@ impl EventHandler for Handler { async fn interaction_create(&self, ctx: Context, interaction: Interaction) { let component = match interaction { Interaction::Component(c) => c, - _ => return, // Only handle component interactions + Interaction::Command(command) => { + self.handle_command_interaction(ctx, command).await; + return; + } + _ => return, }; // Acknowledge the interaction immediately to prevent "This interaction failed" in the UI. @@ -786,6 +1005,302 @@ impl EventHandler for Handler { } } +/// Discord application-command field limits (CHAT_INPUT). +const DISCORD_NAME_MAX: usize = 32; +const DISCORD_DESCRIPTION_MAX: usize = 100; +const DISCORD_CHOICE_CAP: usize = 25; +const DISCORD_CHOICE_VALUE_MAX: usize = 100; + +/// Why a spec can't be expressed as a Discord application command, if it +/// can't. One invalid entry in a bulk `set_commands` call fails the whole +/// batch, so specs are validated up front and invalid ones skipped. +fn discord_spec_violation(spec: &crate::commands::native::NativeCommandSpec) -> Option { + use crate::commands::native::NativeArg; + + // Registry names are ASCII; Discord additionally allows lowercase + // unicode letters, which no command uses. + let name_valid = !spec.name.is_empty() + && spec.name.chars().count() <= DISCORD_NAME_MAX + && spec + .name + .chars() + .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-' || c == '_'); + if !name_valid { + return Some(format!( + "name must be 1-{DISCORD_NAME_MAX} lowercase [a-z0-9_-] characters" + )); + } + if let Some(NativeArg::Choice { options }) = &spec.arg { + if options.len() > DISCORD_CHOICE_CAP { + return Some(format!( + "{} choices exceed discord's cap of {DISCORD_CHOICE_CAP}", + options.len() + )); + } + // A truncated choice value would dispatch different args than the + // user picked, so over-long choices invalidate the command instead. + if options + .iter() + .any(|option| option.is_empty() || option.chars().count() > DISCORD_CHOICE_VALUE_MAX) + { + return Some(format!( + "choice values must be 1-{DISCORD_CHOICE_VALUE_MAX} characters" + )); + } + } + None +} + +/// Clamp a description to Discord's 1-100 character range, falling back to +/// the command name when empty. +fn normalize_discord_description(description: &str, command_name: &str) -> String { + let normalized: String = description.chars().take(DISCORD_DESCRIPTION_MAX).collect(); + if normalized.is_empty() { + format!("/{command_name}") + } else { + normalized + } +} + +/// Build serenity command definitions from the registry's native specs. +/// Returned specs are normalized to Discord limits and match the built +/// commands one-to-one, so the diff comparison sees exactly what was +/// registered. +fn build_discord_create_commands() -> ( + Vec, + Vec, + usize, +) { + use crate::commands::native::discord_commands; + + let (specs, dropped) = discord_commands(); + let specs: Vec<_> = specs + .into_iter() + .filter(|spec| match discord_spec_violation(spec) { + Some(reason) => { + tracing::warn!( + command = %spec.name, + reason, + "command cannot be registered as a discord application command, skipping" + ); + false + } + None => true, + }) + .map(|mut spec| { + spec.description = normalize_discord_description(&spec.description, &spec.name); + spec + }) + .collect(); + let commands = specs + .iter() + .map(|spec| { + let mut command = CreateCommand::new(&spec.name).description(&spec.description); + if let Some((name, description, required, choices)) = expected_option(spec) { + let mut option = + CreateCommandOption::new(CommandOptionType::String, name, description) + .required(required); + for choice in choices { + option = option.add_string_choice(&choice, &choice); + } + command = command.add_option(option); + } + command + }) + .collect(); + (commands, specs, dropped) +} + +/// The option a spec maps to: (name, description, required, choices). +/// Shared between registration and the diff comparison so they can't drift. +fn expected_option( + spec: &crate::commands::native::NativeCommandSpec, +) -> Option<(String, String, bool, Vec)> { + use crate::commands::native::NativeArg; + match &spec.arg { + None => None, + Some(NativeArg::Text { hint, required }) => Some(( + "input".to_string(), + { + let description: String = hint.chars().take(DISCORD_DESCRIPTION_MAX).collect(); + if description.is_empty() { + "input".to_string() + } else { + description + } + }, + *required, + Vec::new(), + )), + Some(NativeArg::Choice { options }) => Some(( + "value".to_string(), + { + let description: String = format!("one of: {}", options.join(", ")) + .chars() + .take(DISCORD_DESCRIPTION_MAX) + .collect(); + description + }, + false, + options.clone(), + )), + } +} + +/// Whether the live command set already matches the registry specs. +/// Comparing name, description, and option shape keeps re-registration a +/// no-op on every reconnect. +fn command_set_matches( + existing: &[ApplicationCommand], + specs: &[crate::commands::native::NativeCommandSpec], +) -> bool { + type Normalized = (String, String, Option<(String, String, bool, Vec)>); + + let mut live: Vec = existing + .iter() + .map(|command| { + let option = command.options.first().map(|option| { + ( + option.name.clone(), + option.description.clone(), + option.required, + option + .choices + .iter() + .map(|choice| choice.name.clone()) + .collect::>(), + ) + }); + (command.name.clone(), command.description.clone(), option) + }) + .collect(); + let mut wanted: Vec = specs + .iter() + .map(|spec| { + ( + spec.name.clone(), + spec.description.clone(), + expected_option(spec), + ) + }) + .collect(); + live.sort(); + wanted.sort(); + live == wanted +} + +/// Diff-only application-command sync. Guild-scoped when bindings declare +/// guilds (instant propagation, scoped to served guilds), global otherwise. +async fn sync_application_commands( + http: &Arc, + permissions: &Arc>, +) { + let (commands, specs, dropped) = build_discord_create_commands(); + if dropped > 0 { + tracing::warn!( + dropped, + cap = crate::commands::native::DISCORD_COMMAND_CAP, + "discord command cap exceeded; trailing registry commands were not registered" + ); + } + + let guild_filter = permissions.load().guild_filter.clone(); + match guild_filter { + Some(guild_ids) => { + // Every configured guild is synced, not just the ones present + // in a READY payload — a served guild missing from one gateway + // session would otherwise get no commands until a reconnect + // happens to include it. Per-guild failures (the bot not being + // a member yet, transient API errors) are logged and the loop + // continues with the remaining guilds. The filter can repeat a + // guild (one binding per channel set), so duplicates are synced + // once. + let mut all_guilds_synced = true; + let mut seen_guilds = std::collections::HashSet::new(); + for guild_id in guild_ids + .iter() + .copied() + .filter(|guild_id| seen_guilds.insert(*guild_id)) + .map(GuildId::new) + { + let in_sync = match guild_id.get_commands(http).await { + Ok(existing) => command_set_matches(&existing, &specs), + Err(error) => { + tracing::warn!(%error, guild_id = %guild_id, "failed to fetch guild commands"); + false + } + }; + if in_sync { + tracing::debug!(guild_id = %guild_id, "discord commands already in sync"); + continue; + } + match guild_id.set_commands(http, commands.clone()).await { + Ok(registered) => { + tracing::info!( + guild_id = %guild_id, + count = registered.len(), + "discord guild commands registered" + ); + } + Err(error) => { + all_guilds_synced = false; + tracing::warn!(%error, guild_id = %guild_id, "failed to register guild commands"); + } + } + } + + // Stale globals are cleared only after every configured guild is + // confirmed in sync — clearing first would leave a guild whose + // registration then failed with no commands at all until a later + // reconnect. A guild that failed keeps the globals as a fallback; + // the next `ready` re-sync retries both. + if !all_guilds_synced { + tracing::warn!( + "leaving global discord commands in place until every configured guild syncs" + ); + return; + } + match ApplicationCommand::get_global_commands(http).await { + Ok(global) if !global.is_empty() => { + if let Err(error) = + ApplicationCommand::set_global_commands(http, Vec::new()).await + { + tracing::warn!(%error, "failed to clear stale global discord commands"); + } + } + Ok(_) => {} + Err(error) => { + tracing::warn!(%error, "failed to fetch global discord commands"); + } + } + } + None => { + match ApplicationCommand::get_global_commands(http).await { + Ok(existing) if command_set_matches(&existing, &specs) => { + tracing::debug!("global discord commands already in sync"); + return; + } + Ok(_) => {} + Err(error) => { + tracing::warn!(%error, "failed to fetch global discord commands"); + return; + } + } + match ApplicationCommand::set_global_commands(http, commands).await { + Ok(registered) => { + tracing::info!( + count = registered.len(), + "global discord commands registered" + ); + } + Err(error) => { + tracing::warn!(%error, "failed to register global discord commands"); + } + } + } + } +} + fn is_mention_or_reply_to_bot(message: &Message, bot_user_id: Option) -> bool { is_mention_to_bot(message, bot_user_id) || is_reply_to_bot(message, bot_user_id) } @@ -1246,6 +1761,97 @@ mod tests { use super::*; use crate::{Button, ButtonStyle, Card, CardField, InteractiveElements, Poll}; + #[test] + fn discord_specs_satisfy_platform_limits() { + let (commands, specs, _) = build_discord_create_commands(); + assert_eq!(commands.len(), specs.len()); + for spec in &specs { + assert!( + discord_spec_violation(spec).is_none(), + "spec /{} violates discord limits", + spec.name + ); + assert!( + (1..=DISCORD_DESCRIPTION_MAX).contains(&spec.description.chars().count()), + "discord description length for /{}", + spec.name + ); + if let Some((_, description, _, choices)) = expected_option(spec) { + assert!( + (1..=DISCORD_DESCRIPTION_MAX).contains(&description.chars().count()), + "discord option description length for /{}", + spec.name + ); + assert!( + choices.len() <= DISCORD_CHOICE_CAP, + "discord choice cap for /{}", + spec.name + ); + } + } + } + + #[test] + fn discord_validation_drops_no_registry_command() { + // Every command the registry exposes on Discord must survive + // validation — a violation here means a registry entry silently + // disappears from Discord instead of failing the build. + let (specs, _) = crate::commands::native::discord_commands(); + let (_, validated, _) = build_discord_create_commands(); + let expected: Vec<&str> = specs.iter().map(|spec| spec.name.as_str()).collect(); + let actual: Vec<&str> = validated.iter().map(|spec| spec.name.as_str()).collect(); + assert_eq!(actual, expected); + } + + #[test] + fn discord_spec_violation_flags_invalid_specs() { + use crate::commands::native::{NativeArg, NativeCommandSpec}; + + let valid = NativeCommandSpec { + name: "mention-only".into(), + description: "only respond when mentioned".into(), + arg: None, + }; + assert!(discord_spec_violation(&valid).is_none()); + + let uppercase = NativeCommandSpec { + name: "Status".into(), + ..valid.clone() + }; + assert!(discord_spec_violation(&uppercase).is_some()); + + let too_long = NativeCommandSpec { + name: "a".repeat(DISCORD_NAME_MAX + 1), + ..valid.clone() + }; + assert!(discord_spec_violation(&too_long).is_some()); + + let too_many_choices = NativeCommandSpec { + arg: Some(NativeArg::Choice { + options: (0..DISCORD_CHOICE_CAP + 1) + .map(|i| format!("option-{i}")) + .collect(), + }), + ..valid.clone() + }; + assert!(discord_spec_violation(&too_many_choices).is_some()); + + let oversized_choice = NativeCommandSpec { + arg: Some(NativeArg::Choice { + options: vec!["x".repeat(DISCORD_CHOICE_VALUE_MAX + 1)], + }), + ..valid + }; + assert!(discord_spec_violation(&oversized_choice).is_some()); + } + + #[test] + fn discord_description_normalization_clamps_and_falls_back() { + let clamped = normalize_discord_description(&"d".repeat(300), "status"); + assert_eq!(clamped.chars().count(), DISCORD_DESCRIPTION_MAX); + assert_eq!(normalize_discord_description("", "status"), "/status"); + } + #[test] fn test_build_embed_limits() { let mut card = Card::default(); diff --git a/src/messaging/email.rs b/src/messaging/email.rs index 884dab226..de732f6a7 100644 --- a/src/messaging/email.rs +++ b/src/messaging/email.rs @@ -182,6 +182,7 @@ impl EmailAdapter { // Build a temporary EmailConfig to reuse build_smtp_transport and shared logic. let email_config = EmailConfig { enabled: config.enabled, + authority: config.authority.clone(), imap_host: config.imap_host.clone(), imap_port: config.imap_port, imap_username: config.imap_username.clone(), diff --git a/src/messaging/slack.rs b/src/messaging/slack.rs index b4e1c0c60..eb21be684 100644 --- a/src/messaging/slack.rs +++ b/src/messaging/slack.rs @@ -21,7 +21,7 @@ //! - Typing indicator via `assistant.threads.setStatus` //! - DM broadcast via `conversations.open` -use crate::config::{SlackCommandConfig, SlackPermissions}; +use crate::config::SlackPermissions; use crate::messaging::apply_runtime_adapter_to_conversation_id; use crate::messaging::traits::{HistoryMessage, InboundStream, Messaging}; use crate::{InboundMessage, MessageContent, OutboundResponse, StatusUpdate}; @@ -41,9 +41,6 @@ struct SlackAdapterState { permissions: Arc>, bot_token: String, bot_user_id: String, - /// Maps slash command string (e.g. `"/ask"`) → agent_id. - /// Built once at start() from the config; read-only afterwards. - commands: Arc>, /// Cache of resolved user identities to avoid repeated `users.info` API calls. user_identity_cache: Arc>>, /// Cache of resolved channel names to avoid repeated `conversations.info` API calls. @@ -71,8 +68,6 @@ pub struct SlackAdapter { /// Maps InboundMessage.id → Slack ts for streaming edits. active_messages: Arc>>, shutdown_tx: Arc>>>, - /// Slash command routing: command string → agent_id. - commands: Arc>, } impl SlackAdapter { @@ -81,7 +76,6 @@ impl SlackAdapter { bot_token: impl Into, app_token: impl Into, permissions: Arc>, - commands: Vec, ) -> anyhow::Result { let runtime_key = runtime_key.into(); let bot_token = bot_token.into(); @@ -89,10 +83,6 @@ impl SlackAdapter { SlackClientHyperConnector::new().context("failed to create slack HTTP connector")?, )); let token = SlackApiToken::new(SlackApiTokenValue(bot_token.clone())); - let commands_map: HashMap = commands - .into_iter() - .map(|c| (c.command, c.agent_id)) - .collect(); Ok(Self { runtime_key, bot_token, @@ -102,7 +92,6 @@ impl SlackAdapter { token, active_messages: Arc::new(RwLock::new(HashMap::new())), shutdown_tx: Arc::new(RwLock::new(None)), - commands: Arc::new(commands_map), }) } @@ -425,6 +414,22 @@ fn slack_error_handler( HttpStatusCode::OK } +/// Resolve a `/spacebot` subcommand token. Resolution mirrors the native +/// listing: a command that isn't available on the Slack surface is treated +/// as unknown, so the umbrella command can't reach commands its own listing +/// hides. An empty token answers with the usage listing. +fn resolve_slack_subcommand( + registry: &crate::commands::CommandRegistry, + subcommand: &str, +) -> Option<&'static crate::commands::CommandDef> { + if subcommand.is_empty() { + return None; + } + registry + .resolve(subcommand) + .filter(|def| def.availability.on(crate::commands::Surface::Slack)) +} + /// Handle Slack slash command events (e.g. `/ask What is the weather?`). /// /// Slack requires an acknowledgement within 3 seconds. This handler acks @@ -495,22 +500,33 @@ async fn handle_command_event( } } - if !adapter_state.commands.contains_key(&command_str) { - tracing::warn!( - command = %command_str, - user_id = %user_id, - "slash command not configured — ignoring" - ); + // One `/spacebot` umbrella command covers the whole registry: + // `/spacebot status`, `/spacebot quiet`, ... . An empty or unknown + // subcommand answers with the generated listing instead of dispatching. + let (subcommand, args) = match text.trim().split_once(char::is_whitespace) { + Some((subcommand, rest)) => (subcommand.to_string(), rest.trim().to_string()), + None => (text.trim().to_string(), String::new()), + }; + let resolved = resolve_slack_subcommand(&crate::commands::REGISTRY, &subcommand); + let Some(def) = resolved else { + let listing = crate::commands::native::slack_subcommands() + .into_iter() + .map(|spec| match spec.arg { + Some(_) => format!("`{} {} `", command_str, spec.name), + None => format!("`{} {}`", command_str, spec.name), + }) + .collect::>() + .join(", "); + let heading = if subcommand.is_empty() { + "usage:".to_string() + } else { + format!("`{subcommand}` is not a spacebot command. usage:") + }; return Ok(SlackCommandEventResponse { - content: SlackMessageContent::new().with_text(format!( - "`{}` is not configured on this Spacebot instance.", - command_str - )), + content: SlackMessageContent::new().with_text(format!("{heading} {listing}")), response_type: Some(SlackMessageResponseType::Ephemeral), }); - } - - let agent_id = adapter_state.commands[&command_str].clone(); + }; let base_conversation_id = format!("slack:{}:{}", team_id, channel_id); let conversation_id = @@ -541,14 +557,11 @@ async fn handle_command_event( "slack_user_mention".into(), serde_json::Value::String(format!("<@{}>", user_id)), ); - // Embed the agent_id hint so the router can honour command-specific routing - // without requiring a separate binding entry per command. - metadata.insert( - "slack_command_agent_id".into(), - serde_json::Value::String(agent_id), - ); - let content = MessageContent::Text(format!("{} {}", command_str, text).trim().to_string()); + let content = MessageContent::Command { + name: def.name.to_string(), + args, + }; let inbound = InboundMessage { id: msg_id, @@ -780,7 +793,6 @@ impl Messaging for SlackAdapter { permissions: self.permissions.clone(), bot_token: self.bot_token.clone(), bot_user_id, - commands: self.commands.clone(), user_identity_cache: Arc::new(RwLock::new(HashMap::new())), channel_name_cache: Arc::new(RwLock::new(HashMap::new())), }); @@ -790,14 +802,6 @@ impl Messaging for SlackAdapter { .with_command_events(handle_command_event) .with_interaction_events(handle_interaction_event); - // The socket mode listener needs its own client instance — it manages - // a persistent WebSocket connection internally and owns that client for - // the lifetime of the connection. The shared `self.client` is for REST calls. - let _listener_client = Arc::new(SlackClient::new( - SlackClientHyperConnector::new() - .context("failed to create slack socket mode connector")?, - )); - // The socket mode listener needs its own client — it owns a persistent // WebSocket connection. The shared self.client is for REST calls only. let listener_client = Arc::new(SlackClient::new( @@ -1733,6 +1737,53 @@ fn resolve_slack_user_identity(user: &SlackUser, user_id: &str) -> SlackUserIden mod tests { use super::*; + #[test] + fn slack_subcommands_resolve_only_when_available_on_slack() { + use crate::commands::{ + ArgSpec, BusyPolicy, CommandAccess, CommandAvailability, CommandCategory, CommandDef, + CommandHandler, CommandRegistry, ControlAction, + }; + + static DEFS: &[CommandDef] = &[ + CommandDef { + name: "everywhere", + description: "available on all surfaces", + category: CommandCategory::Info, + aliases: &[], + args: ArgSpec::None, + handler: CommandHandler::Control(ControlAction::Help), + access: CommandAccess::Everyone, + busy: BusyPolicy::Queue, + availability: CommandAvailability::ALL, + }, + CommandDef { + name: "not-on-slack", + description: "hidden from the slack listing", + category: CommandCategory::Info, + aliases: &[], + args: ArgSpec::None, + handler: CommandHandler::Control(ControlAction::Help), + access: CommandAccess::Everyone, + busy: BusyPolicy::Queue, + availability: CommandAvailability { + portal: true, + discord: true, + slack: false, + telegram: true, + text_adapters: true, + }, + }, + ]; + let registry = CommandRegistry::new(DEFS); + + assert!(resolve_slack_subcommand(®istry, "everywhere").is_some()); + assert!( + resolve_slack_subcommand(®istry, "not-on-slack").is_none(), + "a command unavailable on slack must fall through to the usage listing" + ); + assert!(resolve_slack_subcommand(®istry, "").is_none()); + } + #[test] fn sanitize_reaction_name_unicode_emoji_with_shortcode() { // gemoji maps 👍 to "+1" — verify we get the shortcode, not the unicode back diff --git a/src/messaging/telegram.rs b/src/messaging/telegram.rs index 747d40c45..f3e1e866b 100644 --- a/src/messaging/telegram.rs +++ b/src/messaging/telegram.rs @@ -11,8 +11,8 @@ use regex::Regex; use teloxide::payloads::setters::*; use teloxide::requests::{Request, Requester}; use teloxide::types::{ - ChatAction, ChatId, FileId, InputFile, InputPollOption, MediaKind, MessageId, MessageKind, - ParseMode, ReactionType, ReplyParameters, UpdateKind, UserId, + BotCommand, ChatAction, ChatId, FileId, InputFile, InputPollOption, MediaKind, MessageId, + MessageKind, ParseMode, ReactionType, ReplyParameters, UpdateKind, UserId, }; use teloxide::{ApiError, Bot, RequestError}; @@ -38,6 +38,36 @@ pub struct TelegramAdapter { typing_tasks: Arc>>>, /// Shutdown signal for the polling loop. shutdown_tx: Arc>>>, + /// Whether the registry-generated command menu has been registered via + /// setMyCommands. `start()` re-runs on supervisor restarts; the menu + /// only needs to land once per adapter instance. + commands_registered: Arc, +} + +impl TelegramAdapter { + /// Register the registry-generated command menu with Telegram. Runs at + /// most once per adapter instance; a failure clears the guard so the + /// next supervisor restart retries. + async fn register_command_menu(&self) { + use std::sync::atomic::Ordering; + if self.commands_registered.swap(true, Ordering::AcqRel) { + return; + } + let menu: Vec = crate::commands::native::telegram_menu() + .into_iter() + .map(|entry| BotCommand::new(entry.command, entry.description)) + .collect(); + let count = menu.len(); + match self.bot.set_my_commands(menu).send().await { + Ok(_) => { + tracing::info!(count, "telegram command menu registered"); + } + Err(error) => { + tracing::warn!(%error, "failed to register telegram command menu"); + self.commands_registered.store(false, Ordering::Release); + } + } + } } /// Tracks an in-progress streaming message edit. @@ -73,6 +103,7 @@ impl TelegramAdapter { bot_username: Arc::new(RwLock::new(None)), active_messages: Arc::new(RwLock::new(HashMap::new())), typing_tasks: Arc::new(RwLock::new(HashMap::new())), + commands_registered: Arc::new(std::sync::atomic::AtomicBool::new(false)), shutdown_tx: Arc::new(RwLock::new(None)), } } @@ -129,6 +160,8 @@ impl Messaging for TelegramAdapter { "telegram connected" ); + self.register_command_menu().await; + let bot = self.bot.clone(); let runtime_key = self.runtime_key.clone(); let permissions = self.permissions.clone(); diff --git a/tests/context_dump.rs b/tests/context_dump.rs index 877711b07..2cbaa601c 100644 --- a/tests/context_dump.rs +++ b/tests/context_dump.rs @@ -253,6 +253,8 @@ async fn dump_channel_context() { let state = spacebot::agent::channel::ChannelState { channel_id, kind: spacebot::agent::channel::ChannelKind::User, + turn_active: Arc::new(std::sync::atomic::AtomicBool::new(false)), + response_mode: Arc::new(std::sync::atomic::AtomicU8::new(0)), history: Arc::new(tokio::sync::RwLock::new(Vec::new())), active_branches: Arc::new(tokio::sync::RwLock::new(std::collections::HashMap::new())), worker_handles: Arc::new(tokio::sync::RwLock::new(std::collections::HashMap::new())), @@ -513,6 +515,8 @@ async fn dump_all_contexts() { let state = spacebot::agent::channel::ChannelState { channel_id, kind: spacebot::agent::channel::ChannelKind::User, + turn_active: Arc::new(std::sync::atomic::AtomicBool::new(false)), + response_mode: Arc::new(std::sync::atomic::AtomicU8::new(0)), history: Arc::new(tokio::sync::RwLock::new(Vec::new())), active_branches: Arc::new(tokio::sync::RwLock::new(std::collections::HashMap::new())), worker_handles: Arc::new(tokio::sync::RwLock::new(std::collections::HashMap::new())),