From 6ce4f52972d92c3549c40efcd898e818225c3a14 Mon Sep 17 00:00:00 2001 From: YellowSnnowmann <167776381+YellowSnnowmann@users.noreply.github.com> Date: Tue, 23 Jun 2026 00:47:21 +0530 Subject: [PATCH] fix(channels): repair Discord & Telegram messaging end-to-end (#3712, #3763) (#3794) Co-authored-by: Claude Opus 4.8 Co-authored-by: Steven Enamakel --- app/src/components/channels/DiscordConfig.tsx | 16 +- .../channels/__tests__/DiscordConfig.test.tsx | 51 ++++ .../__tests__/useChannelDefinitions.test.tsx | 103 +++++++ app/src/hooks/useChannelDefinitions.ts | 80 ++++- app/src/lib/channels/definitions.ts | 21 ++ .../api/channelConnectionsApi.test.ts | 30 ++ app/src/services/api/channelConnectionsApi.ts | 42 ++- app/src/types/channels.ts | 25 ++ .../specs/harness-composio-tool-flow.spec.ts | 4 +- src/api/rest.rs | 22 +- src/api/rest_tests.rs | 7 +- src/core/socketio.rs | 124 +++++++- src/openhuman/channels/commands.rs | 2 + .../channels/controllers/definitions.rs | 15 + .../channels/controllers/definitions_tests.rs | 20 ++ .../channels/controllers/ops/connect.rs | 119 +++++++- src/openhuman/channels/controllers/ops/mod.rs | 4 +- .../channels/controllers/ops/types.rs | 7 + .../channels/controllers/ops_tests.rs | 277 ++++++++++++++++++ src/openhuman/channels/controllers/schemas.rs | 55 ++++ src/openhuman/channels/proactive.rs | 152 ++++++++-- .../channels/providers/discord/channel.rs | 98 ++++++- .../providers/discord/channel_tests.rs | 79 ++++- .../providers/telegram/channel_core.rs | 11 + .../providers/telegram/channel_ops.rs | 12 + .../providers/telegram/channel_tests.rs | 22 ++ .../providers/telegram/channel_types.rs | 4 + src/openhuman/channels/runtime/startup.rs | 50 +++- src/openhuman/channels/traits.rs | 17 ++ src/openhuman/config/mod.rs | 1 + src/openhuman/config/schema/channels.rs | 5 + src/openhuman/config/schema/channels_tests.rs | 1 + src/openhuman/config/schema/load_tests.rs | 1 + src/openhuman/credentials/ops.rs | 2 +- src/openhuman/cron/tools/add.rs | 1 + .../migrate_legacy_embedding_provider.rs | 194 ++++++++++++ ...migrate_legacy_embedding_provider_tests.rs | 124 ++++++++ src/openhuman/migrations/mod.rs | 57 +++- src/openhuman/migrations/mod_tests.rs | 49 +++- .../config_auth_app_state_connectivity_e2e.rs | 18 +- tests/keyring_secretstore_e2e.rs | 1 + tests/keyring_secretstore_fresh_e2e.rs | 1 + ...ools_approval_channels_raw_coverage_e2e.rs | 11 +- 43 files changed, 1814 insertions(+), 121 deletions(-) create mode 100644 app/src/hooks/__tests__/useChannelDefinitions.test.tsx create mode 100644 src/openhuman/migrations/migrate_legacy_embedding_provider.rs create mode 100644 src/openhuman/migrations/migrate_legacy_embedding_provider_tests.rs diff --git a/app/src/components/channels/DiscordConfig.tsx b/app/src/components/channels/DiscordConfig.tsx index 5d6d72c66..3676d37ae 100644 --- a/app/src/components/channels/DiscordConfig.tsx +++ b/app/src/components/channels/DiscordConfig.tsx @@ -154,7 +154,10 @@ const DiscordConfig = ({ definition }: DiscordConfigProps) => { const credentials: Record = {}; for (const field of spec.fields) { - const val = fieldValues[key]?.[field.key]?.trim() ?? ''; + // `rawVal` is `undefined` only when the user never touched the field; + // an empty string means they entered something and then cleared it. + const rawVal = fieldValues[key]?.[field.key]; + const val = rawVal?.trim() ?? ''; if (field.required && !val) { dispatch( setChannelConnectionStatus({ @@ -169,7 +172,16 @@ const DiscordConfig = ({ definition }: DiscordConfigProps) => { ); return; } - if (val) credentials[field.key] = val; + if (val) { + credentials[field.key] = val; + } else if (rawVal !== undefined) { + // Field was edited and then cleared — submit an explicit empty value + // instead of omitting it, so the backend can distinguish "cleared" + // from "never entered". For the allowlist this is what makes clearing + // it on reconnect mean "allow everyone" rather than silently reusing + // the previously-saved list (#3794 review — Codex P2). + credentials[field.key] = ''; + } } const result = await channelConnectionsApi.connectChannel('discord', { diff --git a/app/src/components/channels/__tests__/DiscordConfig.test.tsx b/app/src/components/channels/__tests__/DiscordConfig.test.tsx index 056f62f0a..345009aef 100644 --- a/app/src/components/channels/__tests__/DiscordConfig.test.tsx +++ b/app/src/components/channels/__tests__/DiscordConfig.test.tsx @@ -55,6 +55,8 @@ describe('DiscordConfig', () => { renderWithProviders(); expect(screen.getByPlaceholderText(/Your Discord bot token/)).toBeInTheDocument(); expect(screen.getByPlaceholderText(/restrict to a specific server/)).toBeInTheDocument(); + // Issue #3763: the allowlist must be settable in the connect UI. + expect(screen.getByPlaceholderText(/Discord user IDs, or \* for everyone/)).toBeInTheDocument(); }); it('shows Connect buttons for each auth mode', () => { @@ -63,6 +65,55 @@ describe('DiscordConfig', () => { expect(connectButtons.length).toBe(3); }); + // #3794 review (Codex P2): clearing the allowlist must reach the backend as an + // explicit empty value so it means "allow everyone", instead of being omitted + // and silently reusing the previously-saved list on reconnect. + it('submits an explicit empty allowed_users when the field is cleared', async () => { + vi.mocked(channelConnectionsApi.connectChannel).mockResolvedValue({ + status: 'connected', + restart_required: true, + }); + renderWithProviders(); + + fireEvent.change(screen.getByPlaceholderText(/Your Discord bot token/), { + target: { value: 'bot-token-xyz' }, + }); + const allowlist = screen.getByPlaceholderText(/Discord user IDs, or \* for everyone/); + fireEvent.change(allowlist, { target: { value: '111,222' } }); + fireEvent.change(allowlist, { target: { value: '' } }); // user clears it + + // bot_token is the first auth mode, so its Connect button is index 0. + fireEvent.click(screen.getAllByRole('button', { name: 'Connect' })[0]); + + await waitFor(() => { + expect(channelConnectionsApi.connectChannel).toHaveBeenCalledWith('discord', { + authMode: 'bot_token', + credentials: { bot_token: 'bot-token-xyz', allowed_users: '' }, + }); + }); + }); + + it('omits allowed_users entirely when the field is never touched', async () => { + vi.mocked(channelConnectionsApi.connectChannel).mockResolvedValue({ + status: 'connected', + restart_required: true, + }); + renderWithProviders(); + + fireEvent.change(screen.getByPlaceholderText(/Your Discord bot token/), { + target: { value: 'bot-token-xyz' }, + }); + + fireEvent.click(screen.getAllByRole('button', { name: 'Connect' })[0]); + + await waitFor(() => { + expect(channelConnectionsApi.connectChannel).toHaveBeenCalledWith('discord', { + authMode: 'bot_token', + credentials: { bot_token: 'bot-token-xyz' }, + }); + }); + }); + it('passes clearMemory when disconnecting a connected bot token account', async () => { const store = createTestStore(); store.dispatch( diff --git a/app/src/hooks/__tests__/useChannelDefinitions.test.tsx b/app/src/hooks/__tests__/useChannelDefinitions.test.tsx new file mode 100644 index 000000000..b01bf3839 --- /dev/null +++ b/app/src/hooks/__tests__/useChannelDefinitions.test.tsx @@ -0,0 +1,103 @@ +import { renderHook, waitFor } from '@testing-library/react'; +import type { ReactNode } from 'react'; +import { Provider } from 'react-redux'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { channelConnectionsApi } from '../../services/api/channelConnectionsApi'; +import { store } from '../../store'; +import { resetChannelConnectionsState } from '../../store/channelConnectionsSlice'; +import type { ChannelStatusEntry } from '../../types/channels'; +import { resolveStatusPatch, useChannelDefinitions } from '../useChannelDefinitions'; + +vi.mock('../../services/api/channelConnectionsApi', () => ({ + channelConnectionsApi: { + listDefinitions: vi.fn(), + listStatus: vi.fn(), + getDefaultChannel: vi.fn(), + }, +})); + +const mockApi = vi.mocked(channelConnectionsApi); + +function entry(overrides: Partial): ChannelStatusEntry { + return { + channel_id: 'discord', + auth_mode: 'bot_token', + connected: false, + has_credentials: true, + ...overrides, + }; +} + +describe('resolveStatusPatch (issue #3712)', () => { + it('asserts connected and clears any prior error', () => { + expect(resolveStatusPatch(entry({ connected: true, error: 'stale' }), 'error')).toEqual({ + status: 'connected', + lastError: undefined, + }); + }); + + it('surfaces a live listener error with its reason', () => { + expect( + resolveStatusPatch(entry({ connected: false, error: 'gateway closed (4004)' }), 'connected') + ).toEqual({ status: 'error', lastError: 'gateway closed (4004)' }); + }); + + it('does not stomp an in-flight connect when not-connected with no error', () => { + expect(resolveStatusPatch(entry({ connected: false }), 'connecting')).toBeNull(); + }); + + it('downgrades a stale connected entry to disconnected', () => { + expect(resolveStatusPatch(entry({ connected: false }), 'connected')).toEqual({ + status: 'disconnected', + lastError: undefined, + }); + }); + + it('reports disconnected when there is no prior status', () => { + expect(resolveStatusPatch(entry({ connected: false }), undefined)).toEqual({ + status: 'disconnected', + lastError: undefined, + }); + }); +}); + +describe('useChannelDefinitions loadDefinitions (issue #3794)', () => { + const wrapper = ({ children }: { children: ReactNode }) => ( + {children} + ); + + beforeEach(() => { + store.dispatch(resetChannelConnectionsState()); + mockApi.listDefinitions.mockResolvedValue([]); + mockApi.getDefaultChannel.mockResolvedValue('discord'); + mockApi.listStatus.mockResolvedValue([ + { channel_id: 'discord', auth_mode: 'bot_token', connected: true, has_credentials: true }, + // Unknown channel from core must be skipped, not coerced into state (#3794). + { + channel_id: 'bogus', + auth_mode: 'bot_token', + connected: true, + has_credentials: true, + } as ChannelStatusEntry, + ]); + }); + + afterEach(() => { + store.dispatch(resetChannelConnectionsState()); + vi.clearAllMocks(); + }); + + it('seeds the default channel from core, syncs known channels, and skips unknown ones', async () => { + const { result } = renderHook(() => useChannelDefinitions(), { wrapper }); + await waitFor(() => expect(result.current.loading).toBe(false)); + + const state = store.getState().channelConnections; + // Default channel seeded from the core (source of truth). + expect(state.defaultMessagingChannel).toBe('discord'); + // Known channel synced as connected. + expect(state.connections.discord?.bot_token?.status).toBe('connected'); + // Unknown channel_id ignored — never added to state. + expect((state.connections as Record).bogus).toBeUndefined(); + }); +}); diff --git a/app/src/hooks/useChannelDefinitions.ts b/app/src/hooks/useChannelDefinitions.ts index ae1aeb15f..9f28611fb 100644 --- a/app/src/hooks/useChannelDefinitions.ts +++ b/app/src/hooks/useChannelDefinitions.ts @@ -3,15 +3,47 @@ import { useCallback, useEffect, useState } from 'react'; import { FALLBACK_DEFINITIONS } from '../lib/channels/definitions'; import { channelConnectionsApi } from '../services/api/channelConnectionsApi'; +import { store } from '../store'; import { completeBreakingMigration, + setDefaultMessagingChannel, upsertChannelConnection, } from '../store/channelConnectionsSlice'; import { useAppDispatch, useAppSelector } from '../store/hooks'; -import type { ChannelAuthMode, ChannelDefinition, ChannelType } from '../types/channels'; +import { + type ChannelAuthMode, + type ChannelConnectionStatus, + type ChannelDefinition, + type ChannelStatusEntry, + isChannelType, +} from '../types/channels'; const log = debug('channels:definitions'); +/** + * Map a backend channel-status entry to the Redux status patch to apply + * (issue #3712). A connected entry asserts `connected`; a configured-but-failing + * entry asserts `error` and carries the live failure reason. A not-connected + * entry with no error is skipped while a connect flow is still `connecting`, so + * a stale status poll doesn't stomp an in-flight attempt. Returns `null` when + * there is nothing to assert. + */ +export function resolveStatusPatch( + entry: ChannelStatusEntry, + currentStatus: ChannelConnectionStatus | undefined +): { status: ChannelConnectionStatus; lastError: string | undefined } | null { + if (entry.connected) { + return { status: 'connected', lastError: undefined }; + } + if (entry.error) { + return { status: 'error', lastError: entry.error }; + } + if (currentStatus === 'connecting') { + return null; + } + return { status: 'disconnected', lastError: undefined }; +} + export function useChannelDefinitions() { const dispatch = useAppDispatch(); const channelConnections = useAppSelector(state => state.channelConnections); @@ -33,30 +65,54 @@ export function useChannelDefinitions() { setError(null); try { - const [defs, statusEntries] = await Promise.all([ + const [defs, statusEntries, defaultChannel] = await Promise.all([ channelConnectionsApi.listDefinitions().catch(() => null), channelConnectionsApi.listStatus().catch(() => null), + channelConnectionsApi.getDefaultChannel().catch(() => null), ]); if (cancelled) return; + // Seed the default messaging channel from the core (source of truth for + // proactive routing) so the UI reflects the persisted selection across + // reloads, not just redux-persist's local copy (issue #3712). + if (defaultChannel) { + dispatch(setDefaultMessagingChannel(defaultChannel)); + } + const resolvedDefs = defs && Array.isArray(defs) && defs.length > 0 ? defs : FALLBACK_DEFINITIONS; setDefinitions(resolvedDefs); log('loaded %d channel definitions', resolvedDefs.length); if (statusEntries && Array.isArray(statusEntries)) { + // Read live store state (not the closed-over selector value) so the + // connecting-guard in `resolveStatusPatch` sees the current status. + const liveConnections = store.getState().channelConnections.connections; for (const entry of statusEntries) { - const channel = entry.channel_id as ChannelType; - const authMode = entry.auth_mode as ChannelAuthMode; - if (entry.connected) { - dispatch( - upsertChannelConnection({ - channel, - authMode, - patch: { status: 'connected', capabilities: ['read', 'write'] }, - }) - ); + // Skip unknown channels from core rather than coercing them into + // state as if valid (#3794 review). + if (!isChannelType(entry.channel_id)) { + log('ignoring unknown channel_id from status sync: %s', entry.channel_id); + continue; } + const channel = entry.channel_id; + const authMode = entry.auth_mode as ChannelAuthMode; + const currentStatus = liveConnections[channel]?.[authMode]?.status; + const patch = resolveStatusPatch(entry, currentStatus); + if (!patch) continue; + dispatch( + upsertChannelConnection({ + channel, + authMode, + patch: { + status: patch.status, + lastError: patch.lastError, + // Only assert capabilities when actually connected; leave them + // untouched otherwise so a disconnect doesn't wipe them. + ...(entry.connected ? { capabilities: ['read', 'write'] } : {}), + }, + }) + ); } log('synced %d status entries', statusEntries.length); } diff --git a/app/src/lib/channels/definitions.ts b/app/src/lib/channels/definitions.ts index d6a5f3d32..0d58670e2 100644 --- a/app/src/lib/channels/definitions.ts +++ b/app/src/lib/channels/definitions.ts @@ -48,6 +48,13 @@ export const FALLBACK_DEFINITIONS: ChannelDefinition[] = [ required: true, placeholder: '123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11', }, + { + key: 'chat_id', + label: 'Chat ID', + field_type: 'string', + required: false, + placeholder: 'Optional: default chat for outbound messages', + }, { key: 'allowed_users', label: 'Allowed Users', @@ -85,6 +92,20 @@ export const FALLBACK_DEFINITIONS: ChannelDefinition[] = [ required: false, placeholder: 'Optional: restrict to a specific server', }, + { + key: 'channel_id', + label: 'Channel ID', + field_type: 'string', + required: false, + placeholder: 'Optional: default channel for outbound messages', + }, + { + key: 'allowed_users', + label: 'Allowed Users', + field_type: 'string', + required: false, + placeholder: 'Comma-separated Discord user IDs, or * for everyone (blank = everyone)', + }, ], auth_action: undefined, }, diff --git a/app/src/services/api/channelConnectionsApi.test.ts b/app/src/services/api/channelConnectionsApi.test.ts index dea929ec5..388b2874d 100644 --- a/app/src/services/api/channelConnectionsApi.test.ts +++ b/app/src/services/api/channelConnectionsApi.test.ts @@ -38,3 +38,33 @@ describe('channelConnectionsApi.disconnectChannel', () => { }); }); }); + +describe('channelConnectionsApi default channel (issue #3712)', () => { + beforeEach(() => { + mockCallCoreRpc.mockReset(); + }); + + it('updatePreferences persists the default via channels_set_default', async () => { + mockCallCoreRpc.mockResolvedValue({ active_channel: 'discord', restart_required: false }); + await channelConnectionsApi.updatePreferences('discord'); + expect(mockCallCoreRpc).toHaveBeenCalledWith({ + method: 'openhuman.channels_set_default', + params: { channel: 'discord' }, + }); + }); + + it('getDefaultChannel returns the core active_channel', async () => { + mockCallCoreRpc.mockResolvedValue({ active_channel: 'telegram' }); + const result = await channelConnectionsApi.getDefaultChannel(); + expect(mockCallCoreRpc).toHaveBeenCalledWith({ + method: 'openhuman.channels_get_default', + params: {}, + }); + expect(result).toBe('telegram'); + }); + + it('getDefaultChannel returns null when active_channel is absent', async () => { + mockCallCoreRpc.mockResolvedValue({}); + expect(await channelConnectionsApi.getDefaultChannel()).toBeNull(); + }); +}); diff --git a/app/src/services/api/channelConnectionsApi.ts b/app/src/services/api/channelConnectionsApi.ts index f671856bc..7d802aef5 100644 --- a/app/src/services/api/channelConnectionsApi.ts +++ b/app/src/services/api/channelConnectionsApi.ts @@ -1,12 +1,13 @@ -import type { - BotPermissionCheck, - ChannelAuthMode, - ChannelConnectionResult, - ChannelDefinition, - ChannelStatusEntry, - ChannelType, - DiscordGuild, - DiscordTextChannel, +import { + type BotPermissionCheck, + type ChannelAuthMode, + type ChannelConnectionResult, + type ChannelDefinition, + type ChannelStatusEntry, + type ChannelType, + type DiscordGuild, + type DiscordTextChannel, + isChannelType, } from '../../types/channels'; import { callCoreRpc } from '../coreRpcClient'; @@ -244,8 +245,27 @@ export const channelConnectionsApi = { return normalizePermissionCheck(result); }, - /** Placeholder for default channel preference sync. */ + /** + * Persist the default messaging channel to the core (issue #3712). The core + * stores it in `channels_config.active_channel` and applies it live, so the + * agent's proactive delivery follows the selection without a restart. + */ updatePreferences: async (defaultMessagingChannel: ChannelType): Promise => { - void defaultMessagingChannel; + await callCoreRpc({ + method: 'openhuman.channels_set_default', + params: { channel: defaultMessagingChannel }, + }); + }, + + /** Read the core's persisted default messaging channel. */ + getDefaultChannel: async (): Promise => { + const result = await callCoreRpc({ + method: 'openhuman.channels_get_default', + params: {}, + }); + const record = expectObject<{ active_channel?: unknown }>(result, 'Channel get_default'); + // Validate against known slugs so an unexpected core value can't leak into + // Redux/API consumers despite the `ChannelType | null` contract (#3794 review). + return isChannelType(record.active_channel) ? record.active_channel : null; }, }; diff --git a/app/src/types/channels.ts b/app/src/types/channels.ts index 05ca949fa..d397b934f 100644 --- a/app/src/types/channels.ts +++ b/app/src/types/channels.ts @@ -1,5 +1,26 @@ export type ChannelType = 'telegram' | 'discord' | 'web' | 'lark' | 'dingtalk' | 'mcp' | 'yuanbao'; +/** Every valid {@link ChannelType}, for runtime validation of values that arrive + * from the core (which is typed `string`). `satisfies` keeps this list in + * lockstep with the `ChannelType` union — adding a member there without updating + * here is a compile error. */ +export const KNOWN_CHANNEL_TYPES = [ + 'telegram', + 'discord', + 'web', + 'lark', + 'dingtalk', + 'mcp', + 'yuanbao', +] as const satisfies readonly ChannelType[]; + +/** Runtime guard: narrow an untrusted value to a known `ChannelType`. Use before + * coercing core-provided `channel_id` / `active_channel` strings so unknown + * channels never leak into Redux/API consumers (issue #3794 review). */ +export function isChannelType(value: unknown): value is ChannelType { + return typeof value === 'string' && (KNOWN_CHANNEL_TYPES as readonly string[]).includes(value); +} + export type ChannelAuthMode = 'managed_dm' | 'oauth' | 'bot_token' | 'api_key'; export type ChannelConnectionStatus = 'connected' | 'connecting' | 'disconnected' | 'error'; @@ -74,6 +95,10 @@ export interface ChannelStatusEntry { auth_mode: ChannelAuthMode; connected: boolean; has_credentials: boolean; + /** Live failure reason from the supervised listener when the channel is + * configured but its runtime listener is currently failing (issue #3712). + * Absent when healthy, still starting, or for listener-less modes. */ + error?: string; } export interface ChannelConnectionResult { diff --git a/app/test/playwright/specs/harness-composio-tool-flow.spec.ts b/app/test/playwright/specs/harness-composio-tool-flow.spec.ts index f87d069e5..4bc3aa1e8 100644 --- a/app/test/playwright/specs/harness-composio-tool-flow.spec.ts +++ b/app/test/playwright/specs/harness-composio-tool-flow.spec.ts @@ -287,6 +287,8 @@ test.describe('Harness - Composio tool-call prompt flow', () => { await sendMessage(page, 'create a linear issue titled Fix authentication timeout'); await expect(page.getByText(CANARY).first()).toBeVisible({ timeout: 60_000 }); - await expect(page.getByText(/I have created the Linear issue/i)).toBeVisible(); + // `.first()` — the confirmation text also appears in the tool-result echo + // pane, so an unscoped match trips Playwright strict mode (2 elements). + await expect(page.getByText(/I have created the Linear issue/i).first()).toBeVisible(); }); }); diff --git a/src/api/rest.rs b/src/api/rest.rs index 649f25b4b..c3ae10d2f 100644 --- a/src/api/rest.rs +++ b/src/api/rest.rs @@ -272,9 +272,8 @@ struct LoginTokenConsumeEnvelope { } #[derive(Debug, Clone, Deserialize)] -#[serde(rename_all = "camelCase")] struct LoginTokenConsumeData { - jwt_token: String, + jwt: String, } /// Decrypted OAuth token payload for handing off tokens to a local service or skill. @@ -423,17 +422,21 @@ impl BackendOAuthClient { let token = login_token.trim(); anyhow::ensure!(!token.is_empty(), "login token is required"); + // Backend serves `POST /auth/login-token/consume` with the token in a JSON + // body `{ token, audience? }` and returns `{ success, data: { jwt } }` + // (see backend `routes/auth.ts`). The legacy + // `telegram/login-tokens/{token}/consume` path-param route was removed, so + // the old call 404'd and Telegram/OAuth-token login could never complete + // (WIRING_GAPS_AUDIT C1/C2). let url = self .base - .join(&format!( - "telegram/login-tokens/{}/consume", - urlencoding::encode(token) - )) + .join("auth/login-token/consume") .context("build login-token consume URL")?; let resp = self .client .post(url) + .json(&serde_json::json!({ "token": token })) .send() .await .context("consume login token")?; @@ -450,11 +453,8 @@ impl BackendOAuthClient { anyhow::bail!("consume login token unsuccessful: {text}"); } - let jwt = env.data.jwt_token.trim().to_string(); - anyhow::ensure!( - !jwt.is_empty(), - "consume login token response missing jwtToken" - ); + let jwt = env.data.jwt.trim().to_string(); + anyhow::ensure!(!jwt.is_empty(), "consume login token response missing jwt"); Ok(jwt) } diff --git a/src/api/rest_tests.rs b/src/api/rest_tests.rs index 9f13821c9..6ac969998 100644 --- a/src/api/rest_tests.rs +++ b/src/api/rest_tests.rs @@ -170,7 +170,7 @@ async fn spawn_header_capture_server() -> (String, CapturedHeaders) { captured.push(&headers); Json(json!({ "success": true, - "data": { "jwtToken": "mock-jwt-token" } + "data": { "jwt": "mock-jwt-token" } })) } @@ -184,10 +184,7 @@ async fn spawn_header_capture_server() -> (String, CapturedHeaders) { let captured = CapturedHeaders::default(); let app = Router::new() - .route( - "/telegram/login-tokens/{token}/consume", - post(capture_consume), - ) + .route("/auth/login-token/consume", post(capture_consume)) .route("/probe", get(capture_probe)) .with_state(captured.clone()); let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); diff --git a/src/core/socketio.rs b/src/core/socketio.rs index 28b549e7a..38f264c6c 100644 --- a/src/core/socketio.rs +++ b/src/core/socketio.rs @@ -574,6 +574,7 @@ pub fn spawn_web_channel_bridge(io: SocketIo) { let io_memory_sync = io.clone(); let io_agent_meetings = io.clone(); let io_tinyplace = io.clone(); + let io_channel_status = io.clone(); // 2. Dictation hotkey events → broadcast to all connected clients. tokio::spawn(async move { @@ -1143,6 +1144,104 @@ pub fn spawn_web_channel_bridge(io: SocketIo) { } log::debug!("[socketio] tinyplace stream bridge stopped"); }); + + // 11. Channel listener health → broadcast `channel:connection-updated` to + // all clients so the Messaging tab reflects the *live* connection state + // instead of a stale, credential-presence-only "Connected" (issue + // #3712). The supervised listener publishes `ChannelConnected` when it + // (re)enters its recv loop and `ChannelDisconnected { reason }` when it + // errors/exits. Only listener-backed channels (telegram/discord + // `bot_token`) fire these, so we map them to the `bot_token` auth mode — + // the frontend `normalizeChannelConnectionUpdatePayload` drops any + // channel/mode it doesn't recognise. + tokio::spawn(async move { + let bus = { + const RETRY_INTERVAL_MS: u64 = 250; + const MAX_WAIT_SECS: u64 = 30; + let max_attempts = (MAX_WAIT_SECS * 1000) / RETRY_INTERVAL_MS; + let mut attempts: u64 = 0; + loop { + if let Some(bus) = crate::core::event_bus::global() { + break bus; + } + attempts += 1; + if attempts > max_attempts { + log::warn!( + "[socketio] event_bus not initialised after {}s — channel_status bridge giving up", + MAX_WAIT_SECS + ); + return; + } + tokio::time::sleep(std::time::Duration::from_millis(RETRY_INTERVAL_MS)).await; + } + }; + let mut rx = bus.raw_receiver(); + loop { + let event = match rx.recv().await { + Ok(event) => event, + Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => { + log::warn!( + "[socketio] dropped {} event_bus events due to lag (channel_status bridge)", + skipped + ); + continue; + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + }; + let payload = match event { + crate::core::event_bus::DomainEvent::ChannelConnected { channel } => { + log::debug!( + "[socketio] broadcast channel:connection-updated {channel} -> connected" + ); + Some(channel_connection_update_payload( + &channel, + "connected", + None, + )) + } + crate::core::event_bus::DomainEvent::ChannelDisconnected { channel, reason } => { + log::debug!( + "[socketio] broadcast channel:connection-updated {channel} -> error reason_len={}", + reason.len() + ); + Some(channel_connection_update_payload( + &channel, + "error", + Some(&reason), + )) + } + _ => None, + }; + if let Some(payload) = payload { + // Emit both colon and underscore variants for FE compatibility. + let _ = io_channel_status.emit("channel:connection-updated", &payload); + let _ = io_channel_status.emit("channel_connection_updated", &payload); + } + } + log::debug!("[socketio] channel_status bridge stopped"); + }); +} + +/// Build the `channel:connection-updated` payload broadcast when a supervised +/// channel listener changes state (issue #3712). Listener-backed channels are +/// always the `bot_token` auth mode (the only mode that materialises a runtime +/// listener for telegram/discord); `last_error` carries the disconnect reason. +/// Matches the shape consumed by the frontend +/// `normalizeChannelConnectionUpdatePayload`. +pub(crate) fn channel_connection_update_payload( + channel: &str, + status: &str, + last_error: Option<&str>, +) -> serde_json::Value { + let mut payload = serde_json::json!({ + "channel": channel, + "auth_mode": "bot_token", + "status": status, + }); + if let Some(reason) = last_error { + payload["last_error"] = serde_json::Value::String(reason.to_string()); + } + payload } /// Join `socket` to `room`, logging the result. @@ -1252,7 +1351,30 @@ fn emit_with_aliases(socket: &SocketRef, name: &str, payload: &serde_json::Value #[cfg(test)] mod tests { - use super::{event_alias, origin_is_allowed}; + use super::{channel_connection_update_payload, event_alias, origin_is_allowed}; + + #[test] + fn channel_connection_update_payload_connected_omits_error() { + let payload = channel_connection_update_payload("discord", "connected", None); + assert_eq!(payload["channel"], "discord"); + // Listener-backed channels always map to the bot_token auth mode. + assert_eq!(payload["auth_mode"], "bot_token"); + assert_eq!(payload["status"], "connected"); + assert!( + payload.get("last_error").is_none(), + "connected payload must not carry a last_error: {payload}" + ); + } + + #[test] + fn channel_connection_update_payload_error_carries_reason() { + let payload = + channel_connection_update_payload("discord", "error", Some("gateway closed (4004)")); + assert_eq!(payload["channel"], "discord"); + assert_eq!(payload["auth_mode"], "bot_token"); + assert_eq!(payload["status"], "error"); + assert_eq!(payload["last_error"], "gateway closed (4004)"); + } #[test] fn event_alias_translates_between_delimiters() { diff --git a/src/openhuman/channels/commands.rs b/src/openhuman/channels/commands.rs index 105f40e11..19c6178a6 100644 --- a/src/openhuman/channels/commands.rs +++ b/src/openhuman/channels/commands.rs @@ -328,6 +328,7 @@ mod tests { config.channels_config = crate::openhuman::config::ChannelsConfig::default(); config.channels_config.telegram = Some(TelegramConfig { bot_token: "fake:token".into(), + chat_id: None, allowed_users: vec!["user1".into()], stream_mode: StreamMode::default(), draft_update_interval_ms: 2000, @@ -385,6 +386,7 @@ mod tests { config.channels_config = crate::openhuman::config::ChannelsConfig::default(); config.channels_config.telegram = Some(TelegramConfig { bot_token: "fake".into(), + chat_id: None, allowed_users: vec![], stream_mode: StreamMode::default(), draft_update_interval_ms: 2000, diff --git a/src/openhuman/channels/controllers/definitions.rs b/src/openhuman/channels/controllers/definitions.rs index ccb7baad9..28892fd31 100644 --- a/src/openhuman/channels/controllers/definitions.rs +++ b/src/openhuman/channels/controllers/definitions.rs @@ -195,6 +195,13 @@ fn telegram_definition() -> ChannelDefinition { required: true, placeholder: "123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11", }, + FieldRequirement { + key: "chat_id", + label: "Chat ID", + field_type: "string", + required: false, + placeholder: "Optional: default chat for outbound messages", + }, FieldRequirement { key: "allowed_users", label: "Allowed Users", @@ -247,6 +254,14 @@ fn discord_definition() -> ChannelDefinition { required: false, placeholder: "Optional: default channel for outbound messages", }, + FieldRequirement { + key: "allowed_users", + label: "Allowed Users", + field_type: "string", + required: false, + placeholder: + "Comma-separated Discord user IDs, or * for everyone (blank = everyone)", + }, ], auth_action: None, }, diff --git a/src/openhuman/channels/controllers/definitions_tests.rs b/src/openhuman/channels/controllers/definitions_tests.rs index a143591f9..321f550d5 100644 --- a/src/openhuman/channels/controllers/definitions_tests.rs +++ b/src/openhuman/channels/controllers/definitions_tests.rs @@ -77,6 +77,26 @@ fn discord_has_bot_token_and_oauth() { assert_eq!(managed.unwrap().auth_action, Some("discord_managed_link")); } +#[test] +fn discord_bot_token_exposes_allowed_users_field() { + // Issue #3763: the Discord bot_token connect UI must offer an allowed_users + // field (like Telegram) so a self-hosted bot's allowlist is set in-app + // instead of by hand-editing config.toml. + let def = find_channel_definition("discord").expect("discord not found"); + let bot_token = def + .auth_mode_spec(ChannelAuthMode::BotToken) + .expect("discord bot_token mode"); + let allowed = bot_token + .fields + .iter() + .find(|f| f.key == "allowed_users") + .expect("discord bot_token must expose an allowed_users field"); + assert!( + !allowed.required, + "allowed_users must be optional (blank = allow everyone)" + ); +} + #[test] fn find_unknown_channel_returns_none() { assert!(find_channel_definition("nonexistent").is_none()); diff --git a/src/openhuman/channels/controllers/ops/connect.rs b/src/openhuman/channels/controllers/ops/connect.rs index 1021dd7d0..0ba15f3c1 100644 --- a/src/openhuman/channels/controllers/ops/connect.rs +++ b/src/openhuman/channels/controllers/ops/connect.rs @@ -22,6 +22,34 @@ pub(crate) fn credential_provider(channel_id: &str, mode: ChannelAuthMode) -> St format!("channel:{}:{}", channel_id, mode) } +/// Merge a channel's live supervised-listener health into its credential/config +/// derived `connected` flag (issue #3712). +/// +/// Only listener-backed modes (those that materialise a TOML config block — +/// `has_config`) have a `channel:` health component, kept current by the +/// supervisor's `ChannelConnected`/`ChannelDisconnected` events. For those, a +/// live `error` overrides the optimistic presence-based `connected` and carries +/// the failure reason to the UI; an `ok` confirms it. While the listener is +/// still `starting` (or has no component yet) we keep the presence-based value +/// so a freshly-configured channel isn't reported as broken before its first +/// connect attempt. Modes without a runtime listener (e.g. managed-DM) are left +/// untouched. Returns `(connected, error)`. +pub(crate) fn merge_listener_health( + presence_connected: bool, + has_config: bool, + health_status: Option<&str>, + health_last_error: Option<&str>, +) -> (bool, Option) { + if !has_config { + return (presence_connected, None); + } + match health_status { + Some("error") => (false, health_last_error.map(str::to_string)), + Some("ok") => (true, None), + _ => (presence_connected, None), + } +} + pub(crate) fn channel_config_connected( config: &Config, channel_id: &str, @@ -255,6 +283,16 @@ pub async fn connect_channel( .to_string(); let allowed_users = parse_allowed_users(creds_map.get("allowed_users")); let allowed_users_count = allowed_users.len(); + // Default chat for recipient-less proactive sends (mirrors Discord's + // `channel_id`). Read fresh from the form each connect: present ⇒ use it + // (empty ⇒ cleared); absent ⇒ unset. + let chat_id = creds_map + .get("chat_id") + .and_then(|v| v.as_str()) + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(str::to_string); + let has_chat_id = chat_id.is_some(); let mut persisted = config.clone(); let (stream_mode, draft_update_interval_ms, silent_streaming, mention_only) = @@ -276,6 +314,7 @@ pub async fn connect_channel( persisted.channels_config.telegram = Some(TelegramConfig { bot_token, + chat_id, allowed_users, stream_mode, draft_update_interval_ms, @@ -291,6 +330,7 @@ pub async fn connect_channel( tracing::info!( target: "openhuman::channels", allowed_users_count, + has_chat_id, mention_only, "[telegram] connect_channel: wrote channels_config.telegram; restart core for listener to load token" ); @@ -318,13 +358,21 @@ pub async fn connect_channel( let mut persisted = config.clone(); let existing = persisted.channels_config.discord.as_ref(); - let parsed_allowed_users = parse_allowed_users(creds_map.get("allowed_users")); - let allowed_users = if parsed_allowed_users.is_empty() { - existing + // Distinguish an *explicitly cleared* allowlist from an *omitted* one. + // The field is advertised as "blank = everyone" (definitions.rs) and the + // provider treats an empty list as allow-all, but the old logic reused + // the saved list whenever the parsed value was empty — so a user who + // cleared the allowlist on reconnect stayed restricted to the previous + // users (#3794 review — Codex P2). The key is present in `creds_map` + // (even as an empty string) only when the FE sends it; a cleared field + // now submits an explicit empty value. So: present ⇒ honor literally + // (empty ⇒ allow-all); absent ⇒ reuse the saved list (reconnect + // convenience for callers that don't resend the field at all). + let allowed_users = match creds_map.get("allowed_users") { + Some(raw) => parse_allowed_users(Some(raw)), + None => existing .map(|cfg| cfg.allowed_users.clone()) - .unwrap_or_default() - } else { - parsed_allowed_users + .unwrap_or_default(), }; let allowed_users_count = allowed_users.len(); let listen_to_bots = parse_optional_bool(creds_map.get("listen_to_bots")) @@ -505,13 +553,25 @@ pub async fn channel_status( None => all_channel_definitions(), }; + // Snapshot live listener health once so every entry reflects the same + // moment. The supervisor keeps `channel:` components current via + // `ChannelConnected`/`ChannelDisconnected` (issue #3712). + let health = crate::openhuman::health::snapshot(); + let mut entries = Vec::new(); for def in &defs { + let comp = health.components.get(&format!("channel:{}", def.id)); for spec in &def.auth_modes { let provider_key = credential_provider(def.id, spec.mode); let has_creds = stored_providers.iter().any(|p| p == &provider_key); let has_config = channel_config_connected(config, def.id, spec.mode); - let connected = has_creds || has_config; + let presence_connected = has_creds || has_config; + let (connected, error) = merge_listener_health( + presence_connected, + has_config, + comp.map(|c| c.status.as_str()), + comp.and_then(|c| c.last_error.as_deref()), + ); entries.push(ChannelStatusEntry { channel_id: def.id.to_string(), auth_mode: spec.mode, @@ -521,6 +581,7 @@ pub async fn channel_status( // credentials. Collapsing these misleads callers that branch on // credential presence (e.g. "needs re-auth" surfaces). has_credentials: has_creds, + error, }); } } @@ -528,6 +589,50 @@ pub async fn channel_status( Ok(RpcOutcome::new(entries, vec![])) } +/// Set the default messaging channel for proactive agent delivery (issue #3712 +/// — "switch default channel Telegram↔Discord"). Persists +/// `channels_config.active_channel` and applies a runtime override +/// ([`crate::openhuman::channels::proactive::set_runtime_active_channel`]) so the +/// change takes effect immediately, without restarting the channel runtime. +pub async fn set_default_channel( + config: &mut Config, + channel: &str, +) -> Result, String> { + let canonical = channel.trim().to_ascii_lowercase(); + if canonical.is_empty() { + return Err("channel must not be empty".to_string()); + } + // Accept any known channel definition, plus the in-app "web" channel. + if canonical != "web" && find_channel_definition(&canonical).is_none() { + return Err(format!("unknown channel: {channel}")); + } + + config.channels_config.active_channel = Some(canonical.clone()); + config + .save() + .await + .map_err(|e| format!("failed to persist default channel: {e}"))?; + + // Apply live so proactive routing follows the new default immediately. + crate::openhuman::channels::proactive::set_runtime_active_channel(Some(canonical.clone())); + + Ok(RpcOutcome::single_log( + json!({ "active_channel": canonical, "restart_required": false }), + format!("default messaging channel set to {canonical}"), + )) +} + +/// Return the persisted default messaging channel +/// (`channels_config.active_channel`), defaulting to `"web"` when unset. +pub fn get_default_channel(config: &Config) -> Result, String> { + let active = config + .channels_config + .active_channel + .clone() + .unwrap_or_else(|| "web".to_string()); + Ok(RpcOutcome::new(json!({ "active_channel": active }), vec![])) +} + /// Return the slugs of all messaging channels currently connected, /// merging the two storage layers OpenHuman uses for connection state. /// diff --git a/src/openhuman/channels/controllers/ops/mod.rs b/src/openhuman/channels/controllers/ops/mod.rs index d8f024e18..1f238bc0a 100644 --- a/src/openhuman/channels/controllers/ops/mod.rs +++ b/src/openhuman/channels/controllers/ops/mod.rs @@ -23,12 +23,14 @@ pub(crate) use connect::channel_config_connected; #[cfg(test)] pub(crate) use connect::credential_provider; #[cfg(test)] +pub(crate) use connect::merge_listener_health; +#[cfg(test)] pub(crate) use connect::parse_allowed_users; // Re-export public ops functions. pub use connect::{ channel_status, connect_channel, connected_channel_slugs, describe_channel, disconnect_channel, - list_channels, test_channel, + get_default_channel, list_channels, set_default_channel, test_channel, }; pub use discord::{ discord_check_permissions, discord_link_check, discord_link_start, discord_list_channels, diff --git a/src/openhuman/channels/controllers/ops/types.rs b/src/openhuman/channels/controllers/ops/types.rs index ce4a65e98..c0f4af1a1 100644 --- a/src/openhuman/channels/controllers/ops/types.rs +++ b/src/openhuman/channels/controllers/ops/types.rs @@ -25,6 +25,13 @@ pub struct ChannelStatusEntry { pub auth_mode: super::super::definitions::ChannelAuthMode, pub connected: bool, pub has_credentials: bool, + /// Live failure reason from the supervised listener when the channel is + /// configured but its runtime listener is currently in an error state + /// (issue #3712 — surface a real error instead of a false "Connected"). + /// `None` when healthy, still starting, or when the mode has no runtime + /// listener (e.g. managed-DM, which routes through the backend bot). + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option, } /// Result returned by `test_channel`. diff --git a/src/openhuman/channels/controllers/ops_tests.rs b/src/openhuman/channels/controllers/ops_tests.rs index 8b8cbd5f1..1a038dcd6 100644 --- a/src/openhuman/channels/controllers/ops_tests.rs +++ b/src/openhuman/channels/controllers/ops_tests.rs @@ -52,6 +52,7 @@ fn channel_config_connected_covers_config_backed_modes() { config.channels_config.telegram = Some(TelegramConfig { bot_token: "telegram-token".into(), + chat_id: None, allowed_users: vec![], stream_mode: Default::default(), draft_update_interval_ms: 1000, @@ -307,6 +308,120 @@ async fn connect_discord_bot_token_persists_runtime_config() { ); } +#[tokio::test] +async fn connect_telegram_bot_token_persists_chat_id() { + let (_tmp, config) = isolated_test_config(); + let result = connect_channel( + &config, + "telegram", + ChannelAuthMode::BotToken, + serde_json::json!({ + "bot_token": "telegram-token-123", + "chat_id": " 987654 " + }), + ) + .await + .expect("telegram connect should succeed"); + + assert_eq!(result.value.status, "connected"); + assert!(result.value.restart_required); + + let raw = tokio::fs::read_to_string(&config.config_path) + .await + .expect("saved config should exist"); + let parsed: toml::Value = toml::from_str(&raw).expect("saved config should parse"); + let telegram = parsed + .get("channels_config") + .and_then(|v| v.get("telegram")) + .and_then(toml::Value::as_table) + .expect("channels_config.telegram should be persisted"); + + // chat_id is trimmed before persistence (mirrors Discord channel_id). + assert_eq!( + telegram.get("chat_id").and_then(toml::Value::as_str), + Some("987654") + ); +} + +/// Read the persisted Discord `allowed_users` array from the saved config.toml. +async fn reload_discord_allowed_users(config: &Config) -> Vec { + let raw = tokio::fs::read_to_string(&config.config_path) + .await + .expect("saved config should exist"); + let parsed: toml::Value = toml::from_str(&raw).expect("saved config should parse"); + parsed + .get("channels_config") + .and_then(|v| v.get("discord")) + .and_then(|v| v.get("allowed_users")) + .and_then(toml::Value::as_array) + .map(|arr| { + arr.iter() + .filter_map(toml::Value::as_str) + .map(str::to_string) + .collect() + }) + .unwrap_or_default() +} + +fn seed_discord_with_allowlist(config: &mut Config) { + config.channels_config.discord = Some(DiscordConfig { + bot_token: "discord-token-abc".to_string(), + guild_id: None, + channel_id: None, + allowed_users: vec!["111".to_string(), "222".to_string()], + listen_to_bots: false, + mention_only: false, + }); +} + +#[tokio::test] +async fn connect_discord_omitted_allowlist_reuses_existing() { + // Reconnecting without resending `allowed_users` keeps the saved list — the + // reconnect-convenience path (#3794 review — Codex P2). + let (_tmp, mut config) = isolated_test_config(); + seed_discord_with_allowlist(&mut config); + config.save().await.expect("seed should persist"); + + connect_channel( + &config, + "discord", + ChannelAuthMode::BotToken, + serde_json::json!({ "bot_token": "discord-token-abc" }), + ) + .await + .expect("reconnect should succeed"); + + assert_eq!( + reload_discord_allowed_users(&config).await, + vec!["111".to_string(), "222".to_string()], + "omitted allowed_users must reuse the previously-saved list" + ); +} + +#[tokio::test] +async fn connect_discord_cleared_allowlist_allows_everyone() { + // Clearing the allowlist in the UI submits an explicit empty value; the + // backend must honor it (empty ⇒ allow-all) instead of reusing the old list + // (#3794 review — Codex P2). + let (_tmp, mut config) = isolated_test_config(); + seed_discord_with_allowlist(&mut config); + config.save().await.expect("seed should persist"); + + connect_channel( + &config, + "discord", + ChannelAuthMode::BotToken, + serde_json::json!({ "bot_token": "discord-token-abc", "allowed_users": "" }), + ) + .await + .expect("reconnect should succeed"); + + assert!( + reload_discord_allowed_users(&config).await.is_empty(), + "an explicit empty allowed_users must clear the list (allow-all), not reuse it" + ); +} + #[tokio::test] async fn disconnect_discord_bot_token_clears_runtime_config() { let (_tmp, mut config) = isolated_test_config(); @@ -634,6 +749,168 @@ async fn channel_status_reports_managed_dm_credential_as_connected() { assert!(managed_dm.has_credentials); } +// --------------------------------------------------------------------------- +// Issue #3712: `channel_status` must reflect the *live* supervised-listener +// health, not just credential/config presence, so the Messaging tab never +// shows a false "Connected" while the listener is actually failing. +// --------------------------------------------------------------------------- + +#[test] +fn merge_listener_health_ignores_modes_without_a_listener() { + // managed-DM and other listener-less modes have no `channel:` health + // component — presence must pass through untouched and never set an error. + assert_eq!( + merge_listener_health(true, false, Some("error"), Some("boom")), + (true, None) + ); + assert_eq!( + merge_listener_health(false, false, None, None), + (false, None) + ); +} + +#[test] +fn merge_listener_health_error_overrides_presence_and_surfaces_reason() { + // Configured (presence == connected) but the live listener is failing → + // report disconnected and carry the reason to the UI. + assert_eq!( + merge_listener_health(true, true, Some("error"), Some("gateway 4004")), + (false, Some("gateway 4004".to_string())) + ); +} + +#[test] +fn merge_listener_health_ok_confirms_connected() { + assert_eq!( + merge_listener_health(true, true, Some("ok"), None), + (true, None) + ); +} + +#[test] +fn merge_listener_health_starting_keeps_presence() { + // Before the first connect attempt the component is "starting" (or absent): + // keep the presence-based value so a freshly-configured channel isn't shown + // as broken prematurely. + assert_eq!( + merge_listener_health(true, true, Some("starting"), None), + (true, None) + ); + assert_eq!(merge_listener_health(true, true, None, None), (true, None)); +} + +#[tokio::test] +async fn channel_status_surfaces_live_listener_error() { + let (_tmp, mut config) = isolated_test_config(); + + // Configure a bot_token Discord channel (materialises a runtime listener). + config.channels_config.discord = Some(DiscordConfig { + bot_token: "tok".to_string(), + guild_id: None, + channel_id: None, + allowed_users: vec![], + listen_to_bots: false, + mention_only: false, + }); + + // Simulate the supervisor reporting the listener as failed. + crate::openhuman::health::mark_component_error("channel:discord", "gateway closed (4004)"); + + let result = channel_status(&config, Some("discord")) + .await + .expect("channel_status should succeed"); + + let bot_token = result + .value + .iter() + .find(|e| e.auth_mode == ChannelAuthMode::BotToken) + .expect("bot_token entry"); + assert!( + !bot_token.connected, + "a failing listener must report not-connected: {:?}", + result.value + ); + assert_eq!( + bot_token.error.as_deref(), + Some("gateway closed (4004)"), + "the disconnect reason must be surfaced: {:?}", + result.value + ); + + // Recovery: once the supervisor marks the listener healthy, status flips + // back to connected with the error cleared. + crate::openhuman::health::mark_component_ok("channel:discord"); + let recovered = channel_status(&config, Some("discord")) + .await + .expect("channel_status should succeed"); + let bot_token = recovered + .value + .iter() + .find(|e| e.auth_mode == ChannelAuthMode::BotToken) + .expect("bot_token entry"); + assert!( + bot_token.connected, + "healthy listener should report connected" + ); + assert!(bot_token.error.is_none(), "error should clear on recovery"); +} + +// --------------------------------------------------------------------------- +// Issue #3712: default messaging channel switch (Telegram↔Discord). Setting the +// default must persist to `channels_config.active_channel`; an unknown channel +// must be rejected without clobbering the current value. +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn set_default_channel_persists_known_channels() { + let (_tmp, mut config) = isolated_test_config(); + assert!(config.channels_config.active_channel.is_none()); + + set_default_channel(&mut config, "Discord") + .await + .expect("set discord"); + assert_eq!( + config.channels_config.active_channel.as_deref(), + Some("discord"), + "channel must be canonicalised to lowercase and persisted" + ); + + set_default_channel(&mut config, "telegram") + .await + .expect("set telegram"); + assert_eq!( + config.channels_config.active_channel.as_deref(), + Some("telegram") + ); +} + +#[tokio::test] +async fn set_default_channel_rejects_unknown_and_empty() { + let (_tmp, mut config) = isolated_test_config(); + set_default_channel(&mut config, "discord") + .await + .expect("seed discord"); + + assert!(set_default_channel(&mut config, "myspace") + .await + .unwrap_err() + .contains("unknown channel"),); + assert!(set_default_channel(&mut config, " ").await.is_err()); + + // A rejected set must not clobber the previously persisted value. + assert_eq!( + config.channels_config.active_channel.as_deref(), + Some("discord") + ); +} + +#[test] +fn get_default_channel_defaults_to_web_when_unset() { + let (_tmp, config) = isolated_test_config(); + let out = get_default_channel(&config).expect("get default"); + assert_eq!(out.value["active_channel"], "web"); +} + #[tokio::test] async fn connected_channel_slugs_merges_credentials_and_config() { let (_tmp, mut config) = isolated_test_config(); diff --git a/src/openhuman/channels/controllers/schemas.rs b/src/openhuman/channels/controllers/schemas.rs index f204496f5..1a4c50b56 100644 --- a/src/openhuman/channels/controllers/schemas.rs +++ b/src/openhuman/channels/controllers/schemas.rs @@ -47,6 +47,12 @@ struct StatusParams { channel: Option, } +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct SetDefaultParams { + channel: String, +} + #[derive(Debug, Deserialize)] #[serde(rename_all = "camelCase")] struct TestParams { @@ -128,6 +134,8 @@ pub fn all_controller_schemas() -> Vec { schemas("connect"), schemas("disconnect"), schemas("status"), + schemas("set_default"), + schemas("get_default"), schemas("test"), schemas("telegram_login_start"), schemas("telegram_login_check"), @@ -166,6 +174,14 @@ pub fn all_registered_controllers() -> Vec { schema: schemas("status"), handler: handle_status, }, + RegisteredController { + schema: schemas("set_default"), + handler: handle_set_default, + }, + RegisteredController { + schema: schemas("get_default"), + handler: handle_get_default, + }, RegisteredController { schema: schemas("test"), handler: handle_test, @@ -290,6 +306,30 @@ pub fn schemas(function: &str) -> ControllerSchema { "Array of status entries per channel and auth mode.", )], }, + "set_default" => ControllerSchema { + namespace: "channels", + function: "set_default", + description: "Set the default messaging channel for proactive agent \ + delivery (persists active_channel + applies live).", + inputs: vec![required_string( + "channel", + "Channel identifier to make default (e.g. telegram, discord, web).", + )], + outputs: vec![json_output( + "result", + "Object with the new active_channel and restart_required flag.", + )], + }, + "get_default" => ControllerSchema { + namespace: "channels", + function: "get_default", + description: "Get the persisted default messaging channel.", + inputs: vec![], + outputs: vec![json_output( + "result", + "Object with the current active_channel.", + )], + }, "test" => ControllerSchema { namespace: "channels", function: "test", @@ -509,6 +549,21 @@ fn handle_status(params: Map) -> ControllerFuture { }) } +fn handle_set_default(params: Map) -> ControllerFuture { + Box::pin(async move { + let mut config = config_rpc::load_config_with_timeout().await?; + let p = deserialize_params::(params)?; + to_json(ops::set_default_channel(&mut config, p.channel.trim()).await?) + }) +} + +fn handle_get_default(_params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + to_json(ops::get_default_channel(&config)?) + }) +} + fn handle_test(params: Map) -> ControllerFuture { Box::pin(async move { let config = config_rpc::load_config_with_timeout().await?; diff --git a/src/openhuman/channels/proactive.rs b/src/openhuman/channels/proactive.rs index 711228dd3..788dea086 100644 --- a/src/openhuman/channels/proactive.rs +++ b/src/openhuman/channels/proactive.rs @@ -95,6 +95,63 @@ impl ProactiveMessageSubscriber { *guard = channel; } } + + /// Share this subscriber's active-channel handle so the runtime can update it + /// in place after construction (see [`register_active_channel_handle`]). + pub fn active_channel_handle(&self) -> Arc>> { + Arc::clone(&self.active_channel) + } +} + +/// Handle to the live proactive subscriber's `active_channel`, registered at +/// channel-runtime startup (issue #3712 — "switch default channel +/// Telegram↔Discord"). The `channels_set_default` RPC mutates this exact handle +/// via [`set_runtime_active_channel`] so a default-channel switch from the UI +/// takes effect without a restart. Only the full channel runtime registers +/// (the web-only subscriber can't deliver externally), and nothing registers in +/// unit tests — so [`set_runtime_active_channel`] is a no-op there and never +/// leaks across the parallel test suite. The choice is also persisted to +/// `config.channels_config.active_channel`, which seeds the handle on next start. +static ACTIVE_CHANNEL_HANDLE: std::sync::OnceLock>>>>> = + std::sync::OnceLock::new(); + +fn active_channel_handle_slot() -> &'static RwLock>>>> { + ACTIVE_CHANNEL_HANDLE.get_or_init(|| RwLock::new(None)) +} + +/// Register the live subscriber's active-channel handle so the RPC can update it +/// at runtime. Called once from channel-runtime startup; the latest registration +/// wins. +pub fn register_active_channel_handle(handle: Arc>>) { + if let Ok(mut slot) = active_channel_handle_slot().write() { + *slot = Some(handle); + } +} + +/// Update the live proactive subscriber's active channel. No-op when no +/// subscriber has registered a handle (e.g. unit tests, or before the channel +/// runtime starts) — config persistence still applies and the value is read at +/// next startup. +pub fn set_runtime_active_channel(channel: Option) { + // Clone the Arc out and drop the slot read-guard before locking the handle, + // so we never hold two locks at once and the borrow doesn't outlive the read. + let handle = match active_channel_handle_slot().read() { + Ok(slot) => slot.clone(), + Err(_) => return, + }; + let Some(handle) = handle else { + tracing::debug!("[proactive] set_runtime_active_channel: no live subscriber registered"); + return; + }; + // Bind the guard out of the match (rather than `if let`) so the write-lock + // temporary is dropped before `handle`, avoiding an E0597 borrow on the + // local `handle`. + let mut guard = match handle.write() { + Ok(guard) => guard, + Err(_) => return, + }; + tracing::debug!(channel = ?channel, "[proactive] runtime active channel updated"); + *guard = channel; } #[async_trait] @@ -159,6 +216,8 @@ impl EventHandler for ProactiveMessageSubscriber { }); // 2. If an active external channel is configured, deliver there too. + // The `channels_set_default` RPC mutates this handle in place (issue + // #3712), so reading it here picks up a live default-channel switch. let active = self .active_channel .read() @@ -173,6 +232,25 @@ impl EventHandler for ProactiveMessageSubscriber { let key = channel_name.to_ascii_lowercase(); if let Some(ch) = self.channels_by_name.get(&key) { + // Resolve a delivery target before doing any work. Proactive + // sends carry no inbound recipient, so the channel must supply + // its configured default (e.g. Discord's `channel_id`). Channels + // with no resolvable target (e.g. Telegram, which has no stored + // default chat) are skipped with a warning rather than handed an + // empty recipient that would hit the platform API with a blank + // chat/channel id (#3794 review — Codex P2). Web delivery above + // already happened, so skipping only drops the external echo. + let Some(recipient) = ch.proactive_target() else { + tracing::warn!( + source = %source, + channel = %key, + "[proactive] active external channel has no configured \ + delivery target for recipient-less proactive messages; \ + skipping external delivery (web delivery unaffected)" + ); + return; + }; + tracing::debug!( source = %source, channel = %key, @@ -223,7 +301,7 @@ impl EventHandler for ProactiveMessageSubscriber { } } - let send_result = ch.send(&SendMessage::new(message, "")).await; + let send_result = ch.send(&SendMessage::new(message, &recipient)).await; // Record the terminal status on the approval audit // row before we log the outcome — best-effort, see // #2135. `record_execution` itself logs write @@ -281,6 +359,31 @@ mod tests { struct MockChannel { name: String, send_count: Arc, + /// Configured proactive delivery target. `Some` ⇒ the channel can + /// receive recipient-less proactive sends; `None` ⇒ proactive routing + /// skips it (models Telegram, which has no stored default chat). + target: Option, + } + + impl MockChannel { + /// A channel that *can* receive proactive sends (target defaults to its + /// own name, mirroring Discord's configured `channel_id`). + fn new(name: &str, send_count: Arc) -> Self { + Self { + name: name.to_string(), + send_count, + target: Some(name.to_string()), + } + } + + /// A channel with no resolvable proactive target (e.g. Telegram). + fn without_target(name: &str, send_count: Arc) -> Self { + Self { + name: name.to_string(), + send_count, + target: None, + } + } } #[async_trait] @@ -288,6 +391,9 @@ mod tests { fn name(&self) -> &str { &self.name } + fn proactive_target(&self) -> Option { + self.target.clone() + } async fn send(&self, _message: &SendMessage) -> anyhow::Result<()> { self.send_count.fetch_add(1, Ordering::SeqCst); Ok(()) @@ -315,10 +421,7 @@ mod tests { #[tokio::test] async fn routes_to_active_external_channel() { let send_count = Arc::new(AtomicUsize::new(0)); - let ch: Arc = Arc::new(MockChannel { - name: "telegram".into(), - send_count: Arc::clone(&send_count), - }); + let ch: Arc = Arc::new(MockChannel::new("telegram", Arc::clone(&send_count))); let map: HashMap> = [("telegram".into(), ch)].into(); let sub = ProactiveMessageSubscriber::new(Arc::new(map), Some("telegram".into())); @@ -327,13 +430,29 @@ mod tests { assert_eq!(send_count.load(Ordering::SeqCst), 1); } + #[tokio::test] + async fn skips_external_when_channel_has_no_proactive_target() { + // The active channel is the configured default, but it has no resolvable + // delivery target (e.g. Telegram with no stored chat). Proactive routing + // must skip it rather than calling `send` with an empty recipient + // (#3794 review — Codex P2). + let send_count = Arc::new(AtomicUsize::new(0)); + let ch: Arc = Arc::new(MockChannel::without_target( + "telegram", + Arc::clone(&send_count), + )); + let map: HashMap> = [("telegram".into(), ch)].into(); + let sub = ProactiveMessageSubscriber::new(Arc::new(map), Some("telegram".into())); + + sub.handle(&proactive_event()).await; + + assert_eq!(send_count.load(Ordering::SeqCst), 0); + } + #[tokio::test] async fn skips_external_when_active_is_web() { let send_count = Arc::new(AtomicUsize::new(0)); - let ch: Arc = Arc::new(MockChannel { - name: "telegram".into(), - send_count: Arc::clone(&send_count), - }); + let ch: Arc = Arc::new(MockChannel::new("telegram", Arc::clone(&send_count))); let map: HashMap> = [("telegram".into(), ch)].into(); let sub = ProactiveMessageSubscriber::new(Arc::new(map), Some("web".into())); @@ -346,10 +465,7 @@ mod tests { #[tokio::test] async fn skips_external_when_active_is_none() { let send_count = Arc::new(AtomicUsize::new(0)); - let ch: Arc = Arc::new(MockChannel { - name: "telegram".into(), - send_count: Arc::clone(&send_count), - }); + let ch: Arc = Arc::new(MockChannel::new("telegram", Arc::clone(&send_count))); let map: HashMap> = [("telegram".into(), ch)].into(); let sub = ProactiveMessageSubscriber::new(Arc::new(map), None); @@ -361,10 +477,7 @@ mod tests { #[tokio::test] async fn runtime_update_active_channel() { let send_count = Arc::new(AtomicUsize::new(0)); - let ch: Arc = Arc::new(MockChannel { - name: "discord".into(), - send_count: Arc::clone(&send_count), - }); + let ch: Arc = Arc::new(MockChannel::new("discord", Arc::clone(&send_count))); let map: HashMap> = [("discord".into(), ch)].into(); let sub = ProactiveMessageSubscriber::new(Arc::new(map), None); @@ -381,10 +494,7 @@ mod tests { #[tokio::test] async fn ignores_non_proactive_events() { let send_count = Arc::new(AtomicUsize::new(0)); - let ch: Arc = Arc::new(MockChannel { - name: "telegram".into(), - send_count: Arc::clone(&send_count), - }); + let ch: Arc = Arc::new(MockChannel::new("telegram", Arc::clone(&send_count))); let map: HashMap> = [("telegram".into(), ch)].into(); let sub = ProactiveMessageSubscriber::new(Arc::new(map), Some("telegram".into())); diff --git a/src/openhuman/channels/providers/discord/channel.rs b/src/openhuman/channels/providers/discord/channel.rs index 9bd34aaff..d1826e470 100644 --- a/src/openhuman/channels/providers/discord/channel.rs +++ b/src/openhuman/channels/providers/discord/channel.rs @@ -42,10 +42,58 @@ impl DiscordChannel { } /// Check if a Discord user ID is in the allowlist. - /// Empty list means deny everyone until explicitly configured. - /// `"*"` means allow everyone. + /// + /// Empty list ⇒ allow-all: an unconfigured allowlist applies no per-user + /// restriction (the bot is still scoped to its configured guild/channel). + /// Previously an empty list denied *everyone*, so a bot connected via the UI + /// with the default-empty allowlist silently ignored every message and never + /// replied (issue #3712). This now matches the WhatsApp provider's + /// empty-⇒-allow-all convention. `"*"` also allows everyone; populate the + /// list with specific user IDs to restrict. fn is_user_allowed(&self, user_id: &str) -> bool { - self.allowed_users.iter().any(|u| u == "*" || u == user_id) + self.allowed_users.is_empty() || self.allowed_users.iter().any(|u| u == "*" || u == user_id) + } + + /// Decide whether a message passes the bot's guild scoping. + /// + /// - No configured guild → all messages pass (guild filter inactive). + /// - Configured guild, message in a *different* guild → blocked. + /// - Configured guild, message in that guild → allowed. + /// - Configured guild, DM (no `guild_id`) → only when the allowlist is + /// non-empty (explicit). `is_user_allowed` treats an empty list as + /// allow-all (the intended *within-guild* default), but DMs bypass the + /// guild filter — so a blank allowlist must not open a guild-scoped bot to + /// arbitrary DMs (#3794 review — Codex P1). `"*"` (non-empty) still allows. + fn passes_guild_scope( + configured_guild: Option<&str>, + msg_guild: Option<&str>, + allowlist_empty: bool, + ) -> bool { + let Some(gid) = configured_guild else { + return true; + }; + match msg_guild { + Some(g) => g == gid, + None => !allowlist_empty, + } + } + + /// Resolve the outbound recipient channel id. Prefer the message's explicit + /// recipient (e.g. the channel a reply targets); fall back to the bot's + /// configured `channel_id` for recipient-less sends such as proactive + /// cron/heartbeat delivery (#3794 review — Codex P2). `None` when neither is + /// available, so the caller surfaces an error instead of POSTing to an empty + /// channel id. + fn resolve_recipient<'a>( + msg_recipient: &'a str, + configured: Option<&'a str>, + ) -> Option<&'a str> { + let recipient = if msg_recipient.is_empty() { + configured.unwrap_or("") + } else { + msg_recipient + }; + (!recipient.is_empty()).then_some(recipient) } fn bot_user_id_from_token(token: &str) -> Option { @@ -191,14 +239,34 @@ impl Channel for DiscordChannel { "discord" } + /// Recipient-less proactive sends (cron/heartbeat) deliver to the bot's + /// configured default `channel_id`. `None` when unconfigured, so proactive + /// routing skips Discord rather than letting `send` bail on an empty target + /// (#3794 review — Codex P2). + fn proactive_target(&self) -> Option { + self.channel_id + .as_deref() + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(str::to_string) + } + async fn send(&self, message: &SendMessage) -> anyhow::Result<()> { + // Resolve the target channel: explicit recipient (replies) or the + // configured default channel (recipient-less proactive sends). Bail with + // a clear error rather than POSTing to an empty channel id (#3794 review). + let Some(recipient) = + Self::resolve_recipient(&message.recipient, self.channel_id.as_deref()) + else { + anyhow::bail!( + "Discord send: no target channel — message had no recipient and no channel_id is configured" + ); + }; + let chunks = split_message_for_discord(&message.content); for (i, chunk) in chunks.iter().enumerate() { - let url = format!( - "https://discord.com/api/v10/channels/{}/messages", - message.recipient - ); + let url = format!("https://discord.com/api/v10/channels/{recipient}/messages"); let body = json!({ "content": chunk }); @@ -378,15 +446,13 @@ impl Channel for DiscordChannel { continue; } - // Guild filter - if let Some(ref gid) = guild_filter { - let msg_guild = d.get("guild_id").and_then(serde_json::Value::as_str); - // DMs have no guild_id — let them through; for guild messages, enforce the filter - if let Some(g) = msg_guild { - if g != gid { - continue; - } - } + // Guild filter + DM scoping (#3794 review — Codex P1) + if !Self::passes_guild_scope( + guild_filter.as_deref(), + d.get("guild_id").and_then(serde_json::Value::as_str), + self.allowed_users.is_empty(), + ) { + continue; } // Channel filter — only process messages from the configured channel diff --git a/src/openhuman/channels/providers/discord/channel_tests.rs b/src/openhuman/channels/providers/discord/channel_tests.rs index 10eee0945..06c79ee5b 100644 --- a/src/openhuman/channels/providers/discord/channel_tests.rs +++ b/src/openhuman/channels/providers/discord/channel_tests.rs @@ -22,10 +22,13 @@ fn bot_user_id_extraction() { } #[test] -fn empty_allowlist_denies_everyone() { +fn empty_allowlist_allows_everyone() { + // Issue #3712: an unconfigured (empty) allowlist must apply no per-user + // restriction — otherwise a UI-connected bot silently ignores every message + // and never replies. Scope is still enforced by guild/channel filters. let ch = DiscordChannel::new("fake".into(), None, None, vec![], false, false); - assert!(!ch.is_user_allowed("12345")); - assert!(!ch.is_user_allowed("anyone")); + assert!(ch.is_user_allowed("12345")); + assert!(ch.is_user_allowed("anyone")); } #[test] @@ -500,3 +503,73 @@ fn channel_id_defaults_to_none() { let ch = DiscordChannel::new("token".into(), None, None, vec![], false, false); assert!(ch.channel_id.is_none()); } + +#[test] +fn passes_guild_scope_covers_guild_dm_and_unscoped_cases() { + // No configured guild → everything passes (filter inactive). + assert!(DiscordChannel::passes_guild_scope(None, Some("g1"), true)); + assert!(DiscordChannel::passes_guild_scope(None, None, true)); + // Configured guild: same guild passes, other guild blocked. + assert!(DiscordChannel::passes_guild_scope( + Some("g1"), + Some("g1"), + true + )); + assert!(!DiscordChannel::passes_guild_scope( + Some("g1"), + Some("g2"), + true + )); + // #3794 review (Codex P1): DM (no guild_id) under guild scope is blocked + // with a blank allowlist, allowed with an explicit one. + assert!(!DiscordChannel::passes_guild_scope(Some("g1"), None, true)); + assert!(DiscordChannel::passes_guild_scope(Some("g1"), None, false)); +} + +#[test] +fn resolve_recipient_prefers_explicit_then_configured_channel() { + // #3794 review (Codex P2): recipient-less proactive sends fall back to the + // configured channel_id; an explicit recipient always wins. + assert_eq!( + DiscordChannel::resolve_recipient("123", Some("999")), + Some("123") + ); + assert_eq!( + DiscordChannel::resolve_recipient("", Some("999")), + Some("999") + ); + // Neither available → None, so the caller errors instead of POSTing to "". + assert_eq!(DiscordChannel::resolve_recipient("", None), None); + assert_eq!(DiscordChannel::resolve_recipient("", Some("")), None); +} + +#[test] +fn proactive_target_uses_configured_channel_id() { + use crate::openhuman::channels::traits::Channel; + + // Configured channel_id ⇒ recipient-less proactive sends have a target. + let with_channel = DiscordChannel::new( + "fake".into(), + None, + Some("12345".into()), + vec![], + false, + false, + ); + assert_eq!(with_channel.proactive_target(), Some("12345".to_string())); + + // No channel_id ⇒ None, so proactive routing skips Discord (#3794 Codex P2). + let no_channel = DiscordChannel::new("fake".into(), None, None, vec![], false, false); + assert_eq!(no_channel.proactive_target(), None); + + // Whitespace-only channel_id is treated as unset. + let blank_channel = DiscordChannel::new( + "fake".into(), + None, + Some(" ".into()), + vec![], + false, + false, + ); + assert_eq!(blank_channel.proactive_target(), None); +} diff --git a/src/openhuman/channels/providers/telegram/channel_core.rs b/src/openhuman/channels/providers/telegram/channel_core.rs index 8c2628250..61073ca9c 100644 --- a/src/openhuman/channels/providers/telegram/channel_core.rs +++ b/src/openhuman/channels/providers/telegram/channel_core.rs @@ -49,6 +49,7 @@ impl TelegramChannel { Self { bot_token, + chat_id: None, api_base, allowed_users: Arc::new(RwLock::new(normalized_allowed)), pairing, @@ -79,6 +80,16 @@ impl TelegramChannel { self } + /// Set the default chat for recipient-less proactive sends. A blank or + /// whitespace-only value is treated as unset (`None`), so proactive routing + /// skips Telegram rather than POSTing to an empty `chat_id`. + pub fn with_chat_id(mut self, chat_id: Option) -> Self { + self.chat_id = chat_id + .map(|s| s.trim().to_string()) + .filter(|s| !s.is_empty()); + self + } + /// Parse reply_target into (chat_id, optional thread_id). pub(crate) fn parse_reply_target(reply_target: &str) -> (String, Option) { if let Some((chat_id, thread_id)) = reply_target.split_once(':') { diff --git a/src/openhuman/channels/providers/telegram/channel_ops.rs b/src/openhuman/channels/providers/telegram/channel_ops.rs index a9978986f..d5084abce 100644 --- a/src/openhuman/channels/providers/telegram/channel_ops.rs +++ b/src/openhuman/channels/providers/telegram/channel_ops.rs @@ -15,6 +15,18 @@ impl Channel for TelegramChannel { "telegram" } + /// Recipient-less proactive sends (cron/heartbeat) deliver to the bot's + /// configured default `chat_id`. `None` when unconfigured, so proactive + /// routing skips Telegram rather than letting `send` POST to an empty + /// `chat_id` (mirrors Discord — #3712 Telegram parity). + fn proactive_target(&self) -> Option { + self.chat_id + .as_deref() + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(str::to_string) + } + fn supports_reactions(&self) -> bool { true } diff --git a/src/openhuman/channels/providers/telegram/channel_tests.rs b/src/openhuman/channels/providers/telegram/channel_tests.rs index 5ed6394b3..09200abe5 100644 --- a/src/openhuman/channels/providers/telegram/channel_tests.rs +++ b/src/openhuman/channels/providers/telegram/channel_tests.rs @@ -17,6 +17,28 @@ fn telegram_channel_name() { assert_eq!(ch.name(), "telegram"); } +#[test] +fn proactive_target_uses_configured_chat_id() { + // Unset by default ⇒ proactive routing skips Telegram (#3712 parity). + let default = TelegramChannel::new("fake-token".into(), vec!["*".into()], false); + assert_eq!(default.proactive_target(), None); + + // Configured chat_id ⇒ recipient-less proactive sends have a target. + let with_chat = TelegramChannel::new("fake-token".into(), vec!["*".into()], false) + .with_chat_id(Some("12345".into())); + assert_eq!(with_chat.proactive_target(), Some("12345".to_string())); + + // Whitespace-only chat_id is normalized to unset. + let blank = TelegramChannel::new("fake-token".into(), vec!["*".into()], false) + .with_chat_id(Some(" ".into())); + assert_eq!(blank.proactive_target(), None); + + // Explicit None passed to the builder stays unset. + let none = + TelegramChannel::new("fake-token".into(), vec!["*".into()], false).with_chat_id(None); + assert_eq!(none.proactive_target(), None); +} + #[test] fn typing_handle_starts_as_none() { let ch = TelegramChannel::new("fake-token".into(), vec!["*".into()], false); diff --git a/src/openhuman/channels/providers/telegram/channel_types.rs b/src/openhuman/channels/providers/telegram/channel_types.rs index f539b362d..4a0c10e2c 100644 --- a/src/openhuman/channels/providers/telegram/channel_types.rs +++ b/src/openhuman/channels/providers/telegram/channel_types.rs @@ -49,6 +49,10 @@ pub(crate) struct TelegramVoiceAttachment { /// Telegram channel — long-polls the Bot API for updates pub struct TelegramChannel { pub(crate) bot_token: String, + /// Default chat for recipient-less proactive sends. `None` ⇒ proactive + /// routing skips Telegram (see `proactive_target`). Set from + /// `TelegramConfig::chat_id` via [`TelegramChannel::with_chat_id`]. + pub(crate) chat_id: Option, /// Base URL for the Telegram Bot API. Defaults to `https://api.telegram.org`. /// Override via `OPENHUMAN_TELEGRAM_BOT_API_BASE` for E2E testing against a /// mock server. The legacy `OPENHUMAN_TELEGRAM_API_BASE` alias is still accepted. diff --git a/src/openhuman/channels/runtime/startup.rs b/src/openhuman/channels/runtime/startup.rs index 77843dc93..9785dee3b 100644 --- a/src/openhuman/channels/runtime/startup.rs +++ b/src/openhuman/channels/runtime/startup.rs @@ -299,14 +299,41 @@ pub async fn start_channels(mut config: Config) -> Result<()> { let local_embedding = config.workload_local_model("embeddings"); let embedding_api_key = crate::openhuman::embeddings::resolve_api_key(&config, &config.memory.embedding_provider); - let mem: Arc = Arc::from(memory_store::create_memory_with_local_ai( + // Build the memory store. A misconfigured/removed embedding provider (e.g. a + // stale `embedding_provider = "fastembed"` that the factory no longer knows) + // makes the embedder build fail — but that must NOT take every messaging + // channel offline (issue #3712). Fall back to keyword-only memory + // (`embedding_provider = "none"` → NoopEmbedding) so the channel listeners + // still start; semantic memory degrades gracefully instead of the whole + // runtime aborting. + let mem: Arc = match memory_store::create_memory_with_local_ai( &config.memory, local_embedding.as_deref(), &embedding_api_key, &[], Some(&config.storage.provider.config), &config.workspace_dir, - )?); + ) { + Ok(mem) => Arc::from(mem), + Err(e) => { + tracing::error!( + error = %format!("{e:#}"), + provider = %config.memory.embedding_provider, + "[channels] memory embedder build failed — falling back to keyword-only \ + memory so channels still start" + ); + let mut fallback_memory = config.memory.clone(); + fallback_memory.embedding_provider = "none".to_string(); + Arc::from(memory_store::create_memory_with_local_ai( + &fallback_memory, + local_embedding.as_deref(), + &embedding_api_key, + &[], + Some(&config.storage.provider.config), + &config.workspace_dir, + )?) + } + }; // Build system prompt from workspace identity files + skills let workspace = config.workspace_dir.clone(); let tools_registry = Arc::new(tools::all_tools_with_runtime( @@ -451,7 +478,8 @@ pub async fn start_channels(mut config: Config) -> Result<()> { tg.stream_mode, tg.draft_update_interval_ms, tg.silent_streaming, - ), + ) + .with_chat_id(tg.chat_id.clone()), )); } else { tracing::info!( @@ -692,12 +720,16 @@ pub async fn start_channels(mut config: Config) -> Result<()> { // Register the proactive message subscriber so morning briefings, // welcome messages, and other proactive agent output gets routed to // the user's active channel (+ always to web). - let _proactive_handle = bus.subscribe(Arc::new( - crate::openhuman::channels::proactive::ProactiveMessageSubscriber::new( - Arc::clone(&channels_by_name), - config.channels_config.active_channel.clone(), - ), - )); + let proactive_sub = crate::openhuman::channels::proactive::ProactiveMessageSubscriber::new( + Arc::clone(&channels_by_name), + config.channels_config.active_channel.clone(), + ); + // Expose its active-channel handle so the `channels_set_default` RPC can + // switch the default channel at runtime without a restart (issue #3712). + crate::openhuman::channels::proactive::register_active_channel_handle( + proactive_sub.active_channel_handle(), + ); + let _proactive_handle = bus.subscribe(Arc::new(proactive_sub)); let _telegram_remote_handle = if channels_by_name.contains_key("telegram") { let handle = bus.subscribe(Arc::new( crate::openhuman::channels::providers::telegram::TelegramRemoteSubscriber::new( diff --git a/src/openhuman/channels/traits.rs b/src/openhuman/channels/traits.rs index d70c2c2da..4165f1cdc 100644 --- a/src/openhuman/channels/traits.rs +++ b/src/openhuman/channels/traits.rs @@ -62,6 +62,20 @@ pub trait Channel: Send + Sync { /// Human-readable channel name fn name(&self) -> &str; + /// Resolve the delivery target for a *recipient-less* proactive send (cron / + /// heartbeat), where the caller has no inbound message to reply to. + /// + /// Returns the channel's configured default target (e.g. Discord's + /// `channel_id`) or `None` when the channel has no target it can deliver to + /// without an explicit recipient. Proactive routing skips channels that + /// return `None` instead of POSTing to an empty recipient (#3794 review — + /// Codex P2; Telegram has no configured default chat, so its `send` would + /// otherwise call the Bot API with an empty `chat_id`). Default `None` keeps + /// every existing provider opted out until it wires a real target. + fn proactive_target(&self) -> Option { + None + } + /// Send a message through this channel async fn send(&self, message: &SendMessage) -> anyhow::Result<()>; @@ -188,6 +202,9 @@ mod tests { .send(&SendMessage::new("hello", "bob")) .await .is_ok()); + // A provider that does not override `proactive_target` opts out of + // recipient-less proactive delivery (#3794). + assert_eq!(channel.proactive_target(), None); } #[tokio::test] diff --git a/src/openhuman/config/mod.rs b/src/openhuman/config/mod.rs index f89ba9e3a..c7e48e6e6 100644 --- a/src/openhuman/config/mod.rs +++ b/src/openhuman/config/mod.rs @@ -82,6 +82,7 @@ mod tests { fn reexported_channel_configs_are_constructible() { let telegram = TelegramConfig { bot_token: "token".into(), + chat_id: None, allowed_users: vec!["alice".into()], stream_mode: StreamMode::default(), draft_update_interval_ms: 1000, diff --git a/src/openhuman/config/schema/channels.rs b/src/openhuman/config/schema/channels.rs index 60d2c1731..84ce18f8c 100644 --- a/src/openhuman/config/schema/channels.rs +++ b/src/openhuman/config/schema/channels.rs @@ -113,6 +113,11 @@ fn default_silent_streaming() -> bool { #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] pub struct TelegramConfig { pub bot_token: String, + /// Default chat for recipient-less *proactive* sends (morning briefings, + /// cron output, etc.). Mirrors `DiscordConfig::channel_id`: `None` ⇒ proactive + /// routing skips Telegram rather than POSTing to an empty `chat_id`. + #[serde(default)] + pub chat_id: Option, pub allowed_users: Vec, #[serde(default)] pub stream_mode: StreamMode, diff --git a/src/openhuman/config/schema/channels_tests.rs b/src/openhuman/config/schema/channels_tests.rs index 8b2d745f3..b96f8a5c9 100644 --- a/src/openhuman/config/schema/channels_tests.rs +++ b/src/openhuman/config/schema/channels_tests.rs @@ -41,6 +41,7 @@ fn has_listening_integrations_detects_telegram() { let mut cfg = ChannelsConfig::default(); cfg.telegram = Some(TelegramConfig { bot_token: "tok".into(), + chat_id: None, allowed_users: vec![], stream_mode: StreamMode::Off, draft_update_interval_ms: 1000, diff --git a/src/openhuman/config/schema/load_tests.rs b/src/openhuman/config/schema/load_tests.rs index 5e145854d..76aaca731 100644 --- a/src/openhuman/config/schema/load_tests.rs +++ b/src/openhuman/config/schema/load_tests.rs @@ -1818,6 +1818,7 @@ allowed_users = ["@admin"] }; cfg.channels_config.telegram = Some(TelegramConfig { bot_token: known_secret.to_string(), + chat_id: None, allowed_users: vec!["@admin".to_string()], stream_mode: StreamMode::Off, draft_update_interval_ms: 1000, diff --git a/src/openhuman/credentials/ops.rs b/src/openhuman/credentials/ops.rs index 4e912224f..22ff8e330 100644 --- a/src/openhuman/credentials/ops.rs +++ b/src/openhuman/credentials/ops.rs @@ -496,7 +496,7 @@ pub async fn consume_login_token( serde_json::json!({ "jwtToken": jwt_token }), vec![ format!( - "login token consumed via POST /telegram/login-tokens/:token/consume on {}", + "login token consumed via POST /auth/login-token/consume on {}", api_url.trim_end_matches('/') ), "session JWT received".to_string(), diff --git a/src/openhuman/cron/tools/add.rs b/src/openhuman/cron/tools/add.rs index 02d07d8de..4512ff42e 100644 --- a/src/openhuman/cron/tools/add.rs +++ b/src/openhuman/cron/tools/add.rs @@ -590,6 +590,7 @@ mod tests { }; config.channels_config.telegram = Some(TelegramConfig { bot_token: "test-token".into(), + chat_id: None, allowed_users: allowed, stream_mode: Default::default(), draft_update_interval_ms: 1000, diff --git a/src/openhuman/migrations/migrate_legacy_embedding_provider.rs b/src/openhuman/migrations/migrate_legacy_embedding_provider.rs new file mode 100644 index 000000000..0d1b60d31 --- /dev/null +++ b/src/openhuman/migrations/migrate_legacy_embedding_provider.rs @@ -0,0 +1,194 @@ +//! Migration 6 → 7: retire the removed `"fastembed"` embedding provider. +//! +//! ## The problem +//! +//! Older builds shipped a local `"fastembed"` embedding provider (BGE models, +//! 384 dims). It has since been removed from the binary entirely — it is not a +//! cargo feature, it simply no longer exists in +//! [`crate::openhuman::embeddings::factory::create_embedding_provider`], which +//! hard-errors on any unknown provider string. +//! +//! Users who selected (or defaulted to) `"fastembed"` on an older build keep +//! `embedding_provider = "fastembed"` in their persisted `config.toml`. On +//! upgrade, the channel runtime's memory store build calls the factory with +//! that stale value → `Err("unknown embedding provider: \"fastembed\"")` → +//! `start_channels` aborts → **all messaging channels (Telegram/Discord) go +//! offline** with no surfaced error (issue #3712). +//! +//! ## What this migration does +//! +//! A pure, idempotent mutation of the persisted `Config`: when +//! `memory.embedding_provider` is the removed `"fastembed"` value, rewrite it to +//! a still-supported provider and reset the model/dimensions accordingly, since +//! the legacy BGE model/384-dim values are incompatible with both targets. +//! +//! **Target selection (offline-aware).** `fastembed` was a *local* embedder, so +//! a config carrying it belonged to a user who chose local/offline embeddings. +//! To preserve that intent, the caller (`run_pending`) probes for a reachable +//! local Ollama server and passes `prefer_local`: +//! - `prefer_local = true` → `"ollama"` + `bge-m3` (1024-dim; Ollama auto-pulls +//! the model on first embed). Keeps the user local/offline. +//! - `prefer_local = false` → `"managed"` cloud backend + cloud defaults (the +//! fresh-install default), used when no local Ollama is reachable. +//! +//! Both targets are 1024-dim — matching the memory tree's fixed on-disk +//! `EMBEDDING_DIM=1024`. This step only ever rewrites `fastembed`, so cloud-only +//! users (never on `fastembed`) are untouched. +//! +//! Stored vectors written at the old signature are left in place: they are +//! ignored by signature-filtered vector search and re-generated lazily by the +//! existing re-embed backfill ([`crate::openhuman::memory_queue::ensure_reembed_backfill`]) +//! once memory next syncs. No DB surgery happens here — this mirrors the +//! pure-config-mutation contract of the other migration steps. +//! +//! ## Behaviour +//! +//! - `run` is a **pure, synchronous** in-memory mutation of `Config`; the caller +//! (`migrations::run_pending`) persists via `Config::save()` and bumps +//! `schema_version`. The (impure) reachability probe lives in +//! [`local_ollama_reachable`] and is invoked by the caller, not by `run`. +//! - Idempotent: once rewritten the provider is no longer `"fastembed"`, so a +//! second run is a no-op. +//! - Never touches keys/secrets or any other config field. + +use crate::openhuman::config::Config; +use crate::openhuman::embeddings::{ + DEFAULT_CLOUD_EMBEDDING_DIMENSIONS, DEFAULT_CLOUD_EMBEDDING_MODEL, DEFAULT_OLLAMA_DIMENSIONS, + DEFAULT_OLLAMA_MODEL, +}; + +/// The removed provider value that must not reach the embedding factory. +const REMOVED_PROVIDER: &str = "fastembed"; + +/// Managed cloud backend — the current fresh-install default. Used as the +/// rewrite target when no local Ollama is reachable. Matches +/// `create_embedding_provider`'s accepted name. +const MANAGED_PROVIDER: &str = "managed"; + +/// Local Ollama backend — the rewrite target when a local Ollama server is +/// reachable, preserving the offline intent of a `fastembed` config. Matches +/// `create_embedding_provider`'s accepted name. +const OLLAMA_PROVIDER: &str = "ollama"; + +/// Counters returned by [`run`] for diagnostics. Logged at INFO once per run. +#[derive(Debug, Default, Clone)] +pub struct MigrationStats { + /// Whether the removed `"fastembed"` provider was rewritten. + pub provider_migrated: bool, + /// Whether the rewrite target was local Ollama (`true`) vs managed cloud + /// (`false`). Only meaningful when `provider_migrated`. + pub migrated_to_local: bool, + /// Embedding dimensionality before the rewrite (for the log line). + pub old_dimensions: usize, + /// Embedding dimensionality after the rewrite. + pub new_dimensions: usize, +} + +/// Rewrite a persisted `"fastembed"` embedding provider to a still-supported +/// backend. +/// +/// `prefer_local` selects the target (see the module docs): `true` ⇒ local +/// Ollama (`bge-m3`, preserving offline intent), `false` ⇒ managed cloud. The +/// caller derives `prefer_local` from [`local_ollama_reachable`]. +/// +/// Synchronous — pure config mutation, no I/O. Caller persists via +/// `Config::save()` once `schema_version` is also bumped. +/// +/// Returns `anyhow::Result` for uniformity with the other migration steps in +/// [`super`]; this pass has no fallible operations today and always returns +/// `Ok`. +pub fn run(config: &mut Config, prefer_local: bool) -> anyhow::Result { + let mut stats = MigrationStats { + old_dimensions: config.memory.embedding_dimensions, + new_dimensions: config.memory.embedding_dimensions, + ..Default::default() + }; + + if !config + .memory + .embedding_provider + .trim() + .eq_ignore_ascii_case(REMOVED_PROVIDER) + { + log::debug!( + "[migrations][legacy-embedding] embedding_provider is not the removed \ + \"{REMOVED_PROVIDER}\" — nothing to do" + ); + return Ok(stats); + } + + // Both targets are 1024-dim (the memory tree's fixed on-disk EMBEDDING_DIM); + // the legacy 384-dim BGE values are incompatible with either, so stored + // vectors re-embed lazily via backfill regardless of which target we pick. + let (provider, model, dimensions) = if prefer_local { + ( + OLLAMA_PROVIDER, + DEFAULT_OLLAMA_MODEL, + DEFAULT_OLLAMA_DIMENSIONS, + ) + } else { + ( + MANAGED_PROVIDER, + DEFAULT_CLOUD_EMBEDDING_MODEL, + DEFAULT_CLOUD_EMBEDDING_DIMENSIONS, + ) + }; + + config.memory.embedding_provider = provider.to_string(); + config.memory.embedding_model = model.to_string(); + config.memory.embedding_dimensions = dimensions; + + stats.provider_migrated = true; + stats.migrated_to_local = prefer_local; + stats.new_dimensions = dimensions; + + log::info!( + "[migrations][legacy-embedding] embedding_provider \"{REMOVED_PROVIDER}\" -> \ + \"{provider}\" (model={model}, dims {} -> {}, local={prefer_local}); \ + stale vectors re-embed lazily via backfill", + stats.old_dimensions, + stats.new_dimensions, + ); + + Ok(stats) +} + +/// Best-effort probe: is a local Ollama server reachable at `base_url`? +/// +/// Invoked by [`super::run_pending`] to choose the [`run`] target — a reachable +/// local Ollama lets former-`fastembed` (i.e. local-embedding) users stay local +/// (`bge-m3`, auto-pulled on first embed) instead of being forced onto the +/// managed cloud backend. Bounded (1.5s) and non-fatal: any client-build error, +/// transport error, timeout, or non-2xx status ⇒ `false`, so the caller falls +/// back to managed. Kept out of [`run`] so the rewrite itself stays pure/sync. +pub(crate) async fn local_ollama_reachable(base_url: &str) -> bool { + let url = format!("{}/api/tags", base_url.trim_end_matches('/')); + let client = match reqwest::Client::builder() + .timeout(std::time::Duration::from_millis(1500)) + .build() + { + Ok(client) => client, + Err(error) => { + log::debug!("[migrations][legacy-embedding] ollama probe client build failed: {error}"); + return false; + } + }; + match client.get(&url).send().await { + Ok(resp) => { + let reachable = resp.status().is_success(); + log::debug!( + "[migrations][legacy-embedding] ollama probe {url} -> {} (reachable={reachable})", + resp.status() + ); + reachable + } + Err(error) => { + log::debug!("[migrations][legacy-embedding] ollama probe {url} failed: {error}"); + false + } + } +} + +#[cfg(test)] +#[path = "migrate_legacy_embedding_provider_tests.rs"] +mod tests; diff --git a/src/openhuman/migrations/migrate_legacy_embedding_provider_tests.rs b/src/openhuman/migrations/migrate_legacy_embedding_provider_tests.rs new file mode 100644 index 000000000..6a5c9e670 --- /dev/null +++ b/src/openhuman/migrations/migrate_legacy_embedding_provider_tests.rs @@ -0,0 +1,124 @@ +use super::*; +use crate::openhuman::embeddings::{ + DEFAULT_CLOUD_EMBEDDING_DIMENSIONS, DEFAULT_CLOUD_EMBEDDING_MODEL, DEFAULT_OLLAMA_DIMENSIONS, + DEFAULT_OLLAMA_MODEL, +}; + +fn config_with_provider(provider: &str, model: &str, dims: usize) -> Config { + let mut config = Config::default(); + config.memory.embedding_provider = provider.to_string(); + config.memory.embedding_model = model.to_string(); + config.memory.embedding_dimensions = dims; + config +} + +#[test] +fn rewrites_fastembed_to_managed_with_cloud_defaults() { + let mut config = config_with_provider("fastembed", "BGESmallENV15", 384); + + // No local Ollama reachable ⇒ managed cloud target. + let stats = run(&mut config, false).expect("migration should succeed"); + + assert!(stats.provider_migrated, "fastembed must be migrated"); + assert!(!stats.migrated_to_local, "managed target is not local"); + assert_eq!(stats.old_dimensions, 384); + assert_eq!(stats.new_dimensions, DEFAULT_CLOUD_EMBEDDING_DIMENSIONS); + assert_eq!(config.memory.embedding_provider, "managed"); + assert_eq!(config.memory.embedding_model, DEFAULT_CLOUD_EMBEDDING_MODEL); + assert_eq!( + config.memory.embedding_dimensions, + DEFAULT_CLOUD_EMBEDDING_DIMENSIONS + ); +} + +#[test] +fn rewrites_fastembed_to_ollama_when_local_preferred() { + let mut config = config_with_provider("fastembed", "BGESmallENV15", 384); + + // A reachable local Ollama ⇒ stay local (preserve offline intent). + let stats = run(&mut config, true).expect("migration should succeed"); + + assert!(stats.provider_migrated, "fastembed must be migrated"); + assert!(stats.migrated_to_local, "ollama target is local"); + assert_eq!(stats.old_dimensions, 384); + assert_eq!(stats.new_dimensions, DEFAULT_OLLAMA_DIMENSIONS); + assert_eq!(config.memory.embedding_provider, "ollama"); + assert_eq!(config.memory.embedding_model, DEFAULT_OLLAMA_MODEL); + assert_eq!( + config.memory.embedding_dimensions, + DEFAULT_OLLAMA_DIMENSIONS + ); + // Both targets land on the memory tree's fixed 1024-dim format. + assert_eq!( + DEFAULT_OLLAMA_DIMENSIONS, + DEFAULT_CLOUD_EMBEDDING_DIMENSIONS + ); +} + +#[test] +fn is_idempotent() { + let mut config = config_with_provider("fastembed", "BGESmallENV15", 384); + run(&mut config, false).expect("first run"); + let stats = run(&mut config, false).expect("second run"); + assert!( + !stats.provider_migrated, + "second run must be a no-op once provider is rewritten" + ); + assert_eq!(config.memory.embedding_provider, "managed"); +} + +#[test] +fn is_idempotent_after_local_rewrite() { + let mut config = config_with_provider("fastembed", "BGESmallENV15", 384); + run(&mut config, true).expect("first run"); + // Provider is now "ollama" (not "fastembed"), so a second run is a no-op + // regardless of the prefer_local flag. + let stats = run(&mut config, false).expect("second run"); + assert!(!stats.provider_migrated, "second run must be a no-op"); + assert_eq!(config.memory.embedding_provider, "ollama"); +} + +#[test] +fn matches_case_insensitively_and_trims() { + let mut config = config_with_provider(" FastEmbed ", "BGESmallENV15", 384); + let stats = run(&mut config, false).expect("migration should succeed"); + assert!(stats.provider_migrated); + assert_eq!(config.memory.embedding_provider, "managed"); +} + +#[test] +fn leaves_valid_providers_untouched() { + for provider in ["managed", "ollama", "voyage", "none", "openai"] { + // Untouched regardless of the prefer_local flag — only "fastembed" is rewritten. + for prefer_local in [false, true] { + let mut config = config_with_provider(provider, "some-model", 1024); + let stats = run(&mut config, prefer_local).expect("migration should succeed"); + assert!(!stats.provider_migrated, "{provider} must not be migrated"); + assert_eq!(config.memory.embedding_provider, provider); + assert_eq!(config.memory.embedding_model, "some-model"); + assert_eq!(config.memory.embedding_dimensions, 1024); + } + } +} + +#[tokio::test] +async fn reachable_probe_false_when_no_server() { + // Nothing listening ⇒ connection refused ⇒ false (caller falls back to managed). + assert!(!local_ollama_reachable("http://127.0.0.1:1").await); +} + +#[tokio::test] +async fn reachable_probe_true_for_2xx_server() { + use axum::{routing::get, Router}; + use tokio::net::TcpListener; + + let app = Router::new().route("/api/tags", get(|| async { "{\"models\":[]}" })); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + + // Trailing slash exercises the `trim_end_matches('/')` join. + assert!(local_ollama_reachable(&format!("http://{addr}/")).await); +} diff --git a/src/openhuman/migrations/mod.rs b/src/openhuman/migrations/mod.rs index 283cf6258..a245c2140 100644 --- a/src/openhuman/migrations/mod.rs +++ b/src/openhuman/migrations/mod.rs @@ -24,6 +24,7 @@ use crate::openhuman::config::Config; mod expand_autonomy_defaults; +mod migrate_legacy_embedding_provider; mod phase_out_profile_md; mod reconcile_orphaned_providers; mod remove_write_auto_approve; @@ -32,7 +33,7 @@ mod retire_chat_v1_model; mod unify_ai_provider_settings; /// Current target schema version. Bumped alongside every new migration. -pub const CURRENT_SCHEMA_VERSION: u32 = 6; +pub const CURRENT_SCHEMA_VERSION: u32 = 7; /// Run any migrations whose `schema_version` gate hasn't yet been /// crossed for this workspace. @@ -326,6 +327,60 @@ pub async fn run_pending(config: &mut Config) { ); } } + + // 6 -> 7: retire the removed `"fastembed"` embedding provider. Older builds + // shipped a local fastembed provider; it no longer exists in the embedding + // factory, which hard-errors on unknown provider strings. A persisted + // `embedding_provider = "fastembed"` therefore aborts `start_channels`' + // memory build and takes every messaging channel offline (issue #3712). + // `fastembed` was a *local* embedder, so prefer a still-local target when a + // local Ollama server is reachable (preserves the user's offline intent); + // otherwise fall back to the managed cloud default. The probe is bounded and + // best-effort, and only runs for `fastembed` configs (the only ones this step + // rewrites) so unaffected upgrades pay no network cost. Guard on `== 6` so an + // earlier failed step doesn't get skipped. + if config.schema_version == 6 { + let prefer_local = if config + .memory + .embedding_provider + .trim() + .eq_ignore_ascii_case("fastembed") + { + let base = crate::openhuman::inference::local::ollama_base_url_from_config(config); + migrate_legacy_embedding_provider::local_ollama_reachable(&base).await + } else { + false + }; + match migrate_legacy_embedding_provider::run(config, prefer_local) { + Ok(stats) => { + let previous_version = config.schema_version; + config.schema_version = 7; + if let Err(err) = config.save().await { + config.schema_version = previous_version; + log::warn!( + "[migrations] migrate_legacy_embedding_provider ran but config.save \ + failed: {err:#} — rolled in-memory schema_version back to \ + {previous_version}, will retry on next launch" + ); + return; + } + log::info!( + "[migrations] schema_version bumped to 7 (migrate_legacy_embedding_provider \ + provider_migrated={} migrated_to_local={} old_dims={} new_dims={})", + stats.provider_migrated, + stats.migrated_to_local, + stats.old_dimensions, + stats.new_dimensions, + ); + } + Err(err) => { + log::warn!( + "[migrations] migrate_legacy_embedding_provider failed: {err:#} — \ + will retry on next launch" + ); + } + } + } } #[cfg(test)] diff --git a/src/openhuman/migrations/mod_tests.rs b/src/openhuman/migrations/mod_tests.rs index 9d85024fc..d4cadbd3d 100644 --- a/src/openhuman/migrations/mod_tests.rs +++ b/src/openhuman/migrations/mod_tests.rs @@ -113,8 +113,8 @@ async fn run_pending_runs_phase_out_when_version_zero() { let on_disk = std::fs::read_to_string(&config.config_path).unwrap(); assert!( - on_disk.contains("schema_version = 6"), - "saved config.toml must record schema_version=6, got:\n{on_disk}" + on_disk.contains("schema_version = 7"), + "saved config.toml must record schema_version=7, got:\n{on_disk}" ); } @@ -129,7 +129,34 @@ async fn run_pending_bumps_version_on_fresh_install() { assert_eq!(config.schema_version, CURRENT_SCHEMA_VERSION); let on_disk = std::fs::read_to_string(&config.config_path).unwrap(); - assert!(on_disk.contains("schema_version = 6")); + assert!(on_disk.contains("schema_version = 7")); +} + +#[tokio::test] +async fn run_pending_migrates_fastembed_to_managed_without_local_ollama() { + let tmp = TempDir::new().unwrap(); + fs::create_dir_all(tmp.path().join("workspace")).unwrap(); + + let mut config = config_in(&tmp); + config.schema_version = 6; + config.memory.embedding_provider = "fastembed".to_string(); + config.memory.embedding_model = "BGESmallENV15".to_string(); + config.memory.embedding_dimensions = 384; + // Point the local-Ollama probe at a guaranteed-dead address so the rewrite + // target is deterministic (managed) regardless of whether the host happens + // to run Ollama on the default port. + config.local_ai.base_url = Some("http://127.0.0.1:1".to_string()); + + run_pending(&mut config).await; + + assert_eq!(config.schema_version, 7); + assert_eq!( + config.memory.embedding_provider, "managed", + "no reachable local Ollama ⇒ managed cloud target" + ); + assert_eq!(config.memory.embedding_dimensions, 1024); + let on_disk = std::fs::read_to_string(&config.config_path).unwrap(); + assert!(on_disk.contains("schema_version = 7")); } #[tokio::test] @@ -253,8 +280,8 @@ async fn run_pending_expands_autonomy_defaults_from_v3() { // On-disk config must reflect the new schema_version. let on_disk = fs::read_to_string(&config.config_path).unwrap(); assert!( - on_disk.contains("schema_version = 6"), - "saved config.toml must record schema_version=6, got:\n{on_disk}" + on_disk.contains("schema_version = 7"), + "saved config.toml must record schema_version=7, got:\n{on_disk}" ); } @@ -286,8 +313,8 @@ async fn run_pending_v4_to_v5_removes_write_tools_from_auto_approve() { let on_disk = fs::read_to_string(&config.config_path).unwrap(); assert!( - on_disk.contains("schema_version = 6"), - "saved config.toml must record schema_version=6, got:\n{on_disk}" + on_disk.contains("schema_version = 7"), + "saved config.toml must record schema_version=7, got:\n{on_disk}" ); } @@ -322,8 +349,8 @@ async fn run_pending_v5_to_v6_repairs_http_request_limits() { // The version bump must be persisted to disk too. let on_disk = fs::read_to_string(&config.config_path).unwrap(); assert!( - on_disk.contains("schema_version = 6"), - "saved config.toml must record schema_version=6, got:\n{on_disk}" + on_disk.contains("schema_version = 7"), + "saved config.toml must record schema_version=7, got:\n{on_disk}" ); } @@ -357,8 +384,8 @@ async fn run_pending_v5_to_v6_reconciles_orphaned_providers() { let on_disk = fs::read_to_string(&config.config_path).unwrap(); assert!( - on_disk.contains("schema_version = 6"), - "saved config.toml must record schema_version=6, got:\n{on_disk}" + on_disk.contains("schema_version = 7"), + "saved config.toml must record schema_version=7, got:\n{on_disk}" ); } diff --git a/tests/config_auth_app_state_connectivity_e2e.rs b/tests/config_auth_app_state_connectivity_e2e.rs index 2d3041e9b..e5ae84a36 100644 --- a/tests/config_auth_app_state_connectivity_e2e.rs +++ b/tests/config_auth_app_state_connectivity_e2e.rs @@ -154,12 +154,9 @@ async fn serve_mock_backend() -> ( let app = Router::new() .route("/auth/me", get(mock_auth_me)) .route("/api/auth/me", get(mock_auth_me)) + .route("/auth/login-token/consume", post(mock_consume_login_token)) .route( - "/telegram/login-tokens/{token}/consume", - post(mock_consume_login_token), - ) - .route( - "/api/telegram/login-tokens/{token}/consume", + "/api/auth/login-token/consume", post(mock_consume_login_token), ) .route( @@ -370,11 +367,17 @@ async fn static_auth_me( })) } -async fn mock_consume_login_token(AxumPath(token): AxumPath) -> Json { +async fn mock_consume_login_token(Json(body): Json) -> Json { + // Token now arrives in the JSON body (`{ token }`), not the URL path, and the + // response field is `jwt` (matches backend `routes/auth.ts`). + let token = body + .get("token") + .and_then(|v| v.as_str()) + .unwrap_or_default(); Json(json!({ "success": true, "data": { - "jwtToken": format!("jwt-from-{token}") + "jwt": format!("jwt-from-{token}") } })) } @@ -2106,6 +2109,7 @@ async fn config_save_and_load_encrypts_channel_secret_fields() { config.search.querit.api_key = Some("querit-secret".into()); config.channels_config.telegram = Some(TelegramConfig { bot_token: "telegram-secret".into(), + chat_id: None, allowed_users: vec!["alice".into()], stream_mode: Default::default(), draft_update_interval_ms: 1000, diff --git a/tests/keyring_secretstore_e2e.rs b/tests/keyring_secretstore_e2e.rs index 263df8c0f..5adaea2b6 100644 --- a/tests/keyring_secretstore_e2e.rs +++ b/tests/keyring_secretstore_e2e.rs @@ -77,6 +77,7 @@ async fn config_secrets_roundtrip_via_keyring_backed_master_key_migration() { channels_config: openhuman_core::openhuman::config::schema::ChannelsConfig { telegram: Some(TelegramConfig { bot_token: "tg-bot-secret".into(), + chat_id: None, allowed_users: vec!["alice".into()], stream_mode: StreamMode::default(), draft_update_interval_ms: 1000, diff --git a/tests/keyring_secretstore_fresh_e2e.rs b/tests/keyring_secretstore_fresh_e2e.rs index b3a115b01..455c7b836 100644 --- a/tests/keyring_secretstore_fresh_e2e.rs +++ b/tests/keyring_secretstore_fresh_e2e.rs @@ -74,6 +74,7 @@ async fn config_secrets_create_master_key_in_keyring_on_fresh_install() { channels_config: openhuman_core::openhuman::config::schema::ChannelsConfig { telegram: Some(TelegramConfig { bot_token: "fresh-tg-secret".into(), + chat_id: None, allowed_users: vec!["bob".into()], stream_mode: StreamMode::default(), draft_update_interval_ms: 1000, diff --git a/tests/tools_approval_channels_raw_coverage_e2e.rs b/tests/tools_approval_channels_raw_coverage_e2e.rs index 90601f667..c9d29003c 100644 --- a/tests/tools_approval_channels_raw_coverage_e2e.rs +++ b/tests/tools_approval_channels_raw_coverage_e2e.rs @@ -252,6 +252,13 @@ impl Channel for CapturingChannel { "capture" } + // External channel exposes a proactive delivery target so the proactive + // router resolves a recipient for it (#3794 — recipient-less proactive sends + // are skipped for channels that return `None`). + fn proactive_target(&self) -> Option { + Some("capture".to_string()) + } + async fn send(&self, message: &SendMessage) -> Result<()> { self.sent .lock() @@ -2207,7 +2214,9 @@ async fn proactive_subscriber_routes_web_and_active_external_channel_without_net let sent = capture.sent.lock().expect("capture lock").clone(); assert_eq!(sent.len(), 1); assert_eq!(sent[0].content, "send through active external channel"); - assert_eq!(sent[0].recipient, ""); + // Recipient is now resolved from the channel's proactive_target (#3794), + // rather than the previously-empty recipient. + assert_eq!(sent[0].recipient, "capture"); subscriber.set_active_channel(Some("web".into())); subscriber