diff --git a/src/__tests__/frontend-scoping.test.ts b/src/__tests__/frontend-scoping.test.ts new file mode 100644 index 000000000..bb5a0bb0d --- /dev/null +++ b/src/__tests__/frontend-scoping.test.ts @@ -0,0 +1,69 @@ +/** + * Frontend tool-server scoping — a chat gets its OWNING frontend's + * tool server, not one per configured frontend. + * + * The bug this pins down: with `frontend: ["telegram", "native"]`, a + * native-app conversation used to register BOTH `native-tools` and + * `telegram-tools`, so the model saw `mcp__telegram-tools__*` names in + * the native client — confusing UI, and an invitation to deliver a + * reply to the wrong surface. + */ + +import { describe, it, expect } from "vitest"; +import { + frontendForChatId, + frontendsForChat, + nonTerminalFrontends, +} from "../backend/shared/frontends.js"; + +const ALL = ["telegram", "discord", "teams", "native"] as const; + +describe("frontendForChatId", () => { + it("classifies each frontend's chat-id shape", () => { + expect(frontendForChatId("d_1751640000000_ab12cd")).toBe("native"); + expect(frontendForChatId("teams_chat_19:abc@thread.v2")).toBe("teams"); + expect(frontendForChatId("discord_guild_123_456")).toBe("discord"); + expect(frontendForChatId("discord_dm_789")).toBe("discord"); + expect(frontendForChatId("-1001426819337")).toBe("telegram"); + expect(frontendForChatId("123456789")).toBe("telegram"); + }); + + it("returns null for cross-surface contexts", () => { + expect(frontendForChatId("heartbeat")).toBeNull(); + expect(frontendForChatId("t_1751640000000")).toBeNull(); + }); +}); + +describe("frontendsForChat", () => { + it("scopes an owned chat to exactly its frontend", () => { + expect(frontendsForChat("d_1751640000000_ab12cd", [...ALL])).toEqual([ + "native", + ]); + expect(frontendsForChat("-100123", [...ALL])).toEqual(["telegram"]); + expect(frontendsForChat("discord_dm_789", [...ALL])).toEqual(["discord"]); + }); + + it("keeps the full set for heartbeat / terminal / unknown chats", () => { + expect(frontendsForChat("heartbeat", [...ALL])).toEqual([...ALL]); + expect(frontendsForChat("t_1751640000000", [...ALL])).toEqual([...ALL]); + }); + + it("falls back to the full set when the owner is not configured", () => { + // A native chat id while only telegram is configured — mis-config + // or migration; better every tool than none. + expect(frontendsForChat("d_1751640000000_ab12cd", ["telegram"])).toEqual([ + "telegram", + ]); + }); +}); + +describe("nonTerminalFrontends", () => { + it("normalises string/array and drops terminal", () => { + expect(nonTerminalFrontends("telegram")).toEqual(["telegram"]); + expect(nonTerminalFrontends(["telegram", "terminal", "native"])).toEqual([ + "telegram", + "native", + ]); + expect(nonTerminalFrontends(undefined)).toEqual([]); + }); +}); diff --git a/src/__tests__/scheduled-messages.test.ts b/src/__tests__/scheduled-messages.test.ts new file mode 100644 index 000000000..2161502e3 --- /dev/null +++ b/src/__tests__/scheduled-messages.test.ts @@ -0,0 +1,274 @@ +/** + * Persistent scheduled messages — the store (storage/scheduled-store) + * and the Telegram schedule/cancel/list/restore handlers. + * + * The behaviour pinned here is the whole point of the feature: a + * scheduled send survives a restart. Before this, `send(..., + * delay_seconds=N)` lived only in an in-process setTimeout — the model + * promises "reminder in 30 minutes", Talon restarts, and the promise + * silently evaporates. + */ + +import { describe, it, expect, beforeEach, afterEach, vi } from "vitest"; +import { mkdtempSync, rmSync } from "node:fs"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; + +vi.mock("../util/log.js", () => ({ + log: vi.fn(), + logError: vi.fn(), + logWarn: vi.fn(), + logDebug: vi.fn(), +})); + +const envBackup = process.env.TALON_DB_PATH; +let tempHome: string; +let closeDb: (() => void) | null = null; + +async function freshImports() { + // Re-import per test so module-scoped DB handles pick up the + // per-test TALON_DB_PATH. + vi.resetModules(); + const db = await import("../storage/db.js"); + closeDb = db.closeDatabase; + const store = await import("../storage/scheduled-store.js"); + const telegram = await import("../frontend/telegram/actions/messaging.js"); + return { store, telegram }; +} + +/** Minimal fake grammY bot capturing sendMessage calls. */ +function fakeBot() { + const sent: Array<{ chatId: number; text: string; opts?: unknown }> = []; + return { + sent, + api: { + sendMessage: vi.fn( + async (chatId: number, text: string, opts?: unknown) => { + sent.push({ chatId, text, opts }); + return { message_id: sent.length }; + }, + ), + }, + }; +} + +type Ctx = { + bot: ReturnType; + scheduledMessages: Map>; +}; +function fakeCtx(): Ctx { + return { bot: fakeBot(), scheduledMessages: new Map() }; +} + +beforeEach(() => { + tempHome = mkdtempSync(join(tmpdir(), "talon-sched-")); + process.env.TALON_DB_PATH = join(tempHome, "talon.db"); + vi.useFakeTimers(); +}); + +afterEach(() => { + vi.useRealTimers(); + closeDb?.(); + closeDb = null; + if (envBackup === undefined) delete process.env.TALON_DB_PATH; + else process.env.TALON_DB_PATH = envBackup; + rmSync(tempHome, { recursive: true, force: true }); +}); + +describe("scheduled-store", () => { + it("saves, lists (soonest first, per frontend/chat), and deletes", async () => { + const { store } = await freshImports(); + const base = { + frontend: "telegram", + text: "hi", + createdAt: Date.now(), + }; + store.saveScheduled({ ...base, id: "b", chatId: "1", fireAt: 2000 }); + store.saveScheduled({ ...base, id: "a", chatId: "1", fireAt: 1000 }); + store.saveScheduled({ ...base, id: "c", chatId: "2", fireAt: 500 }); + store.saveScheduled({ + ...base, + id: "d", + chatId: "1", + fireAt: 100, + frontend: "discord", + }); + + expect(store.listScheduled("telegram").map((e) => e.id)).toEqual([ + "c", + "a", + "b", + ]); + expect( + store.listScheduledForChat("telegram", "1").map((e) => e.id), + ).toEqual(["a", "b"]); + expect(store.deleteScheduled("a")).toBe(true); + expect(store.deleteScheduled("a")).toBe(false); + expect( + store.listScheduledForChat("telegram", "1").map((e) => e.id), + ).toEqual(["b"]); + }); +}); + +describe("telegram schedule_message", () => { + it("persists the entry, fires it, and cleans up store + timer map", async () => { + const { store, telegram } = await freshImports(); + const ctx = fakeCtx(); + + const res = (await telegram.messagingHandlers.schedule_message( + { text: "future ping", delay_seconds: 60 }, + 42, + ctx as never, + )) as { ok: boolean; schedule_id: string }; + expect(res.ok).toBe(true); + + // Persisted and armed. + expect(store.listScheduledForChat("telegram", "42")).toHaveLength(1); + expect(ctx.scheduledMessages.size).toBe(1); + + await vi.advanceTimersByTimeAsync(60_000); + expect(ctx.bot.sent).toHaveLength(1); + expect(ctx.bot.sent[0].text).toContain("future ping"); + // Fired → gone from both store and map. + expect(store.listScheduledForChat("telegram", "42")).toHaveLength(0); + expect(ctx.scheduledMessages.size).toBe(0); + }); + + it("cancel_scheduled clears both the timer and the persisted entry", async () => { + const { store, telegram } = await freshImports(); + const ctx = fakeCtx(); + const res = (await telegram.messagingHandlers.schedule_message( + { text: "never", delay_seconds: 60 }, + 42, + ctx as never, + )) as unknown as { schedule_id: string }; + + const cancel = await telegram.messagingHandlers.cancel_scheduled( + { schedule_id: res.schedule_id }, + 42, + ctx as never, + ); + expect(cancel).toMatchObject({ ok: true, cancelled: true }); + expect(store.listScheduled("telegram")).toHaveLength(0); + + await vi.advanceTimersByTimeAsync(120_000); + expect(ctx.bot.sent).toHaveLength(0); + }); + + it("cancel_scheduled works on a store-only entry (scheduled before a restart)", async () => { + const { store, telegram } = await freshImports(); + store.saveScheduled({ + id: "sched_orphan", + frontend: "telegram", + chatId: "42", + text: "pre-restart", + fireAt: Date.now() + 60_000, + createdAt: Date.now(), + }); + const ctx = fakeCtx(); // fresh map — no live timer for the entry + const cancel = await telegram.messagingHandlers.cancel_scheduled( + { schedule_id: "sched_orphan" }, + 42, + ctx as never, + ); + expect(cancel).toMatchObject({ ok: true, cancelled: true }); + expect(store.listScheduled("telegram")).toHaveLength(0); + }); + + it("list_scheduled reports pending sends for this chat only", async () => { + const { telegram } = await freshImports(); + const ctx = fakeCtx(); + await telegram.messagingHandlers.schedule_message( + { text: "mine", delay_seconds: 120 }, + 42, + ctx as never, + ); + await telegram.messagingHandlers.schedule_message( + { text: "other chat", delay_seconds: 120 }, + 99, + ctx as never, + ); + const res = (await telegram.messagingHandlers.list_scheduled( + {}, + 42, + ctx as never, + )) as { ok: boolean; count: number; scheduled: Array<{ text: string }> }; + expect(res.count).toBe(1); + expect(res.scheduled[0].text).toBe("mine"); + }); +}); + +describe("restoreScheduledMessages (post-restart)", () => { + it("re-arms future entries so they still fire", async () => { + const { store, telegram } = await freshImports(); + store.saveScheduled({ + id: "sched_future", + frontend: "telegram", + chatId: "42", + text: "survived the restart", + fireAt: Date.now() + 30_000, + createdAt: Date.now(), + }); + + const ctx = fakeCtx(); + telegram.restoreScheduledMessages(ctx.bot as never, ctx.scheduledMessages); + expect(ctx.scheduledMessages.size).toBe(1); + + await vi.advanceTimersByTimeAsync(30_000); + expect(ctx.bot.sent).toHaveLength(1); + expect(ctx.bot.sent[0].text).toContain("survived the restart"); + expect(store.listScheduled("telegram")).toHaveLength(0); + }); + + it("fires overdue entries immediately (late beats never)", async () => { + const { store, telegram } = await freshImports(); + store.saveScheduled({ + id: "sched_overdue", + frontend: "telegram", + chatId: "42", + text: "sorry I'm late", + fireAt: Date.now() - 5_000, + createdAt: Date.now() - 65_000, + }); + + const ctx = fakeCtx(); + telegram.restoreScheduledMessages(ctx.bot as never, ctx.scheduledMessages); + await vi.advanceTimersByTimeAsync(0); + expect(ctx.bot.sent).toHaveLength(1); + expect(store.listScheduled("telegram")).toHaveLength(0); + }); + + it("drops entries stale past the overdue cap instead of spamming", async () => { + const { store, telegram } = await freshImports(); + store.saveScheduled({ + id: "sched_ancient", + frontend: "telegram", + chatId: "42", + text: "week-old reminder", + fireAt: Date.now() - 8 * 24 * 60 * 60 * 1000, + createdAt: Date.now() - 8 * 24 * 60 * 60 * 1000, + }); + + const ctx = fakeCtx(); + telegram.restoreScheduledMessages(ctx.bot as never, ctx.scheduledMessages); + await vi.advanceTimersByTimeAsync(0); + expect(ctx.bot.sent).toHaveLength(0); + expect(store.listScheduled("telegram")).toHaveLength(0); + }); + + it("leaves other frontends' entries alone", async () => { + const { store, telegram } = await freshImports(); + store.saveScheduled({ + id: "sched_discord", + frontend: "discord", + chatId: "42", + text: "not yours", + fireAt: Date.now() + 30_000, + createdAt: Date.now(), + }); + const ctx = fakeCtx(); + telegram.restoreScheduledMessages(ctx.bot as never, ctx.scheduledMessages); + expect(ctx.scheduledMessages.size).toBe(0); + expect(store.listScheduled("discord")).toHaveLength(1); + }); +}); diff --git a/src/__tests__/shared-system-prompt.test.ts b/src/__tests__/shared-system-prompt.test.ts index f403ee7d4..4e15d1b31 100644 --- a/src/__tests__/shared-system-prompt.test.ts +++ b/src/__tests__/shared-system-prompt.test.ts @@ -15,7 +15,11 @@ import { clearSystemPromptSnapshots, prepareSystemPrompt, } from "../backend/shared/system-prompt.js"; -import { rebuildSystemPrompt, type TalonConfig } from "../util/config.js"; +import { + rebuildSystemPrompt, + buildSystemPromptPartsFor, + type TalonConfig, +} from "../util/config.js"; // Each rebuild stamps a fresh version into the config — simulating the // real volatility (workspace listing sizes, daily-memory pointer) that @@ -34,6 +38,12 @@ vi.mock("../util/config.js", async (importOriginal) => { }; config.systemPrompt = `static v${rebuildCount}\n\n---\n\ndynamic v${rebuildCount}`; }), + buildSystemPromptPartsFor: vi.fn( + (_config: unknown, _additions: unknown, frontend: string) => ({ + staticText: `static for ${frontend}`, + dynamicText: `dynamic for ${frontend}`, + }), + ), }; }); @@ -243,3 +253,92 @@ describe("appendBackendSuffix", () => { expect(appendBackendSuffix(undefined, "x")).toBe("x"); }); }); + +describe("per-chat frontend prompt flavour (multi-frontend)", () => { + beforeEach(() => { + clearSystemPromptSnapshots(); + rebuildCount = 0; + vi.mocked(rebuildSystemPrompt).mockClear(); + vi.mocked(buildSystemPromptPartsFor).mockClear(); + }); + + function multiConfig(): TalonConfig { + return { + frontend: ["telegram", "native"], + systemPrompt: "initial", + systemPromptParts: { staticText: "initial", dynamicText: "" }, + } as unknown as TalonConfig; + } + + it("builds a native chat's prompt for the native frontend, not frontends[0]", () => { + const config = multiConfig(); + const prepared = prepareSystemPrompt({ + config, + previousTurns: 0, + chatId: "d_1751640000000_ab12cd", + sessionEpoch: 1000, + }); + + expect(prepared.staticText).toBe("static for native"); + expect(buildSystemPromptPartsFor).toHaveBeenCalledWith( + config, + [], + "native", + ); + // Non-primary flavour must NOT clobber the global config prompt. + expect(rebuildSystemPrompt).not.toHaveBeenCalled(); + expect(config.systemPrompt).toBe("initial"); + }); + + it("keeps the primary-frontend path (global rebuild) for telegram chats", () => { + const config = multiConfig(); + const prepared = prepareSystemPrompt({ + config, + previousTurns: 0, + chatId: "-1001426819337", + sessionEpoch: 1000, + }); + + expect(rebuildSystemPrompt).toHaveBeenCalledTimes(1); + expect(prepared.staticText).toBe("static v1"); + }); + + it("falls back to the primary frontend for cross-surface chat ids", () => { + const config = multiConfig(); + prepareSystemPrompt({ + config, + previousTurns: 0, + chatId: "heartbeat", + sessionEpoch: 1000, + }); + expect(rebuildSystemPrompt).toHaveBeenCalledTimes(1); + }); + + it("freezes each chat's flavour independently per session", () => { + const config = multiConfig(); + const native1 = prepareSystemPrompt({ + config, + previousTurns: 0, + chatId: "d_1751640000000_ab12cd", + sessionEpoch: 1000, + }); + const tg1 = prepareSystemPrompt({ + config, + previousTurns: 0, + chatId: "-100123", + sessionEpoch: 1000, + }); + // Second turns return the frozen snapshots, no extra builds. + const native2 = prepareSystemPrompt({ + config, + previousTurns: 1, + chatId: "d_1751640000000_ab12cd", + sessionEpoch: 1000, + }); + expect(native2).toBe(native1); + expect(native1.staticText).toBe("static for native"); + expect(tg1.staticText).toBe("static v1"); + expect(buildSystemPromptPartsFor).toHaveBeenCalledTimes(1); + expect(rebuildSystemPrompt).toHaveBeenCalledTimes(1); + }); +}); diff --git a/src/backend/claude-sdk/handler.ts b/src/backend/claude-sdk/handler.ts index f0910dfef..905102239 100644 --- a/src/backend/claude-sdk/handler.ts +++ b/src/backend/claude-sdk/handler.ts @@ -35,6 +35,7 @@ import { makeBareModelRef } from "../../core/agent-runtime/model-ref.js"; import { applyRetryDecisionStream } from "../shared/handle-retry.js"; import { getConfig } from "./state.js"; import { buildSdkOptions, getActiveFrontends } from "./options.js"; +import { frontendsForChat } from "../shared/frontends.js"; import { createStreamState, isSystemInit, @@ -127,11 +128,18 @@ export async function* runChatTurn( const session = getSession(chatId); const t0 = Date.now(); - // Primary messaging frontend, if any. Drives the delivery-contract - // suffix and the frontend-aware flow-violation text. Empty in - // terminal mode, where no delivery tools exist and the strict - // tool-only contract must not be asserted. - const frontend: string | undefined = getActiveFrontends()[0]; + // The chat's OWNING messaging frontend (falling back to the primary + // for cross-surface chats). Drives the delivery-contract suffix and + // the frontend-aware flow-violation text — the tool NAMES differ per + // frontend (native's send tool is send_message, telegram's is send), + // and with tool servers scoped per chat the wrong contract would + // instruct a tool that doesn't exist. Empty in terminal mode, where + // no delivery tools exist and the strict tool-only contract must not + // be asserted. + const frontend: string | undefined = frontendsForChat( + chatId, + getActiveFrontends(), + )[0]; // Frozen per-session prompt (keyed by session epoch) — stable across // turns so the provider's prompt-cache prefix survives other chats' diff --git a/src/backend/claude-sdk/options.ts b/src/backend/claude-sdk/options.ts index ff5c80963..905816be4 100644 --- a/src/backend/claude-sdk/options.ts +++ b/src/backend/claude-sdk/options.ts @@ -25,7 +25,7 @@ import { pluginHubUrl, hubPluginServerNames, } from "../../core/mcp-hub/index.js"; -import { nonTerminalFrontends } from "../shared/frontends.js"; +import { nonTerminalFrontends, frontendsForChat } from "../shared/frontends.js"; import { log, logError } from "../../util/log.js"; import { getConfig, getBridgePort } from "./state.js"; import { ALLOWED_TOOLS_CHAT, EFFORT_MAP } from "./constants.js"; @@ -78,7 +78,12 @@ export function buildMcpServers(chatId: string): Record { const servers: Record = {}; - for (const frontend of getActiveFrontends()) { + // Scope to the chat's owning frontend: a native-app chat gets + // native-tools only, not every configured frontend's server (stray + // mcp__telegram-tools__* in a native chat is confusing UI and a + // wrong-surface delivery hazard). Cross-surface contexts (heartbeat, + // one-shots) keep the full set. + for (const frontend of frontendsForChat(chatId, getActiveFrontends())) { servers[`${frontend}-tools`] = { type: "http", url: talonHubUrl(bridgeUrl, frontend, chatId), diff --git a/src/backend/codex/constants.ts b/src/backend/codex/constants.ts index 65ed0f2bc..109eb1006 100644 --- a/src/backend/codex/constants.ts +++ b/src/backend/codex/constants.ts @@ -15,13 +15,16 @@ import { buildDeliveryContract } from "../shared/delivery-contract.js"; * * Both routes are valid — the shared text-or-tools contract * (prompts/system/contract-text-or-tools.md) documents the choice. - * Telegram-shaped tool names: this constant is static and the prior - * hand-written text hardcoded the same names. + * Tool names are frontend-specific (native's send tool is + * send_message, telegram's is send), so the suffix is built per chat + * from the chat's owning frontend. */ -export const CODEX_SYSTEM_PROMPT_SUFFIX = `\n\n${buildDeliveryContract( - "text-or-tools", - "telegram", -)}\n`; +export function codexSystemPromptSuffix(frontend: string): string { + return `\n\n${buildDeliveryContract("text-or-tools", frontend)}\n`; +} + +/** Telegram-shaped default, kept for tests and legacy callers. */ +export const CODEX_SYSTEM_PROMPT_SUFFIX = codexSystemPromptSuffix("telegram"); /** * Default model used by the Codex backend when none is configured AND diff --git a/src/backend/codex/handler/message.ts b/src/backend/codex/handler/message.ts index 7c8719e4e..e764ca4b2 100644 --- a/src/backend/codex/handler/message.ts +++ b/src/backend/codex/handler/message.ts @@ -42,12 +42,16 @@ import { } from "../../shared/index.js"; import { - CODEX_SYSTEM_PROMPT_SUFFIX, + codexSystemPromptSuffix, CODEX_DEFAULT_MODEL, CODEX_CHATGPT_DEFAULT_MODEL, CODEX_THREAD_PERMISSIONS, CODEX_LIVE_POLL_INTERVAL_MS, } from "../constants.js"; +import { + frontendsForChat, + nonTerminalFrontends, +} from "../../shared/frontends.js"; import { getState } from "../state.js"; import { ensureCodex, getCodexAuthInfo } from "../init.js"; import { @@ -242,7 +246,10 @@ export async function handleMessage( const { text: systemPrompt } = prepareSystemPrompt({ config, previousTurns, - backendSuffix: CODEX_SYSTEM_PROMPT_SUFFIX, + backendSuffix: codexSystemPromptSuffix( + frontendsForChat(chatId, nonTerminalFrontends(config.frontend))[0] ?? + "telegram", + ), chatId, sessionEpoch: session.createdAt, }); diff --git a/src/backend/codex/mcp-config.ts b/src/backend/codex/mcp-config.ts index 244056309..414cafcad 100644 --- a/src/backend/codex/mcp-config.ts +++ b/src/backend/codex/mcp-config.ts @@ -30,6 +30,7 @@ import { hubPluginServerNames, } from "../../core/mcp-hub/index.js"; import type { ToolExclusionConfig } from "../../core/tools/mcp-env.js"; +import { frontendsForChat } from "../shared/frontends.js"; /** * AppToolApproval values accepted by Codex's `mcp_servers.` table. @@ -131,10 +132,11 @@ export function buildCodexMcpServers(args: { const servers: Record = {}; - // Frontend MCP tool servers (one per non-terminal frontend) — served - // in-process by the hub; the (frontend, chatId) binding travels in - // the URL. Tool-surface trimming is applied hub-side (initHub). - for (const frontend of frontends) { + // Frontend MCP tool servers — served in-process by the hub; the + // (frontend, chatId) binding travels in the URL. Tool-surface + // trimming is applied hub-side (initHub). Scoped to the chat's + // owning frontend; cross-surface contexts keep the full set. + for (const frontend of frontendsForChat(chatId, frontends)) { servers[`${frontend}-tools`] = { url: talonHubUrl(bridgeUrl, frontend, chatId), default_tools_approval_mode: TALON_MCP_DEFAULT_APPROVAL, diff --git a/src/backend/kilo/handler/message.ts b/src/backend/kilo/handler/message.ts index 03a2954f2..c30710f25 100644 --- a/src/backend/kilo/handler/message.ts +++ b/src/backend/kilo/handler/message.ts @@ -26,8 +26,12 @@ import { resolveProviderID, parseStoredKiloModelSelection, getConfig, - KILO_SYSTEM_PROMPT_SUFFIX, + kiloSystemPromptSuffix, } from "../server.js"; +import { + frontendsForChat, + nonTerminalFrontends, +} from "../../shared/frontends.js"; import { getKiloTurnSummary } from "../sessions.js"; import { createStreamState, @@ -99,7 +103,10 @@ export async function handleMessage( const { text: systemPrompt } = prepareSystemPrompt({ config, previousTurns, - backendSuffix: KILO_SYSTEM_PROMPT_SUFFIX, + backendSuffix: kiloSystemPromptSuffix( + frontendsForChat(chatId, nonTerminalFrontends(config.frontend))[0] ?? + "telegram", + ), chatId, sessionEpoch: session.createdAt, }); diff --git a/src/backend/kilo/server.ts b/src/backend/kilo/server.ts index d32b14af4..078259cb0 100644 --- a/src/backend/kilo/server.ts +++ b/src/backend/kilo/server.ts @@ -81,10 +81,12 @@ export const TALON_MCP_SERVER_NAME = SHARED_TALON_MCP_SERVER_NAME; * Telegram-shaped tool names: this constant is static and the prior * hand-written text hardcoded the same names. */ -export const KILO_SYSTEM_PROMPT_SUFFIX = `\n\n${buildDeliveryContract( - "text-or-tools", - "telegram", -)}\n`; +export function kiloSystemPromptSuffix(frontend: string): string { + return `\n\n${buildDeliveryContract("text-or-tools", frontend)}\n`; +} + +/** Telegram-shaped default, kept for one-shot (cross-surface) paths. */ +export const KILO_SYSTEM_PROMPT_SUFFIX = kiloSystemPromptSuffix("telegram"); // ── State ─────────────────────────────────────────────────────────────────── diff --git a/src/backend/openai-agents/handler/message.ts b/src/backend/openai-agents/handler/message.ts index 64936eac2..a48677602 100644 --- a/src/backend/openai-agents/handler/message.ts +++ b/src/backend/openai-agents/handler/message.ts @@ -52,6 +52,7 @@ import { } from "../constants.js"; import { getState, getOrCreateSession } from "../state.js"; import { getActiveFrontends } from "../init.js"; +import { frontendsForChat } from "../../shared/frontends.js"; import { getOrCreateBundle } from "../mcp-pool.js"; import { OPENAI_AGENTS_BUILTIN_TOOLS } from "../builtins.js"; import { activeAborts } from "./state.js"; @@ -121,11 +122,13 @@ export async function handleMessage( OPENAI_AGENTS_DEFAULT_MODEL; log("agent", `[${chatId}] OpenAI Agents model resolved: ${activeModel}`); - // Primary messaging frontend drives the delivery-contract suffix and - // the frontend-aware flow-violation/first-turn text. Empty in terminal - // mode (no delivery tools — contract enforcement is skipped below anyway). + // The chat's OWNING messaging frontend (falling back to the primary + // for cross-surface chats) drives the delivery-contract suffix and + // the frontend-aware flow-violation/first-turn text — tool names + // differ per frontend. Empty in terminal mode (no delivery tools — + // contract enforcement is skipped below anyway). const frontends = getActiveFrontends(); - const frontend: string | undefined = frontends[0]; + const frontend: string | undefined = frontendsForChat(chatId, frontends)[0]; // Per-session frozen prompt + Agents-specific delivery suffix. const { text: systemPrompt } = prepareSystemPrompt({ diff --git a/src/backend/openai-agents/mcp-pool.ts b/src/backend/openai-agents/mcp-pool.ts index 8665a227f..151f87179 100644 --- a/src/backend/openai-agents/mcp-pool.ts +++ b/src/backend/openai-agents/mcp-pool.ts @@ -31,6 +31,7 @@ import { hubPluginServerNames, } from "../../core/mcp-hub/index.js"; import type { ToolExclusionConfig } from "../../core/tools/mcp-env.js"; +import { frontendsForChat } from "../shared/frontends.js"; import { log, logWarn } from "../../util/log.js"; /** @@ -173,11 +174,12 @@ async function buildBundle(args: BundleInputs): Promise { const built: MCPServerStreamableHttp[] = []; - // Frontend MCP tool servers (one per non-terminal frontend). Each - // exposes the Talon-native delivery surface (send, react, end_turn, - // …) for that frontend, scoped to `chatId` via the hub URL — the - // server itself runs in-process inside the daemon. - for (const frontend of frontends) { + // Frontend MCP tool servers. Each exposes the Talon-native delivery + // surface (send, react, end_turn, …) for that frontend, scoped to + // `chatId` via the hub URL — the server itself runs in-process + // inside the daemon. Scoped to the chat's owning frontend; + // cross-surface contexts keep the full set. + for (const frontend of frontendsForChat(chatId, frontends)) { built.push( new MCPServerStreamableHttp({ name: `${frontend}-tools`, diff --git a/src/backend/opencode/handler/message.ts b/src/backend/opencode/handler/message.ts index 8b60f6532..a585f7358 100644 --- a/src/backend/opencode/handler/message.ts +++ b/src/backend/opencode/handler/message.ts @@ -27,8 +27,12 @@ import { resolveProviderID, parseStoredOpenCodeModelSelection, getConfig, - OPENCODE_SYSTEM_PROMPT_SUFFIX, + opencodeSystemPromptSuffix, } from "../server.js"; +import { + frontendsForChat, + nonTerminalFrontends, +} from "../../shared/frontends.js"; import { getOpenCodeTurnSummary } from "../sessions.js"; import { createStreamState, @@ -96,7 +100,10 @@ export async function handleMessage( const { text: systemPrompt } = prepareSystemPrompt({ config, previousTurns, - backendSuffix: OPENCODE_SYSTEM_PROMPT_SUFFIX, + backendSuffix: opencodeSystemPromptSuffix( + frontendsForChat(chatId, nonTerminalFrontends(config.frontend))[0] ?? + "telegram", + ), chatId, sessionEpoch: session.createdAt, }); diff --git a/src/backend/opencode/server.ts b/src/backend/opencode/server.ts index 09458d091..378fe978c 100644 --- a/src/backend/opencode/server.ts +++ b/src/backend/opencode/server.ts @@ -59,10 +59,12 @@ const TALON_MCP_SERVER_NAME = SHARED_TALON_MCP_SERVER_NAME; // Text-preferred delivery: plain assistant text is the reply; tools // only for genuine side effects. Single-sourced from the shared // contract templates (prompts/system/contract-text-preferred.md). -const OPENCODE_SYSTEM_PROMPT_SUFFIX = `\n\n${buildDeliveryContract( - "text-preferred", - "telegram", -)}\n`; +export function opencodeSystemPromptSuffix(frontend: string): string { + return `\n\n${buildDeliveryContract("text-preferred", frontend)}\n`; +} + +/** Telegram-shaped default, kept for one-shot (cross-surface) paths. */ +const OPENCODE_SYSTEM_PROMPT_SUFFIX = opencodeSystemPromptSuffix("telegram"); // ── State ─────────────────────────────────────────────────────────────────── diff --git a/src/backend/remote-server/mcp.ts b/src/backend/remote-server/mcp.ts index 73e84bec1..686390799 100644 --- a/src/backend/remote-server/mcp.ts +++ b/src/backend/remote-server/mcp.ts @@ -43,6 +43,12 @@ import { import type { RemoteAgentClient } from "./client.js"; import type { RemoteServerState } from "./state.js"; import { errMsg } from "./state.js"; +import type { TalonConfig } from "../../util/config.js"; +import type { FrontendName } from "../../core/agent-runtime/backend-registry.js"; +import { + frontendForChatId, + nonTerminalFrontends, +} from "../shared/frontends.js"; // ── Constants ─────────────────────────────────────────────────────────────── @@ -77,6 +83,27 @@ export function isTalonToolID(toolID: string): boolean { ); } +/** + * The frontend whose tool surface a chat should get: its owning + * frontend (by chat-id shape) when that frontend is configured, else + * the process-primary frontend — which also covers the heartbeat + * sentinel and other cross-surface contexts. + */ +function resolveChatToolFrontend( + state: { frontendName: FrontendName; config: TalonConfig | null }, + chatId: string, +): FrontendName { + const owner = frontendForChatId(chatId); + if ( + owner && + owner !== state.frontendName && + nonTerminalFrontends(state.config?.frontend).includes(owner) + ) { + return owner as FrontendName; + } + return state.frontendName; +} + // ── Public API ───────────────────────────────────────────────────────────── /** @@ -146,9 +173,13 @@ export async function ensureChatMcpServer( // streamable HTTP — no subprocess on the agent-server side, and // the (frontend, chatId) binding travels in the URL instead of // env. Tool-surface trimming is applied hub-side (initHub). + // Bind the chat's OWNING frontend, not the process-primary one: + // in multi-frontend deployments a native chat served by this + // backend must get native-tools (send_message/end_turn), not + // the primary frontend's tool surface. url: talonHubUrl( `http://127.0.0.1:${state.gatewayPortFn()}`, - state.frontendName, + resolveChatToolFrontend(state, chatId), chatId, ), }, diff --git a/src/backend/shared/frontends.ts b/src/backend/shared/frontends.ts index 27cefb38b..f147a0c59 100644 --- a/src/backend/shared/frontends.ts +++ b/src/backend/shared/frontends.ts @@ -6,6 +6,13 @@ * each carried their own copy of the same normalise-and-filter logic. */ +import { + isNativeChatId, + isTeamsChatId, + isDiscordChatId, + isTelegramChatId, +} from "../../util/chat-id.js"; + /** * Normalise a config `frontend` value (string or array, possibly * undefined) to the list of messaging frontends — i.e. those that @@ -19,3 +26,41 @@ export function nonTerminalFrontends( const all = Array.isArray(frontend) ? frontend : [frontend as string]; return all.filter((f) => f !== "terminal"); } + +/** + * The messaging frontend that owns a chat, inferred from the chat-id + * shape (the same convention the gateway uses to route actions): + * native `d_*`, teams `teams_chat_*`, discord `discord_*`, telegram + * numeric. Returns null for cross-surface contexts — the heartbeat + * sentinel, isolated one-shots, terminal sessions. + */ +export function frontendForChatId(chatId: string): string | null { + if (isNativeChatId(chatId)) return "native"; + if (isTeamsChatId(chatId)) return "teams"; + if (isDiscordChatId(chatId)) return "discord"; + if (isTelegramChatId(chatId)) return "telegram"; + return null; +} + +/** + * Which frontends' tool servers a chat's turn should register. + * + * A chat owned by one messaging frontend gets exactly that frontend's + * tools: a native-app conversation must see `native-tools`, not a + * `telegram-tools` server that happens to share the daemon — the + * model reads surface identity out of tool names, and stray + * `mcp__telegram-tools__*` entries in a native chat are both confusing + * UI and an invitation to deliver the reply to the wrong surface. + * + * Chats with no owning frontend (heartbeat sentinel, one-shot cron, + * terminal sessions) keep every configured messaging frontend: those + * contexts legitimately reach across surfaces via explicit chat_id. + */ +export function frontendsForChat( + chatId: string, + configured: readonly string[], +): readonly string[] { + const owner = frontendForChatId(chatId); + if (owner && configured.includes(owner)) return [owner]; + return configured; +} diff --git a/src/backend/shared/system-prompt.ts b/src/backend/shared/system-prompt.ts index f4acbd950..8458f4942 100644 --- a/src/backend/shared/system-prompt.ts +++ b/src/backend/shared/system-prompt.ts @@ -34,11 +34,14 @@ import { rebuildSystemPrompt, + buildSystemPromptPartsFor, + primaryFrontend, joinSystemPromptParts, type SystemPromptParts, type TalonConfig, } from "../../util/config.js"; import { getPluginPromptAdditions } from "../../core/plugin/index.js"; +import { frontendForChatId, nonTerminalFrontends } from "./frontends.js"; // ── Types ─────────────────────────────────────────────────────────────────── @@ -135,9 +138,26 @@ export function prepareSystemPrompt( } // New session (or first touch after restart / backend switch): - // rebuild the global template once, then freeze it for this session. - rebuildSystemPrompt(inputs.config, getPluginPromptAdditions()); - const prepared = buildPrepared(inputs.config, suffix); + // build the prompt once for the chat's OWNING frontend, then + // freeze it for this session. In multi-frontend deployments the + // flavours differ (native.md vs telegram.md guidance); building + // for `frontends[0]` gave every chat the primary frontend's + // platform prompt. The primary frontend keeps the historical + // behaviour of also refreshing the global config prompt (warm-up + // and legacy readers consume it). + const chatFrontend = resolveChatFrontend(inputs.config, inputs.chatId); + let prepared: PreparedSystemPrompt; + if (chatFrontend === primaryFrontend(inputs.config)) { + rebuildSystemPrompt(inputs.config, getPluginPromptAdditions()); + prepared = buildPrepared(inputs.config, suffix); + } else { + const parts = buildSystemPromptPartsFor( + inputs.config, + getPluginPromptAdditions(), + chatFrontend, + ); + prepared = buildPreparedFromParts(parts, suffix); + } if (snapshots.size >= MAX_SNAPSHOTS && !snapshots.has(inputs.chatId)) { const oldest = snapshots.keys().next().value; @@ -196,6 +216,14 @@ function buildPrepared( staticText: config.systemPrompt ?? "", dynamicText: "", }; + return buildPreparedFromParts(parts, suffix); +} + +/** Assemble a PreparedSystemPrompt from explicit parts (pure). */ +function buildPreparedFromParts( + parts: SystemPromptParts, + suffix: string, +): PreparedSystemPrompt { const staticText = appendBackendSuffix(parts.staticText, suffix); const dynamicText = parts.dynamicText.trim(); return { @@ -204,3 +232,17 @@ function buildPrepared( dynamicText, }; } + +/** + * The frontend whose prompt flavour a chat should get: its owning + * frontend when the chat-id shape names a configured one, else the + * primary (first configured) frontend — which also covers heartbeat / + * one-shot / terminal contexts. + */ +function resolveChatFrontend(config: TalonConfig, chatId: string): string { + const owner = frontendForChatId(chatId); + if (owner && nonTerminalFrontends(config.frontend).includes(owner)) { + return owner; + } + return primaryFrontend(config); +} diff --git a/src/core/tools/messaging.ts b/src/core/tools/messaging.ts index a08baec80..a7880d2d1 100644 --- a/src/core/tools/messaging.ts +++ b/src/core/tools/messaging.ts @@ -221,7 +221,9 @@ Examples: delay_seconds: z .number() .optional() - .describe("Schedule: delay before sending (1-3600)"), + .describe( + "Schedule: delay before sending, in seconds (1-86400). Scheduled sends persist across restarts; manage with list_scheduled / cancel_scheduled.", + ), chat_id: chatIdSchema .optional() .describe( @@ -243,9 +245,13 @@ Examples: switch (type) { case "text": { if (params.delay_seconds) { + // Buttons and reply threading survive the delay — the + // schedule handler replays them at fire time. return bridge("schedule_message", { text: params.text, delay_seconds: params.delay_seconds, + rows: params.buttons, + reply_to_message_id: params.reply_to, chat_id, }); } diff --git a/src/core/tools/scheduling.ts b/src/core/tools/scheduling.ts index a94fead4b..b171dcf1f 100644 --- a/src/core/tools/scheduling.ts +++ b/src/core/tools/scheduling.ts @@ -15,6 +15,16 @@ export const schedulingTools: ToolDefinition[] = [ tag: "scheduling", }, + { + name: "list_scheduled", + description: + "List pending scheduled messages in this chat (id, seconds until send, text). Scheduled messages persist across restarts.", + schema: {}, + execute: (_params, bridge) => bridge("list_scheduled", {}), + frontends: ["telegram", "discord"], + tag: "scheduling", + }, + { name: "create_cron_job", description: `Create a persistent scheduled job. Jobs survive restarts. diff --git a/src/frontend/discord/actions/index.ts b/src/frontend/discord/actions/index.ts index 03edd2220..b72993d76 100644 --- a/src/frontend/discord/actions/index.ts +++ b/src/frontend/discord/actions/index.ts @@ -24,7 +24,7 @@ import type { Client } from "discord.js"; import type { Gateway } from "../../../core/engine/gateway.js"; import type { ActionResult } from "../../../core/types.js"; import { resolveChannel } from "./shared.js"; -import { messagingHandlers } from "./messaging.js"; +import { messagingHandlers, restoreScheduledMessages } from "./messaging.js"; import { mediaHandlers } from "./media.js"; import { chatInfoHandlers } from "./chat-info.js"; import type { DiscordActionContext, DiscordActionHandlers } from "./types.js"; @@ -40,6 +40,10 @@ const handlers: DiscordActionHandlers = Object.assign(Object.create(null), { export function createDiscordActionHandler(client: Client, gateway: Gateway) { const scheduledMessages = new Map>(); + // Re-arm scheduled sends that were persisted before the last + // shutdown — the timers died with the process, the store didn't. + restoreScheduledMessages(client, scheduledMessages); + return async ( body: Record, chatId: number, @@ -53,7 +57,8 @@ export function createDiscordActionHandler(client: Client, gateway: Gateway) { // For non-channel actions (e.g. cancel_scheduled) that don't need a // resolved channel, fall through. Most Discord actions need it — return // error if missing. - const needsChannel = action !== "cancel_scheduled"; + const needsChannel = + action !== "cancel_scheduled" && action !== "list_scheduled"; if (needsChannel && !channel) { return { ok: false, diff --git a/src/frontend/discord/actions/messaging.ts b/src/frontend/discord/actions/messaging.ts index 7ddd88b4c..93f67e810 100644 --- a/src/frontend/discord/actions/messaging.ts +++ b/src/frontend/discord/actions/messaging.ts @@ -3,13 +3,79 @@ * inline-button messages, typing indicator, and scheduled messages. */ +import type { Client } from "discord.js"; import { withRetry } from "../../../core/engine/gateway.js"; import { sendChunked } from "../handlers/index.js"; import { suppressMentions, DISCORD_MAX_TEXT } from "../formatting.js"; -import { logError } from "../../../util/log.js"; +import { log, logError } from "../../../util/log.js"; +import { + saveScheduled, + deleteScheduled, + listScheduled, + listScheduledForChat, + MAX_OVERDUE_MS, + type ScheduledMessage, +} from "../../../storage/scheduled-store.js"; import { tryAction, resolveChannel, buildButtonRows } from "./shared.js"; import type { DiscordActionHandlers } from "./types.js"; +// ── Scheduled sends (persistent) ───────────────────────────────────────────── + +/** Longest schedulable delay: 24h. Timers re-arm from the store on boot. */ +const MAX_DELAY_SEC = 24 * 60 * 60; + +/** Arm a timer for a stored entry; fires, then cleans up store + map. */ +function armScheduled( + client: Client, + entry: ScheduledMessage, + timers: Map>, +): void { + const delayMs = Math.max(0, entry.fireAt - Date.now()); + const timer = setTimeout(async () => { + try { + const c = await resolveChannel(client, Number(entry.chatId)); + if (c) await sendChunked(c, entry.text); + } catch (err) { + logError( + "discord", + `Scheduled message failed (chat=${entry.chatId})`, + err, + ); + } + deleteScheduled(entry.id); + timers.delete(entry.id); + }, delayMs); + timers.set(entry.id, timer); +} + +/** + * Re-arm persisted scheduled sends after a restart. Overdue entries + * fire immediately — late beats never — unless stale past + * MAX_OVERDUE_MS, which drops them. + */ +export function restoreScheduledMessages( + client: Client, + timers: Map>, +): void { + const entries = listScheduled("discord"); + if (entries.length === 0) return; + let restored = 0; + let dropped = 0; + for (const entry of entries) { + if (Date.now() - entry.fireAt > MAX_OVERDUE_MS) { + deleteScheduled(entry.id); + dropped++; + continue; + } + armScheduled(client, entry, timers); + restored++; + } + log( + "discord", + `Restored ${restored} scheduled message(s) from store${dropped > 0 ? `, dropped ${dropped} stale` : ""}`, + ); +} + export const messagingHandlers: DiscordActionHandlers = { send_message: (body, chatId, { channel, gateway }) => { const text = String(body.text ?? ""); @@ -144,20 +210,21 @@ export const messagingHandlers: DiscordActionHandlers = { const text = String(body.text ?? ""); const delaySec = Math.max( 1, - Math.min(3600, Number(body.delay_seconds ?? 60)), + Math.min(MAX_DELAY_SEC, Number(body.delay_seconds ?? 60)), ); - const scheduleId = `sched_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`; - const timer = setTimeout(async () => { - try { - const c = await resolveChannel(client, chatId); - if (c) await sendChunked(c, text); - } catch (err) { - logError("discord", `Scheduled message failed (chat=${chatId})`, err); - } - scheduledMessages.delete(scheduleId); - }, delaySec * 1000); - scheduledMessages.set(scheduleId, timer); - return { ok: true, schedule_id: scheduleId, delay_seconds: delaySec }; + const entry: ScheduledMessage = { + id: `sched_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`, + frontend: "discord", + chatId: String(chatId), + text, + fireAt: Date.now() + delaySec * 1000, + createdAt: Date.now(), + }; + // Persist before arming: if we crash between the two, the restore + // path delivers it; the reverse order could lose it forever. + saveScheduled(entry); + armScheduled(client, entry, scheduledMessages); + return { ok: true, schedule_id: entry.id, delay_seconds: delaySec }; }, cancel_scheduled: (body, _chatId, { scheduledMessages }) => { @@ -166,9 +233,27 @@ export const messagingHandlers: DiscordActionHandlers = { if (timer) { clearTimeout(timer); scheduledMessages.delete(id); - return { ok: true, cancelled: true }; } - return { ok: false, error: "Schedule not found" }; + // The store is the source of truth — an entry can exist without a + // live timer (scheduled before a restart, not yet re-armed here). + const existed = deleteScheduled(id) || Boolean(timer); + return existed + ? { ok: true, cancelled: true } + : { ok: false, error: "Schedule not found" }; + }, + + list_scheduled: (body, chatId) => { + const pending = listScheduledForChat("discord", String(chatId)).map( + (e) => ({ + schedule_id: e.id, + fires_in_seconds: Math.max( + 0, + Math.round((e.fireAt - Date.now()) / 1000), + ), + text: e.text.length > 200 ? `${e.text.slice(0, 200)}…` : e.text, + }), + ); + return { ok: true, scheduled: pending, count: pending.length }; }, }; diff --git a/src/frontend/discord/actions/types.ts b/src/frontend/discord/actions/types.ts index bedc96872..ef348c88c 100644 --- a/src/frontend/discord/actions/types.ts +++ b/src/frontend/discord/actions/types.ts @@ -6,9 +6,9 @@ * `DiscordActionContext` carrying the client, gateway, per-handler scheduled * timers, and the channel pre-resolved by the dispatcher. * - * `channel` is non-null for every action except `cancel_scheduled` (the only - * action that doesn't need a resolved channel) — the dispatcher guarantees - * this, so handlers can safely use `ctx.channel!`. + * `channel` is non-null for every action except `cancel_scheduled` and + * `list_scheduled` (store-only actions that don't need a resolved channel) — + * the dispatcher guarantees this, so handlers can safely use `ctx.channel!`. */ import type { Client, TextBasedChannel } from "discord.js"; @@ -20,7 +20,7 @@ export interface DiscordActionContext { gateway: Gateway; /** Active scheduled-message timers, keyed by schedule id (for cancellation). */ scheduledMessages: Map>; - /** Channel resolved from the chat id; null only for cancel_scheduled. */ + /** Channel resolved from the chat id; null only for cancel_scheduled / list_scheduled. */ channel: TextBasedChannel | null; } diff --git a/src/frontend/telegram/actions/index.ts b/src/frontend/telegram/actions/index.ts index 873f26c60..4fafb16a0 100644 --- a/src/frontend/telegram/actions/index.ts +++ b/src/frontend/telegram/actions/index.ts @@ -19,7 +19,7 @@ import type { Bot, InputFile as GrammyInputFile } from "grammy"; import type { Gateway } from "../../../core/engine/gateway.js"; import type { ActionResult } from "../../../core/types.js"; -import { messagingHandlers } from "./messaging.js"; +import { messagingHandlers, restoreScheduledMessages } from "./messaging.js"; import { mediaHandlers } from "./media.js"; import { chatInfoHandlers } from "./chat-info.js"; import type { TelegramActionContext, TelegramActionHandlers } from "./types.js"; @@ -52,6 +52,10 @@ export function createTelegramActionHandler( scheduledMessages: new Map>(), }; + // Re-arm scheduled sends that were persisted before the last + // shutdown — the timers died with the process, the store didn't. + restoreScheduledMessages(bot, ctx.scheduledMessages); + return async ( body: Record, chatId: number, diff --git a/src/frontend/telegram/actions/messaging.ts b/src/frontend/telegram/actions/messaging.ts index d97fa9422..d00999da1 100644 --- a/src/frontend/telegram/actions/messaging.ts +++ b/src/frontend/telegram/actions/messaging.ts @@ -3,12 +3,97 @@ * inline-button messages, chat actions, and scheduled messages. */ +import type { Bot } from "grammy"; import { markdownToTelegramHtml } from "../formatting.js"; import { withRetry } from "../../../core/engine/gateway.js"; -import { logError, logWarn } from "../../../util/log.js"; +import { log, logError, logWarn } from "../../../util/log.js"; +import { + saveScheduled, + deleteScheduled, + listScheduled, + listScheduledForChat, + MAX_OVERDUE_MS, + type ScheduledMessage, +} from "../../../storage/scheduled-store.js"; import { sendText, toPositiveId } from "./shared.js"; import { TELEGRAM_MAX_TEXT, type TelegramActionHandlers } from "./types.js"; +// ── Scheduled sends (persistent) ───────────────────────────────────────────── + +/** Longest schedulable delay: 24h. Timers re-arm from the store on boot. */ +const MAX_DELAY_SEC = 24 * 60 * 60; + +/** Deliver one scheduled entry (shared by the live timer and restore). */ +async function fireScheduled(bot: Bot, entry: ScheduledMessage): Promise { + const chatId = Number(entry.chatId); + if (entry.rows) { + const keyboard = entry.rows.map((row) => + row.map((btn) => + btn.url + ? { text: btn.text, url: btn.url } + : { text: btn.text, callback_data: btn.callback_data ?? btn.text }, + ), + ); + await bot.api.sendMessage(chatId, markdownToTelegramHtml(entry.text), { + parse_mode: "HTML", + reply_markup: { inline_keyboard: keyboard }, + reply_parameters: entry.replyTo + ? { message_id: entry.replyTo } + : undefined, + }); + } else { + await sendText(bot, chatId, entry.text, entry.replyTo); + } +} + +/** Arm a timer for a stored entry; fires, then cleans up store + map. */ +function armScheduled( + bot: Bot, + entry: ScheduledMessage, + timers: Map>, +): void { + const delayMs = Math.max(0, entry.fireAt - Date.now()); + const timer = setTimeout(async () => { + try { + await fireScheduled(bot, entry); + } catch (err) { + logError("bot", `Scheduled message failed (chat=${entry.chatId})`, err); + } + deleteScheduled(entry.id); + timers.delete(entry.id); + }, delayMs); + timers.set(entry.id, timer); +} + +/** + * Re-arm persisted scheduled sends after a restart. Called once when + * the action handler is created. Overdue entries fire immediately — + * late beats never — unless they are stale past MAX_OVERDUE_MS, which + * drops them (a day-late reminder is noise, not delivery). + */ +export function restoreScheduledMessages( + bot: Bot, + timers: Map>, +): void { + const entries = listScheduled("telegram"); + if (entries.length === 0) return; + let restored = 0; + let dropped = 0; + for (const entry of entries) { + if (Date.now() - entry.fireAt > MAX_OVERDUE_MS) { + deleteScheduled(entry.id); + dropped++; + continue; + } + armScheduled(bot, entry, timers); + restored++; + } + log( + "bot", + `Restored ${restored} scheduled message(s) from store${dropped > 0 ? `, dropped ${dropped} stale` : ""}`, + ); +} + export const messagingHandlers: TelegramActionHandlers = { send_message: async (body, chatId, { bot, gateway }) => { const text = String(body.text ?? ""); @@ -157,51 +242,54 @@ export const messagingHandlers: TelegramActionHandlers = { schedule_message: (body, chatId, { bot, scheduledMessages }) => { const text = String(body.text ?? ""); const replyTo = toPositiveId(body.reply_to_message_id); - const rows = body.rows as - | Array> - | undefined; + const rows = body.rows as ScheduledMessage["rows"]; const delaySec = Math.max( 1, - Math.min(3600, Number(body.delay_seconds ?? 60)), + Math.min(MAX_DELAY_SEC, Number(body.delay_seconds ?? 60)), ); - const scheduleId = `sched_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`; - const timer = setTimeout(async () => { - try { - if (rows) { - const keyboard = rows.map((row) => - row.map((btn) => - btn.url - ? { text: btn.text, url: btn.url } - : { - text: btn.text, - callback_data: btn.callback_data ?? btn.text, - }, - ), - ); - await bot.api.sendMessage(chatId, markdownToTelegramHtml(text), { - parse_mode: "HTML", - reply_markup: { inline_keyboard: keyboard }, - reply_parameters: replyTo ? { message_id: replyTo } : undefined, - }); - } else { - await sendText(bot, chatId, text, replyTo); - } - } catch (err) { - logError("bot", `Scheduled message failed (chat=${chatId})`, err); - } - scheduledMessages.delete(scheduleId); - }, delaySec * 1000); - scheduledMessages.set(scheduleId, timer); - return { ok: true, schedule_id: scheduleId, delay_seconds: delaySec }; + const entry: ScheduledMessage = { + id: `sched_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`, + frontend: "telegram", + chatId: String(chatId), + text, + fireAt: Date.now() + delaySec * 1000, + createdAt: Date.now(), + replyTo, + rows, + }; + // Persist before arming: if we crash between the two, the restore + // path delivers it; the reverse order could lose it forever. + saveScheduled(entry); + armScheduled(bot, entry, scheduledMessages); + return { ok: true, schedule_id: entry.id, delay_seconds: delaySec }; }, cancel_scheduled: (body, _chatId, { scheduledMessages }) => { - const timer = scheduledMessages.get(String(body.schedule_id ?? "")); + const id = String(body.schedule_id ?? ""); + const timer = scheduledMessages.get(id); if (timer) { clearTimeout(timer); - scheduledMessages.delete(String(body.schedule_id)); - return { ok: true, cancelled: true }; + scheduledMessages.delete(id); } - return { ok: false, error: "Schedule not found" }; + // The store is the source of truth — an entry can exist without a + // live timer (scheduled before a restart, not yet re-armed here). + const existed = deleteScheduled(id) || Boolean(timer); + return existed + ? { ok: true, cancelled: true } + : { ok: false, error: "Schedule not found" }; + }, + + list_scheduled: (body, chatId) => { + const pending = listScheduledForChat("telegram", String(chatId)).map( + (e) => ({ + schedule_id: e.id, + fires_in_seconds: Math.max( + 0, + Math.round((e.fireAt - Date.now()) / 1000), + ), + text: e.text.length > 200 ? `${e.text.slice(0, 200)}…` : e.text, + }), + ); + return { ok: true, scheduled: pending, count: pending.length }; }, }; diff --git a/src/storage/scheduled-store.ts b/src/storage/scheduled-store.ts new file mode 100644 index 000000000..d91756abd --- /dev/null +++ b/src/storage/scheduled-store.ts @@ -0,0 +1,84 @@ +/** + * Persistent scheduled-message store, backed by the `kv` table + * (single blob at "scheduled.messages" — the volume is a handful of + * pending sends, well inside kv's small-state doctrine). + * + * Why this exists: `send(..., delay_seconds=N)` used to live only in + * an in-process setTimeout. The model tells the user "reminder in 30 + * minutes", Talon restarts at minute 20, and the promise silently + * evaporates — the exact failure the "never promise without a + * mechanism" rule warns about, baked into the platform. Cron jobs + * survive restarts; scheduled sends now do too. + * + * The store is the source of truth; the per-frontend timer maps are + * just the armed alarms. Frontends call `restoreScheduled*` variants + * at action-handler creation: overdue entries fire immediately (late + * beats never — up to a staleness cap), future entries re-arm. + */ + +import { kvGet, kvSet } from "./kv.js"; + +const KEY = "scheduled.messages"; + +/** Skip (and drop) entries overdue by more than this — a week-late + * "dinner in an hour" reminder is noise, not delivery. */ +export const MAX_OVERDUE_MS = 24 * 60 * 60 * 1000; + +export type ScheduledButtonRow = Array<{ + text: string; + url?: string; + callback_data?: string; +}>; + +export type ScheduledMessage = { + id: string; + /** Which frontend owns delivery ("telegram" | "discord"). */ + frontend: string; + /** Chat / channel id, in the owning frontend's native shape. */ + chatId: string; + text: string; + /** Epoch ms when the message should be sent. */ + fireAt: number; + createdAt: number; + /** Telegram: numeric message id to reply to. */ + replyTo?: number; + /** Telegram: inline keyboard rows. */ + rows?: ScheduledButtonRow[]; +}; + +function readAll(): Record { + return kvGet>(KEY) ?? {}; +} + +export function saveScheduled(entry: ScheduledMessage): void { + const all = readAll(); + all[entry.id] = entry; + kvSet(KEY, all); +} + +export function deleteScheduled(id: string): boolean { + const all = readAll(); + if (!(id in all)) return false; + delete all[id]; + kvSet(KEY, all); + return true; +} + +export function getScheduled(id: string): ScheduledMessage | undefined { + return readAll()[id]; +} + +/** All pending entries for a frontend, soonest first. */ +export function listScheduled(frontend: string): ScheduledMessage[] { + return Object.values(readAll()) + .filter((e) => e.frontend === frontend) + .sort((a, b) => a.fireAt - b.fireAt); +} + +/** Pending entries for one chat of a frontend, soonest first. */ +export function listScheduledForChat( + frontend: string, + chatId: string, +): ScheduledMessage[] { + return listScheduled(frontend).filter((e) => e.chatId === chatId); +} diff --git a/src/util/config.ts b/src/util/config.ts index f2c6ff347..deb727495 100644 --- a/src/util/config.ts +++ b/src/util/config.ts @@ -587,14 +587,39 @@ export function rebuildSystemPrompt( config: TalonConfig, pluginAdditions: string[], ): void { + const promptParts = buildSystemPromptPartsFor( + config, + pluginAdditions, + primaryFrontend(config), + ); + config.systemPromptParts = promptParts; + config.systemPrompt = joinSystemPromptParts(promptParts); +} + +/** The first configured frontend — the default prompt flavour. */ +export function primaryFrontend(config: TalonConfig): string { const frontends = Array.isArray(config.frontend) ? config.frontend : [config.frontend]; - const promptParts = assembleSystemPrompt({ - frontend: frontends[0], + return frontends[0]; +} + +/** + * Build system-prompt parts for a specific frontend WITHOUT mutating + * the config. Multi-frontend deployments need per-chat prompt + * flavours (a native-app chat must get native.md guidance, not the + * telegram.md that `frontends[0]` happens to be) — the per-session + * snapshot layer (backend/shared/system-prompt.ts) calls this with the + * chat's owning frontend and freezes the result per session. + */ +export function buildSystemPromptPartsFor( + config: TalonConfig, + pluginAdditions: string[], + frontend: string, +): SystemPromptParts { + return assembleSystemPrompt({ + frontend, pluginPromptAdditions: pluginAdditions.length > 0 ? pluginAdditions : undefined, }); - config.systemPromptParts = promptParts; - config.systemPrompt = joinSystemPromptParts(promptParts); }