fix(channels): repair Discord & Telegram messaging end-to-end (#3712, #3763) (#3794)

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
Co-authored-by: Steven Enamakel <enamakel@tinyhumans.ai>
This commit is contained in:
YellowSnnowmann
2026-06-22 12:17:21 -07:00
committed by GitHub
co-authored by Claude Opus 4.8 Steven Enamakel
parent 9198444ffd
commit 6ce4f52972
43 changed files with 1814 additions and 121 deletions
+14 -2
View File
@@ -154,7 +154,10 @@ const DiscordConfig = ({ definition }: DiscordConfigProps) => {
const credentials: Record<string, string> = {};
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', {
@@ -55,6 +55,8 @@ describe('DiscordConfig', () => {
renderWithProviders(<DiscordConfig definition={discordDef} />);
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(<DiscordConfig definition={discordDef} />);
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(<DiscordConfig definition={discordDef} />);
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(
@@ -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>): 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 }) => (
<Provider store={store}>{children}</Provider>
);
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<string, unknown>).bogus).toBeUndefined();
});
});
+68 -12
View File
@@ -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);
}
+21
View File
@@ -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,
},
@@ -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();
});
});
+31 -11
View File
@@ -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> => {
void defaultMessagingChannel;
await callCoreRpc({
method: 'openhuman.channels_set_default',
params: { channel: defaultMessagingChannel },
});
},
/** Read the core's persisted default messaging channel. */
getDefaultChannel: async (): Promise<ChannelType | null> => {
const result = await callCoreRpc<unknown>({
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;
},
};
+25
View File
@@ -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 {
@@ -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();
});
});
+11 -11
View File
@@ -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)
}
+2 -5
View File
@@ -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();
+123 -1
View File
@@ -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() {
+2
View File
@@ -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,
@@ -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,
},
@@ -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());
@@ -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:<id>` 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<String>) {
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:<id>` 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<RpcOutcome<Value>, 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<RpcOutcome<Value>, 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.
///
@@ -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,
@@ -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<String>,
}
/// Result returned by `test_channel`.
@@ -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<String> {
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:<id>` 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();
@@ -47,6 +47,12 @@ struct StatusParams {
channel: Option<String>,
}
#[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<ControllerSchema> {
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<RegisteredController> {
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<String, Value>) -> ControllerFuture {
})
}
fn handle_set_default(params: Map<String, Value>) -> ControllerFuture {
Box::pin(async move {
let mut config = config_rpc::load_config_with_timeout().await?;
let p = deserialize_params::<SetDefaultParams>(params)?;
to_json(ops::set_default_channel(&mut config, p.channel.trim()).await?)
})
}
fn handle_get_default(_params: Map<String, Value>) -> 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<String, Value>) -> ControllerFuture {
Box::pin(async move {
let config = config_rpc::load_config_with_timeout().await?;
+131 -21
View File
@@ -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<RwLock<Option<String>>> {
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<RwLock<Option<Arc<RwLock<Option<String>>>>>> =
std::sync::OnceLock::new();
fn active_channel_handle_slot() -> &'static RwLock<Option<Arc<RwLock<Option<String>>>>> {
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<RwLock<Option<String>>>) {
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<String>) {
// 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<AtomicUsize>,
/// 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<String>,
}
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<AtomicUsize>) -> 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<AtomicUsize>) -> 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<String> {
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<dyn Channel> = Arc::new(MockChannel {
name: "telegram".into(),
send_count: Arc::clone(&send_count),
});
let ch: Arc<dyn Channel> = Arc::new(MockChannel::new("telegram", Arc::clone(&send_count)));
let map: HashMap<String, Arc<dyn Channel>> = [("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<dyn Channel> = Arc::new(MockChannel::without_target(
"telegram",
Arc::clone(&send_count),
));
let map: HashMap<String, Arc<dyn Channel>> = [("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<dyn Channel> = Arc::new(MockChannel {
name: "telegram".into(),
send_count: Arc::clone(&send_count),
});
let ch: Arc<dyn Channel> = Arc::new(MockChannel::new("telegram", Arc::clone(&send_count)));
let map: HashMap<String, Arc<dyn Channel>> = [("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<dyn Channel> = Arc::new(MockChannel {
name: "telegram".into(),
send_count: Arc::clone(&send_count),
});
let ch: Arc<dyn Channel> = Arc::new(MockChannel::new("telegram", Arc::clone(&send_count)));
let map: HashMap<String, Arc<dyn Channel>> = [("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<dyn Channel> = Arc::new(MockChannel {
name: "discord".into(),
send_count: Arc::clone(&send_count),
});
let ch: Arc<dyn Channel> = Arc::new(MockChannel::new("discord", Arc::clone(&send_count)));
let map: HashMap<String, Arc<dyn Channel>> = [("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<dyn Channel> = Arc::new(MockChannel {
name: "telegram".into(),
send_count: Arc::clone(&send_count),
});
let ch: Arc<dyn Channel> = Arc::new(MockChannel::new("telegram", Arc::clone(&send_count)));
let map: HashMap<String, Arc<dyn Channel>> = [("telegram".into(), ch)].into();
let sub = ProactiveMessageSubscriber::new(Arc::new(map), Some("telegram".into()));
@@ -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<String> {
@@ -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<String> {
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
@@ -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);
}
@@ -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<String>) -> 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<String>) {
if let Some((chat_id, thread_id)) = reply_target.split_once(':') {
@@ -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<String> {
self.chat_id
.as_deref()
.map(str::trim)
.filter(|s| !s.is_empty())
.map(str::to_string)
}
fn supports_reactions(&self) -> bool {
true
}
@@ -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);
@@ -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<String>,
/// 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.
+41 -9
View File
@@ -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<dyn Memory> = 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<dyn Memory> = 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(
+17
View File
@@ -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<String> {
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]
+1
View File
@@ -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,
+5
View File
@@ -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<String>,
pub allowed_users: Vec<String>,
#[serde(default)]
pub stream_mode: StreamMode,
@@ -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,
@@ -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,
+1 -1
View File
@@ -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(),
+1
View File
@@ -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,
@@ -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<MigrationStats> {
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;
@@ -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);
}
+56 -1
View File
@@ -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)]
+38 -11
View File
@@ -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}"
);
}
@@ -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<String>) -> Json<Value> {
async fn mock_consume_login_token(Json(body): Json<Value>) -> Json<Value> {
// 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,
+1
View File
@@ -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,
+1
View File
@@ -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,
@@ -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<String> {
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