From b833dcbb0c631292798b3af19a71ea0470f36ddc Mon Sep 17 00:00:00 2001 From: Leonel Rivas Date: Tue, 6 Oct 2026 15:09:09 -0700 Subject: [PATCH] feat(telegram): add Telegram channel integration, bot responder and settings --- README.md | 1 + apps/server/src/auth/RpcAuthorization.ts | 1 + apps/server/src/mcp/McpHttpServer.ts | 5 + apps/server/src/mcp/toolkits/core.test.ts | 2 + .../mcp/toolkits/telegram/handlers.test.ts | 118 ++ .../src/mcp/toolkits/telegram/handlers.ts | 32 + .../server/src/mcp/toolkits/telegram/tools.ts | 42 + .../toolkits/worktree/registration.test.ts | 2 + .../src/observability/RpcInstrumentation.ts | 1 + apps/server/src/persistence/Errors.ts | 1 + apps/server/src/persistence/Migrations.ts | 2 + .../Migrations/055_OrchestrationV2.test.ts | 4 +- .../Migrations/060_TelegramMessages.ts | 21 + .../src/persistence/TelegramMessages.ts | 106 ++ .../reconcileV2PreviewMigration.test.ts | 2 + apps/server/src/server.ts | 12 + apps/server/src/serverSettings.test.ts | 68 ++ apps/server/src/serverSettings.ts | 71 +- apps/server/src/telegram/TelegramBotApi.ts | 214 ++++ .../src/telegram/TelegramChannel.test.ts | 1037 +++++++++++++++++ apps/server/src/telegram/TelegramChannel.ts | 688 +++++++++++ .../server/src/telegram/telegramRules.test.ts | 176 +++ apps/server/src/telegram/telegramRules.ts | 182 +++ apps/server/src/ws.ts | 6 + .../settings/BitbucketCredentialsSettings.tsx | 2 +- .../settings/IntegrationsSettings.tsx | 2 + .../settings/TelegramSettings.logic.test.ts | 133 +++ .../settings/TelegramSettings.logic.ts | 119 ++ .../components/settings/TelegramSettings.tsx | 257 ++++ .../src/components/settings/settingsSearch.ts | 7 + docs/README.md | 1 + docs/user/telegram.md | 36 + packages/client-runtime/src/state/server.ts | 4 + .../client-runtime/src/t3ToolSummary.test.ts | 18 + packages/client-runtime/src/t3ToolSummary.ts | 3 + .../src/work-log/presentation.ts | 1 + packages/contracts/src/index.ts | 1 + packages/contracts/src/rpc.ts | 9 + packages/contracts/src/settings.ts | 25 + packages/contracts/src/telegram.ts | 80 ++ packages/shared/src/t3McpToolPresentation.ts | 4 +- 41 files changed, 3488 insertions(+), 8 deletions(-) create mode 100644 apps/server/src/mcp/toolkits/telegram/handlers.test.ts create mode 100644 apps/server/src/mcp/toolkits/telegram/handlers.ts create mode 100644 apps/server/src/mcp/toolkits/telegram/tools.ts create mode 100644 apps/server/src/persistence/Migrations/060_TelegramMessages.ts create mode 100644 apps/server/src/persistence/TelegramMessages.ts create mode 100644 apps/server/src/telegram/TelegramBotApi.ts create mode 100644 apps/server/src/telegram/TelegramChannel.test.ts create mode 100644 apps/server/src/telegram/TelegramChannel.ts create mode 100644 apps/server/src/telegram/telegramRules.test.ts create mode 100644 apps/server/src/telegram/telegramRules.ts create mode 100644 apps/web/src/components/settings/TelegramSettings.logic.test.ts create mode 100644 apps/web/src/components/settings/TelegramSettings.logic.ts create mode 100644 apps/web/src/components/settings/TelegramSettings.tsx create mode 100644 docs/user/telegram.md create mode 100644 packages/contracts/src/telegram.ts diff --git a/README.md b/README.md index ad6ef84031403..e908ba2a611df 100644 --- a/README.md +++ b/README.md @@ -97,6 +97,7 @@ Full docs live in [docs/](./docs). There's no docs site yet. - [Connect Claude Code, Codex, ChatGPT and other agents over MCP](./docs/user/outside-agents.md) - [Keeping app and server in sync](./docs/user/updating.md) - [Source control integrations](./docs/user/source-control.md) +- [Telegram](./docs/user/telegram.md) - Multiple accounts: [Codex](./docs/user/providers-codex.md) · [Claude](./docs/user/providers-claude.md) - [Run T3 Code as a background service](./docs/user/background-service.md) diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index d434ca86a3a54..66e9bd63ee8b2 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -81,6 +81,7 @@ export const RPC_REQUIRED_SCOPES = { [WS_METHODS.serverRemoveKeybinding]: AuthSettingsWriteScope, [WS_METHODS.serverGetSettings]: AuthOrchestrationReadScope, [WS_METHODS.serverUpdateSettings]: AuthSettingsWriteScope, + [WS_METHODS.serverTestTelegram]: AuthSettingsWriteScope, [WS_METHODS.serverSearchAcpRegistry]: AuthOrchestrationReadScope, [WS_METHODS.serverPrepareAcpRegistryAgent]: AuthProvidersManageScope, [WS_METHODS.serverUninstallAcpRegistryManagedBinary]: AuthProvidersManageScope, diff --git a/apps/server/src/mcp/McpHttpServer.ts b/apps/server/src/mcp/McpHttpServer.ts index b63cd7b95d7ed..6cd79bc0335d8 100644 --- a/apps/server/src/mcp/McpHttpServer.ts +++ b/apps/server/src/mcp/McpHttpServer.ts @@ -49,6 +49,8 @@ import { WorktreeToolkit } from "./toolkits/worktree/tools.ts"; import * as WorktreeMcpService from "./WorktreeMcpService.ts"; import * as PullRequestsHandlers from "./toolkits/pullRequests/handlers.ts"; import { PullRequestsToolkit } from "./toolkits/pullRequests/tools.ts"; +import * as TelegramHandlers from "./toolkits/telegram/handlers.ts"; +import { TelegramToolkit } from "./toolkits/telegram/tools.ts"; import * as DeviceHandlers from "./toolkits/device/handlers.ts"; import { DeviceScreenshotTool, @@ -825,6 +827,8 @@ export const layerPullRequestsToolkit = toolkitRegistration( PullRequestsHandlers.layer, ); +const layerTelegramRegistration = toolkitRegistration(TelegramToolkit, TelegramHandlers.layer); + const layerDeviceStandardToolkitRegistration = toolkitRegistration( DeviceStandardToolkit, DeviceHandlers.layerStandard, @@ -857,6 +861,7 @@ export const layer = Layer.mergeAll( layerPreviewControlsRegistration, layerWorktreeToolkitRegistration, layerPullRequestsToolkit, + layerTelegramRegistration, layerDeviceToolkit, layerHtmlToolkit, ).pipe(Layer.provideMerge(layerMcpTransport)); diff --git a/apps/server/src/mcp/toolkits/core.test.ts b/apps/server/src/mcp/toolkits/core.test.ts index 634133b62e9e1..8c1e3c9833bab 100644 --- a/apps/server/src/mcp/toolkits/core.test.ts +++ b/apps/server/src/mcp/toolkits/core.test.ts @@ -58,6 +58,7 @@ const declaredFailure = (result: McpSchema.CallToolResult) => { }; import { PullRequestsToolkit } from "./pullRequests/tools.ts"; import { HtmlToolkit } from "./html/tools.ts"; +import { TelegramToolkit } from "./telegram/tools.ts"; import { resolveT3McpToolDefinition, resolveT3McpToolPresentation, @@ -81,6 +82,7 @@ it("publishes unique tool names with reference-free object-root inputs", () => { DeviceToolkit, PullRequestsToolkit, HtmlToolkit, + TelegramToolkit, ]) { for (const tool of Object.values(toolkit.tools)) { expect(names.has(tool.name)).toBe(false); diff --git a/apps/server/src/mcp/toolkits/telegram/handlers.test.ts b/apps/server/src/mcp/toolkits/telegram/handlers.test.ts new file mode 100644 index 0000000000000..7dfbf079785c2 --- /dev/null +++ b/apps/server/src/mcp/toolkits/telegram/handlers.test.ts @@ -0,0 +1,118 @@ +import { expect, it } from "@effect/vitest"; +import { + EnvironmentId, + ProviderInstanceId, + TelegramSendError, + ThreadId, + type OrchestrationV2ThreadShell, +} from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Stream from "effect/Stream"; + +import * as ThreadManagement from "../../../orchestration-v2/ThreadManagementService.ts"; +import * as TelegramChannel from "../../../telegram/TelegramChannel.ts"; +import * as McpInvocationContext from "../../McpInvocationContext.ts"; +import * as McpToolAccess from "../../McpToolAccess.ts"; +import * as TelegramHandlers from "./handlers.ts"; +import { TelegramToolkit } from "./tools.ts"; + +const callerThreadId = ThreadId.make("telegram-caller"); +const providerInstanceId = ProviderInstanceId.make("codex"); + +const callTelegramSend = ( + send: TelegramChannel.TelegramChannel["Service"]["send"], + parameters: { + readonly text?: string; + readonly path?: string; + readonly chatId?: number; + readonly chat?: string; + }, + runtimeMode: OrchestrationV2ThreadShell["runtimeMode"] = "full-access", +) => { + const layerDependencies = Layer.mergeAll( + Layer.succeed(McpInvocationContext.McpInvocationContext, { + environmentId: EnvironmentId.make("environment"), + requestNamespace: "session", + thread: { threadId: callerThreadId, providerSessionId: "session", providerInstanceId }, + client: undefined, + issuedAt: 0, + capabilities: new Set(["orchestration" as const]), + }), + Layer.mock(ThreadManagement.ThreadManagementService)({ + getThreadShell: () => + Effect.succeed({ + id: callerThreadId, + providerInstanceId, + runtimeMode, + interactionMode: "default", + activeRunId: "active-run", + archivedAt: null, + deletedAt: null, + } as OrchestrationV2ThreadShell), + }), + Layer.mock(TelegramChannel.TelegramChannel)({ send }), + ); + return TelegramToolkit.pipe( + Effect.provide( + McpToolAccess.HandlersLayer.layer(TelegramHandlers.layer).pipe( + Layer.provide(layerDependencies), + ), + ), + Effect.flatMap((toolkit) => toolkit.handle("telegram_send", parameters)), + Stream.unwrap, + Stream.runCollect, + Effect.map((results) => results.at(-1)?.result), + Effect.provide(layerDependencies), + ); +}; + +it.effect("sends on behalf of the calling thread, so the answer comes back to it", () => + Effect.gen(function* () { + const requests: Array = []; + const result = yield* callTelegramSend( + (input) => { + requests.push(input); + return Effect.succeed({ chatId: 42, messageId: 7, replyAvailable: true }); + }, + { text: "Build finished", chatId: 42 }, + ); + expect(result).toEqual({ chatId: 42, messageId: 7, replyAvailable: true }); + // A full-access thread may send a file from outside its workspace. + expect(requests).toEqual([ + { threadId: callerThreadId, text: "Build finished", chatId: 42, hostFiles: true }, + ]); + }), +); + +it.effect("does not let a thread without full access send files from outside its workspace", () => + Effect.gen(function* () { + const requests: Array = []; + yield* callTelegramSend( + (input) => { + requests.push(input); + return Effect.succeed({ chatId: 42, messageId: 7, replyAvailable: true }); + }, + { path: "/etc/hosts" }, + "approval-required", + ); + expect(requests).toEqual([{ threadId: callerThreadId, path: "/etc/hosts", hostFiles: false }]); + }), +); + +it.effect.each([ + { reason: "file-unreadable", code: "invalid_request" }, + { reason: "not-configured", code: "invalid_request" }, + { reason: "rejected", code: "invalid_request" }, + { reason: "unreachable", code: "orchestration_error" }, + { reason: "context-unavailable", code: "orchestration_error" }, + { reason: "reply-unavailable", code: "invalid_request" }, + { reason: "chat-not-found", code: "invalid_request" }, + { reason: "file-outside-workspace", code: "invalid_request" }, +] as const)("tells the agent why a $reason send failed", ({ reason, code }) => + Effect.gen(function* () { + const error = new TelegramSendError({ reason }); + const result = yield* callTelegramSend(() => Effect.fail(error), { text: "Build finished" }); + expect(result).toMatchObject({ _tag: "OrchestratorMcpFailure", code, message: error.message }); + }), +); diff --git a/apps/server/src/mcp/toolkits/telegram/handlers.ts b/apps/server/src/mcp/toolkits/telegram/handlers.ts new file mode 100644 index 0000000000000..c478afba54a14 --- /dev/null +++ b/apps/server/src/mcp/toolkits/telegram/handlers.ts @@ -0,0 +1,32 @@ +import { OrchestratorMcpFailure } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; + +import * as TelegramChannel from "../../../telegram/TelegramChannel.ts"; +import * as McpToolAccess from "../../McpToolAccess.ts"; +import { loadCaller } from "../../threadAccess.ts"; +import { TelegramToolkit } from "./tools.ts"; + +export const layer = McpToolAccess.toLayer(TelegramToolkit, { + telegram_send: McpToolAccess.actsAsCaller((input) => + Effect.gen(function* () { + const { caller, limits } = yield* loadCaller(); + const telegram = yield* TelegramChannel.TelegramChannel; + return yield* telegram + .send({ ...input, threadId: caller?.id, hostFiles: limits.runtimeMode === "full-access" }) + .pipe( + Effect.mapError( + (error) => + new OrchestratorMcpFailure({ + // Only a send Telegram never answered is worth another try; what it refused, + // and what the request itself got wrong, will fail the same way again. + code: + error.reason === "unreachable" || error.reason === "context-unavailable" + ? "orchestration_error" + : "invalid_request", + message: error.message, + }), + ), + ); + }), + ), +}); diff --git a/apps/server/src/mcp/toolkits/telegram/tools.ts b/apps/server/src/mcp/toolkits/telegram/tools.ts new file mode 100644 index 0000000000000..fb7025895e802 --- /dev/null +++ b/apps/server/src/mcp/toolkits/telegram/tools.ts @@ -0,0 +1,42 @@ +import { OrchestratorMcpFailure, TelegramId, TelegramSendResult } from "@t3tools/contracts"; +import * as Schema from "effect/Schema"; +import { Tool, Toolkit } from "effect/ai"; + +import * as ThreadManagementService from "../../../orchestration-v2/ThreadManagementService.ts"; +import * as TelegramChannel from "../../../telegram/TelegramChannel.ts"; +import * as McpInvocationContext from "../../McpInvocationContext.ts"; + +const TelegramSendTool = Tool.make("telegram_send", { + description: + "Send a message or a file to the user's Telegram chat, or to another chat or group listed in Settings. Use it when the user asks to be told or sent something on Telegram, and to answer a message that says it was sent from Telegram. Give text, path, or both. Text is shown as written, without Markdown, up to 4096 characters, or 1024 as the caption of a file. JPG, PNG, and WebP images and MP4 videos play in the chat; any other file, such as a PDF, arrives as a document. Files can be up to 50 MB. The user can answer the message in Telegram, and the answer arrives in this thread. A successful result means the message was sent: if replyAvailable is false, explain the warning but do not resend the file.", + parameters: Schema.Struct({ + text: Schema.optional(Schema.String).annotate({ + description: "The message, or the caption when a file is sent.", + }), + path: Schema.optional(Schema.String).annotate({ + description: + "A file to upload. A relative path is read from this thread's workspace; an absolute path is read in place.", + }), + chatId: Schema.optional(TelegramId).annotate({ + description: + "Use the chatId supplied by an incoming Telegram message when answering it. Otherwise use a configured destination. When omitted, replies go to the latest Telegram sender in the active turn; other sends use the first configured chat.", + }), + chat: Schema.optional(Schema.String).annotate({ + description: + "The name of a chat or group from Settings, when the user names one, such as the team group. Part of the name is enough when only one chat fits. Leave it out to answer a Telegram message or to use the default chat.", + }), + }), + success: TelegramSendResult, + failure: OrchestratorMcpFailure, + failureMode: "return", + dependencies: [ + McpInvocationContext.McpInvocationContext, + ThreadManagementService.ThreadManagementService, + TelegramChannel.TelegramChannel, + ], +}) + .annotate(Tool.Title, "Send to Telegram") + .annotate(Tool.Destructive, false) + .annotate(Tool.OpenWorld, true); + +export const TelegramToolkit = Toolkit.make(TelegramSendTool); diff --git a/apps/server/src/mcp/toolkits/worktree/registration.test.ts b/apps/server/src/mcp/toolkits/worktree/registration.test.ts index 588d11d9c7152..f5b630de6831b 100644 --- a/apps/server/src/mcp/toolkits/worktree/registration.test.ts +++ b/apps/server/src/mcp/toolkits/worktree/registration.test.ts @@ -21,6 +21,7 @@ import * as ProviderRegistry from "../../../provider/ProviderRegistry.ts"; import * as ScheduledTaskService from "../../../scheduledTasks/ScheduledTaskService.ts"; import * as SecretRequests from "../../../secrets/SecretRequests.ts"; import * as ServerSettings from "../../../serverSettings.ts"; +import * as TelegramChannel from "../../../telegram/TelegramChannel.ts"; import * as VcsStatusBroadcaster from "../../../vcs/VcsStatusBroadcaster.ts"; import * as McpHttpServer from "../../McpHttpServer.ts"; import * as McpSessionRegistry from "../../McpSessionRegistry.ts"; @@ -37,6 +38,7 @@ const layerStubServices = Layer.mergeAll( Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), Layer.mock(SecretRequests.SecretRequests)({}), Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(TelegramChannel.TelegramChannel)({}), ServerSettings.layerTest({}), Layer.mock(GitWorkflowService.GitWorkflowService)({}), Layer.mock(ProjectSetupScriptRunner.ProjectSetupScriptRunner)({}), diff --git a/apps/server/src/observability/RpcInstrumentation.ts b/apps/server/src/observability/RpcInstrumentation.ts index d01c7812bc1a1..e08e02308eb72 100644 --- a/apps/server/src/observability/RpcInstrumentation.ts +++ b/apps/server/src/observability/RpcInstrumentation.ts @@ -54,6 +54,7 @@ const RPC_AGGREGATES = { [WS_METHODS.serverRemoveKeybinding]: "server", [WS_METHODS.serverGetSettings]: "server", [WS_METHODS.serverUpdateSettings]: "server", + [WS_METHODS.serverTestTelegram]: "server", [WS_METHODS.serverSearchAcpRegistry]: "server", [WS_METHODS.serverPrepareAcpRegistryAgent]: "server", [WS_METHODS.serverUninstallAcpRegistryManagedBinary]: "server", diff --git a/apps/server/src/persistence/Errors.ts b/apps/server/src/persistence/Errors.ts index 67290369b0f42..73bab88543046 100644 --- a/apps/server/src/persistence/Errors.ts +++ b/apps/server/src/persistence/Errors.ts @@ -129,5 +129,6 @@ export type ProviderSessionRuntimeRepositoryError = PersistenceSqlError | Persis export type AuthPairingLinkRepositoryError = PersistenceSqlError | PersistenceDecodeError; export type AuthSessionRepositoryError = PersistenceSqlError | PersistenceDecodeError; export type PullRequestFilesViewedRepositoryError = PersistenceSqlError | PersistenceDecodeError; +export type TelegramMessageRepositoryError = PersistenceSqlError | PersistenceDecodeError; export type ProjectionRepositoryError = PersistenceSqlError | PersistenceDecodeError; diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 1e1f207a7dac8..c20128becebb1 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -73,6 +73,7 @@ import Migration0056 from "./Migrations/056_RemoveRedundantProjectionIndexes.ts" import Migration0057 from "./Migrations/057_ScheduledTaskWebhooks.ts"; import Migration0058 from "./Migrations/058_WebhookRelayDeliveries.ts"; import Migration0059 from "./Migrations/059_McpAppModelContext.ts"; +import Migration0060 from "./Migrations/060_TelegramMessages.ts"; /** * Migration loader with all migrations defined inline. @@ -146,6 +147,7 @@ export const migrationEntries = [ [57, "ScheduledTaskWebhooks", Migration0057], [58, "WebhookRelayDeliveries", Migration0058], [59, "McpAppModelContext", Migration0059], + [60, "TelegramMessages", Migration0060], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts b/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts index 5f2e05f2809f3..f287980ce3beb 100644 --- a/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts +++ b/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts @@ -13,7 +13,7 @@ layer("055_OrchestrationV2", (it) => { Effect.sync(() => { assert.deepStrictEqual( migrationEntries.map(([id]) => id), - Array.from({ length: 59 }, (_, index) => index + 1), + Array.from({ length: 60 }, (_, index) => index + 1), ); }), ); @@ -31,6 +31,7 @@ layer("055_OrchestrationV2", (it) => { [57, "ScheduledTaskWebhooks"], [58, "WebhookRelayDeliveries"], [59, "McpAppModelContext"], + [60, "TelegramMessages"], ]); assert.deepStrictEqual(yield* runMigrations(), []); @@ -56,6 +57,7 @@ layer("055_OrchestrationV2", (it) => { { migration_id: 57, name: "ScheduledTaskWebhooks" }, { migration_id: 58, name: "WebhookRelayDeliveries" }, { migration_id: 59, name: "McpAppModelContext" }, + { migration_id: 60, name: "TelegramMessages" }, ]); const tables = yield* sql<{ readonly name: string }>` diff --git a/apps/server/src/persistence/Migrations/060_TelegramMessages.ts b/apps/server/src/persistence/Migrations/060_TelegramMessages.ts new file mode 100644 index 0000000000000..0862009ad9182 --- /dev/null +++ b/apps/server/src/persistence/Migrations/060_TelegramMessages.ts @@ -0,0 +1,21 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // One row per message the bot sent on behalf of a thread. Telegram only tells a bot which + // message somebody answered, so this is what turns that answer into the thread it is for. + // A message id is only unique within one bot's view of one chat, so a different bot's + // messages in the same chat start over at the same numbers. + yield* sql` + CREATE TABLE IF NOT EXISTS telegram_messages ( + bot_id TEXT NOT NULL, + chat_id INTEGER NOT NULL, + message_id INTEGER NOT NULL, + thread_id TEXT NOT NULL, + sent_at TEXT NOT NULL, + PRIMARY KEY (bot_id, chat_id, message_id) + ) WITHOUT ROWID + `; +}); diff --git a/apps/server/src/persistence/TelegramMessages.ts b/apps/server/src/persistence/TelegramMessages.ts new file mode 100644 index 0000000000000..cf22d7d916d3d --- /dev/null +++ b/apps/server/src/persistence/TelegramMessages.ts @@ -0,0 +1,106 @@ +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; +import * as SqlClient from "effect/sql/SqlClient"; +import * as SqlSchema from "effect/sql/SqlSchema"; + +import { TelegramId, ThreadId } from "@t3tools/contracts"; + +import { + PersistenceDecodeError, + PersistenceSqlError, + type TelegramMessageRepositoryError, +} from "./Errors.ts"; + +/** A message as one bot sees it. Its id is only unique within that bot's view of that chat. */ +const TelegramMessageKey = Schema.Struct({ + botId: Schema.String, + chatId: TelegramId, + messageId: Schema.Int, +}); +type TelegramMessageKey = typeof TelegramMessageKey.Type; + +const TelegramMessageThread = Schema.Struct({ threadId: ThreadId }); + +interface RecordTelegramMessageInput extends TelegramMessageKey { + readonly threadId: ThreadId; + /** When the bot sent it, as an ISO instant. */ + readonly sentAt: string; + /** Rows sent before this ISO instant are dropped in the same write. */ + readonly dropBefore: string; +} + +/** + * Which thread each message the bot sent belongs to, so an answer to that message can be + * delivered to it. Rows expire: nobody answers a months-old message, and a row that outlives + * its thread is only ever a lookup that finds nothing to deliver to. + */ +export class TelegramMessageRepository extends Context.Service< + TelegramMessageRepository, + { + readonly record: ( + input: RecordTelegramMessageInput, + ) => Effect.Effect; + readonly threadFor: ( + input: TelegramMessageKey, + ) => Effect.Effect, TelegramMessageRepositoryError>; + } +>()("t3/persistence/TelegramMessages/TelegramMessageRepository") {} + +const make = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + const findThread = SqlSchema.findOneOption({ + Request: TelegramMessageKey, + Result: TelegramMessageThread, + execute: ({ botId, chatId, messageId }) => + sql` + SELECT thread_id AS "threadId" + FROM telegram_messages + WHERE bot_id = ${botId} + AND chat_id = ${chatId} + AND message_id = ${messageId} + `, + }); + + return TelegramMessageRepository.of({ + record: (input) => + sql + .withTransaction( + Effect.gen(function* () { + yield* sql`DELETE FROM telegram_messages WHERE sent_at < ${input.dropBefore}`; + yield* sql` + INSERT INTO telegram_messages (bot_id, chat_id, message_id, thread_id, sent_at) + VALUES ( + ${input.botId}, + ${input.chatId}, + ${input.messageId}, + ${input.threadId}, + ${input.sentAt} + ) + ON CONFLICT (bot_id, chat_id, message_id) + DO UPDATE SET thread_id = excluded.thread_id, sent_at = excluded.sent_at + `; + }), + ) + .pipe( + Effect.mapError( + (cause) => new PersistenceSqlError({ operation: "recordTelegramMessage", cause }), + ), + ), + + threadFor: (input) => + findThread(input).pipe( + Effect.map(Option.map((row) => row.threadId)), + Effect.mapError((cause) => + Schema.isSchemaError(cause) + ? PersistenceDecodeError.fromSchemaError("TelegramMessageThread", cause) + : new PersistenceSqlError({ operation: "findTelegramMessageThread", cause }), + ), + ), + }); +}); + +export const layer = Layer.effect(TelegramMessageRepository, make); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 826dc4831d8db..84a9c0a7426b3 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -40,6 +40,7 @@ describe("V2 preview upgrade", () => { [57, "ScheduledTaskWebhooks"], [58, "WebhookRelayDeliveries"], [59, "McpAppModelContext"], + [60, "TelegramMessages"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -122,6 +123,7 @@ describe("V2 preview upgrade", () => { [57, "ScheduledTaskWebhooks"], [58, "WebhookRelayDeliveries"], [59, "McpAppModelContext"], + [60, "TelegramMessages"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index ebbc73758a437..d0b3440035fba 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -37,6 +37,7 @@ import * as PullRequestProviderRegistry from "./pullRequest/PullRequestProviderR import * as PullRequestService from "./pullRequest/PullRequestService.ts"; import * as SqlitePersistence from "./persistence/Sqlite.ts"; import * as PullRequestFilesViewed from "./persistence/PullRequestFilesViewed.ts"; +import * as TelegramMessages from "./persistence/TelegramMessages.ts"; import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts"; import * as AnalyticsService from "./telemetry/AnalyticsService.ts"; import * as ProviderEventIngestor from "./orchestration-v2/ProviderEventIngestor.ts"; @@ -53,6 +54,7 @@ import * as BitbucketApi from "./sourceControl/BitbucketApi.ts"; import * as GitHubCli from "./sourceControl/GitHubCli.ts"; import * as GitLabCli from "./sourceControl/GitLabCli.ts"; import * as ForgejoCli from "./sourceControl/ForgejoCli.ts"; +import * as TelegramChannel from "./telegram/TelegramChannel.ts"; import * as TextGeneration from "./textGeneration/TextGeneration.ts"; import * as ProviderInstanceRegistryHydration from "./provider/ProviderInstanceRegistryHydration.ts"; import * as TerminalManager from "./terminal/Manager.ts"; @@ -324,6 +326,8 @@ const layerPullRequestService = PullRequestService.layer.pipe( Layer.provide(SourceControlRateLimit.layer), ); +const layerTelegramChannel = TelegramChannel.layer.pipe(Layer.provide(TelegramMessages.layer)); + const layerGitManager = GitManager.layer.pipe( // Per-project git settings resolve the acting thread's project. Layer.provide(Layer.merge(ProjectionStoreV2.layer, ProjectStore.layer)), @@ -552,6 +556,12 @@ const layerRuntimeCoreDependenciesBase = Layer.mergeAll( Layer.provide(layerPullRequestService), Layer.provide(ProjectionStoreV2.layer), ), + Layer.effectDiscard( + Effect.gen(function* () { + const channel = yield* TelegramChannel.TelegramChannel; + yield* channel.start(); + }), + ).pipe(Layer.provide(layerTelegramChannel)), // Subscribes to `account.rate-limits.updated` so usage bars track live // telemetry instead of waiting for the next status probe. ProviderUsageLimitsIngestion.layer, @@ -690,6 +700,8 @@ const layerMakeRoutes = Layer.mergeAll( Layer.provide(ServerBrowser.layer.pipe(Layer.provide(DesktopBrowserChannel.layer))), // Server browser tabs and HTML render previews install and run the same headless browser. Layer.provide(PreviewBrowser.layer), + // The listener, the Settings test, and the agents' tool all go through one bot. + Layer.provide(layerTelegramChannel), Layer.provide(PreviewAutomationBroker.layer), Layer.provide(ServerSelfUpdate.layer.pipe(Layer.provide(layerDesktopAppUpdate))), Layer.provide(layerCommandReadiness), diff --git a/apps/server/src/serverSettings.test.ts b/apps/server/src/serverSettings.test.ts index 051f053316136..a839c2d0626b2 100644 --- a/apps/server/src/serverSettings.test.ts +++ b/apps/server/src/serverSettings.test.ts @@ -1714,6 +1714,74 @@ it.layer(NodeServices.layer)("server settings", (it) => { }).pipe(Effect.provide(layerServerSettingsWithSecrets())), ); + it.effect( + "keeps the Telegram bot token in the secret store and tells clients only that one is set", + () => + Effect.gen(function* () { + const serverSettings = yield* ServerSettingsModule.ServerSettingsService; + const secrets = yield* ServerSecretStore.ServerSecretStore; + const serverConfig = yield* ServerConfig.ServerConfig; + const fileSystem = yield* FileSystem.FileSystem; + + const saved = yield* serverSettings.updateSettings({ + telegram: { botToken: "123:bot-token", chatIds: [42, -1001], ownerIds: [42] }, + }); + assert.deepEqual(saved.telegram, { + botToken: "123:bot-token", + chatIds: [42, -1001], + ownerIds: [42], + }); + + const raw = yield* fileSystem.readFileString(serverConfig.settingsPath); + assert.notInclude(raw, "123:bot-token"); + assert.include(raw, "-1001"); + + const forClient = ServerSettingsModule.redactServerSettingsForClient(saved).telegram; + assert.notInclude(forClient.botToken, "123:bot-token"); + assert.isAbove(forClient.botToken.length, 0); + assert.deepEqual(forClient.chatIds, [42, -1001]); + + // A client echoing the redacted token back, or omitting it, keeps the saved one. + yield* serverSettings.updateSettings({ telegram: forClient }); + yield* serverSettings.updateSettings({ telegram: { ownerIds: [] } }); + assert.deepEqual((yield* serverSettings.getSettings).telegram, { + botToken: "123:bot-token", + chatIds: [42, -1001], + ownerIds: [], + }); + + const cleared = yield* serverSettings.updateSettings({ telegram: { botToken: "" } }); + assert.equal(cleared.telegram.botToken, ""); + assert.isTrue(Option.isNone(yield* secrets.get("telegram-bot-token"))); + }).pipe(Effect.provide(layerServerSettingsWithSecrets())), + ); + + it.effect("moves a hand-edited Telegram bot token into the secret store when settings load", () => + Effect.gen(function* () { + const serverConfig = yield* ServerConfig.ServerConfig; + const fileSystem = yield* FileSystem.FileSystem; + const secrets = yield* ServerSecretStore.ServerSecretStore; + const serverSettings = yield* ServerSettingsModule.ServerSettingsService; + yield* fileSystem.writeFileString( + serverConfig.settingsPath, + '{"telegram":{"botToken":"hand-edited-token","chatIds":[42]}}', + ); + + const loaded = yield* serverSettings.getSettings; + + assert.equal(loaded.telegram.botToken, "hand-edited-token"); + assert.notInclude( + yield* fileSystem.readFileString(serverConfig.settingsPath), + "hand-edited-token", + ); + const stored = yield* secrets.get("telegram-bot-token"); + assert.equal( + Option.isSome(stored) ? new TextDecoder().decode(stored.value) : null, + "hand-edited-token", + ); + }).pipe(Effect.provide(layerServerSettingsWithSecrets())), + ); + it.effect("removes a Bitbucket secret once its token is cleared by hand in settings.json", () => Effect.gen(function* () { const serverConfig = yield* ServerConfig.ServerConfig; diff --git a/apps/server/src/serverSettings.ts b/apps/server/src/serverSettings.ts index 7cbcd895ac140..1ce720d5bead0 100644 --- a/apps/server/src/serverSettings.ts +++ b/apps/server/src/serverSettings.ts @@ -160,6 +160,7 @@ const BITBUCKET_SECRET_NAMES = { apiToken: "bitbucket-api-token", } as const; const BITBUCKET_SECRET_FIELDS = ["accessToken", "apiToken"] as const; +const TELEGRAM_SECRET_NAME = "telegram-bot-token"; /** Hosts are case-insensitive; a patch can arrive before decoding lowercased its keys. */ function gitHubTokenSecretName(host: string): string { @@ -215,7 +216,8 @@ export function redactServerSettingsForClient(settings: ServerSettings): ServerS Object.entries(settings.github.tokens).map(([host, token]) => [host, redactSecret(token)]), ), }; - return { ...settings, providerInstances, usageLimitSources, bitbucket, github }; + const telegram = { ...settings.telegram, botToken: redactSecret(settings.telegram.botToken) }; + return { ...settings, providerInstances, usageLimitSources, bitbucket, github, telegram }; } export function applyProviderInstanceMutation( @@ -693,11 +695,11 @@ const make = Effect.gen(function* () { ); /** - * Moves Bitbucket tokens hand-edited into settings.json into the secret store as they load, + * Moves Bitbucket and Telegram tokens hand-edited into settings.json into the secret store as they load, * so plaintext does not stay on disk. If the store is unavailable, the token keeps working * from the file and the move is retried on the next load. */ - const moveInlineBitbucketTokens = (settings: ServerSettings) => + const moveInlineTokens = (settings: ServerSettings) => Effect.gen(function* () { const bitbucket = { ...settings.bitbucket }; let moved = false; @@ -735,7 +737,26 @@ const make = Effect.gen(function* () { tokens[host] = SECRET_REDACTED; moved = true; } - return moved ? { ...settings, bitbucket, github: { ...settings.github, tokens } } : settings; + let telegram = settings.telegram; + if (telegram.botToken.length > 0 && telegram.botToken !== SECRET_REDACTED) { + const stored = yield* secretStore + .set(TELEGRAM_SECRET_NAME, textEncoder.encode(telegram.botToken)) + .pipe( + Effect.as(true), + Effect.catch(() => + Effect.logWarning("failed to move the Telegram bot token into the secret store").pipe( + Effect.as(false), + ), + ), + ); + if (stored) { + telegram = { ...telegram, botToken: SECRET_REDACTED }; + moved = true; + } + } + return moved + ? { ...settings, bitbucket, github: { ...settings.github, tokens }, telegram } + : settings; }); const loadSettingsFromDisk = Effect.gen(function* () { @@ -823,7 +844,7 @@ const make = Effect.gen(function* () { ? foldLegacyProjectSettings(loaded, legacyProjectRows) : loaded; // Only rewrite a file that decoded cleanly; an untrusted one stays for the user to repair. - const migrated = settingsFileTrusted ? yield* moveInlineBitbucketTokens(folded) : folded; + const migrated = settingsFileTrusted ? yield* moveInlineTokens(folded) : folded; if (migrated !== loaded) { yield* writeSettingsAtomically(migrated); } @@ -925,12 +946,27 @@ const make = Effect.gen(function* () { ); tokens[host] = Option.isSome(secret) ? textDecoder.decode(secret.value) : ""; } + let telegram = settings.telegram; + if (telegram.botToken === SECRET_REDACTED) { + const secret = yield* secretStore + .get(TELEGRAM_SECRET_NAME) + .pipe( + Effect.mapError( + (cause) => new ServerSettingsError({ settingsPath, operation: "read-secret", cause }), + ), + ); + telegram = { + ...telegram, + botToken: Option.isSome(secret) ? textDecoder.decode(secret.value) : "", + }; + } return { ...settings, providerInstances: providerInstances as ServerSettings["providerInstances"], usageLimitSources: usageLimitSources as ServerSettings["usageLimitSources"], bitbucket, github: { ...settings.github, tokens }, + telegram, }; }); @@ -1124,6 +1160,30 @@ const make = Effect.gen(function* () { }); } + let telegram = next.telegram; + const inlineBotToken = current.telegram.botToken; + // Same rule as Bitbucket: the marker keeps what is saved, or moves a hand-edited value. + const botToken = + telegram.botToken !== SECRET_REDACTED + ? telegram.botToken + : inlineBotToken === SECRET_REDACTED || inlineBotToken.length === 0 + ? null + : inlineBotToken; + if (botToken !== null && botToken.length === 0) { + changes.push({ + kind: "remove", + secretName: TELEGRAM_SECRET_NAME, + operation: "remove-secret", + }); + } else if (botToken !== null) { + changes.push({ + kind: "write", + secretName: TELEGRAM_SECRET_NAME, + value: textEncoder.encode(botToken), + }); + telegram = { ...telegram, botToken: SECRET_REDACTED }; + } + return { settings: { ...next, @@ -1131,6 +1191,7 @@ const make = Effect.gen(function* () { usageLimitSources: usageLimitSources as ServerSettings["usageLimitSources"], bitbucket, github: { ...next.github, tokens }, + telegram, }, changes, }; diff --git a/apps/server/src/telegram/TelegramBotApi.ts b/apps/server/src/telegram/TelegramBotApi.ts new file mode 100644 index 0000000000000..8421593bb1948 --- /dev/null +++ b/apps/server/src/telegram/TelegramBotApi.ts @@ -0,0 +1,214 @@ +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Schema from "effect/Schema"; +import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/http"; + +const TELEGRAM_API_ORIGIN = "https://api.telegram.org"; +const REQUEST_TIMEOUT = Duration.seconds(20); +/** An upload of the largest file Telegram takes, on a slow uplink. */ +const UPLOAD_TIMEOUT = Duration.minutes(5); +/** How long Telegram holds a `getUpdates` request open when nothing is waiting. */ +const LONG_POLL_SECONDS = 25; +const LONG_POLL_TIMEOUT = Duration.seconds(LONG_POLL_SECONDS + 15); + +export type TelegramFileKind = "photo" | "video" | "document"; + +const FILE_METHOD = { + photo: "sendPhoto", + video: "sendVideo", + document: "sendDocument", +} as const satisfies Record; + +const reply = (result: Result) => + Schema.Struct({ + ok: Schema.Boolean, + result: Schema.optional(result), + error_code: Schema.optional(Schema.Int), + description: Schema.optional(Schema.String), + parameters: Schema.optional(Schema.Struct({ retry_after: Schema.optional(Schema.Int) })), + }); + +/** A chat as Telegram describes it. A group has a title; a person has names. */ +const TelegramChat = Schema.Struct({ + id: Schema.Int, + type: Schema.String, + title: Schema.optional(Schema.String), + first_name: Schema.optional(Schema.String), + last_name: Schema.optional(Schema.String), + username: Schema.optional(Schema.String), +}); +export type TelegramChat = typeof TelegramChat.Type; + +const BotReply = reply(Schema.Struct({ username: Schema.String })); +const ChatReply = reply(TelegramChat); +const SentMessageReply = reply( + Schema.Struct({ message_id: Schema.Int, chat: Schema.optional(TelegramChat) }), +); +// The payloads stay undecoded here: one update this build cannot read must not stop the +// ones after it from being acknowledged. +const UpdatesReply = reply( + Schema.Array( + Schema.Struct({ + update_id: Schema.Int, + message: Schema.optional(Schema.Unknown), + my_chat_member: Schema.optional(Schema.Unknown), + }), + ), +); + +/** + * A Bot API call that did not return a result. `rejected` carries Telegram's own answer; + * `unreachable` never got one. There is no `cause`: the request URL holds the bot token, and + * a transport error quotes that URL. + */ +export class TelegramApiError extends Schema.TaggedError()("TelegramApiError", { + method: Schema.String, + reason: Schema.Literals(["rejected", "unreachable"]), + errorCode: Schema.optionalKey(Schema.Int), + description: Schema.optionalKey(Schema.String), + retryAfterSeconds: Schema.optionalKey(Schema.Int), +}) { + override get message(): string { + return this.reason === "unreachable" + ? `Telegram could not be reached for ${this.method}.` + : `Telegram rejected ${this.method}: ${this.description ?? "no reason given"}.`; + } +} + +export interface TelegramTextMessage { + readonly chatId: number; + readonly text: string; + readonly replyToMessageId?: number; +} + +export interface TelegramFileMessage { + readonly chatId: number; + readonly kind: TelegramFileKind; + readonly file: File; + readonly caption: string; +} + +/** The Bot API calls the Telegram channel makes. */ +export const make = Effect.gen(function* () { + const httpClient = yield* HttpClient.HttpClient; + const call = (input: { + readonly token: string; + readonly method: string; + readonly schema: Schema.Struct<{ + readonly ok: typeof Schema.Boolean; + readonly result: Schema.optional; + readonly error_code: Schema.optional; + readonly description: Schema.optional; + readonly parameters: ReturnType>["fields"]["parameters"]; + }>; + readonly body: ( + request: HttpClientRequest.HttpClientRequest, + ) => HttpClientRequest.HttpClientRequest; + readonly timeout: Duration.Duration; + }) => + httpClient + .execute( + input.body( + HttpClientRequest.post(`${TELEGRAM_API_ORIGIN}/bot${input.token}/${input.method}`), + ), + ) + .pipe( + // Telegram answers a refused call with a non-2xx status and the same JSON envelope. + Effect.flatMap(HttpClientResponse.schemaBodyJson(input.schema)), + Effect.timeout(input.timeout), + Effect.mapError( + () => new TelegramApiError({ method: input.method, reason: "unreachable" }), + ), + Effect.flatMap((answer) => + answer.ok && answer.result !== undefined + ? Effect.succeed(answer.result) + : Effect.fail( + new TelegramApiError({ + method: input.method, + reason: "rejected", + ...(answer.error_code === undefined ? {} : { errorCode: answer.error_code }), + ...(answer.description === undefined ? {} : { description: answer.description }), + ...(answer.parameters?.retry_after === undefined + ? {} + : { retryAfterSeconds: answer.parameters.retry_after }), + }), + ), + ), + // The client's request span records the full URL, which holds the bot token. + Effect.provideService(HttpClient.TracerDisabledWhen, () => true), + ); + + return { + /** The bot a token belongs to. The cheapest call that proves the token works. */ + getMe: (token: string) => + call({ + token, + method: "getMe", + schema: BotReply, + body: (request) => request, + timeout: REQUEST_TIMEOUT, + }), + + /** A chat the bot can see, which is how a configured id gets its name. */ + getChat: (token: string, chatId: number) => + call({ + token, + method: "getChat", + schema: ChatReply, + body: HttpClientRequest.bodyJsonUnsafe({ chat_id: chatId }), + timeout: REQUEST_TIMEOUT, + }), + + sendMessage: (token: string, message: TelegramTextMessage) => + call({ + token, + method: "sendMessage", + schema: SentMessageReply, + body: HttpClientRequest.bodyJsonUnsafe({ + chat_id: message.chatId, + text: message.text, + ...(message.replyToMessageId === undefined + ? {} + : { + // Still sent when the message it answers was deleted in the meantime. + reply_parameters: { + message_id: message.replyToMessageId, + allow_sending_without_reply: true, + }, + }), + }), + timeout: REQUEST_TIMEOUT, + }), + + sendFile: (token: string, message: TelegramFileMessage) => + call({ + token, + method: FILE_METHOD[message.kind], + schema: SentMessageReply, + body: HttpClientRequest.bodyFormDataRecord({ + chat_id: message.chatId, + caption: message.caption.length > 0 ? message.caption : undefined, + [message.kind]: message.file, + }), + timeout: UPLOAD_TIMEOUT, + }), + + /** + * Waits for messages sent to the bot. Passing the id after the last update seen is what + * tells Telegram the earlier ones were handled; without it they are delivered again. + */ + getUpdates: (token: string, offset: number | undefined) => + call({ + token, + method: "getUpdates", + schema: UpdatesReply, + body: HttpClientRequest.bodyJsonUnsafe({ + timeout: LONG_POLL_SECONDS, + // `my_chat_member` is how the bot learns it was added to a group. + allowed_updates: ["message", "my_chat_member"], + ...(offset === undefined ? {} : { offset }), + }), + timeout: LONG_POLL_TIMEOUT, + }), + }; +}); diff --git a/apps/server/src/telegram/TelegramChannel.test.ts b/apps/server/src/telegram/TelegramChannel.test.ts new file mode 100644 index 0000000000000..b7e3451b1c984 --- /dev/null +++ b/apps/server/src/telegram/TelegramChannel.test.ts @@ -0,0 +1,1037 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, describe, it } from "@effect/vitest"; +import { + ProjectId, + RunId, + ThreadId, + type OrchestrationV2ConversationMessage, + type OrchestrationV2ThreadShell, + type TelegramSettings, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; +import * as TestClock from "effect/testing/TestClock"; +import * as HttpClient from "effect/http/HttpClient"; +import * as HttpClientResponse from "effect/http/HttpClientResponse"; + +import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts"; +import type { ProjectionRecords } from "../orchestration-v2/ProjectionStore.ts"; +import { PersistenceSqlError } from "../persistence/Errors.ts"; +import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import * as TelegramMessages from "../persistence/TelegramMessages.ts"; +import * as Project from "../project/ProjectService.ts"; +import * as ServerSettings from "../serverSettings.ts"; +import * as TelegramChannel from "./TelegramChannel.ts"; + +const BOT_TOKEN = "123:secret"; +const OWNER = 42; +const STRANGER = 77; +const GROUP = -1001; +/** How the stub's Telegram describes the chats the tests use. */ +const CHATS: Record> = { + [OWNER]: { id: OWNER, type: "private", first_name: "Olivia", last_name: "Owner" }, + [GROUP]: { id: GROUP, type: "supergroup", title: "Team" }, + [-1002]: { id: -1002, type: "supergroup", title: "Team leads" }, +}; +const PROJECT_ID = ProjectId.make("telegram-project"); + +const JsonRecord = Schema.fromJsonString(Schema.Record(Schema.String, Schema.Unknown)); +const decodeJsonRecord = Schema.decodeUnknownSync(JsonRecord); +const encodeJson = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown)); + +interface BotCall { + /** The Bot API method, such as `sendMessage`. */ + readonly method: string; + /** The JSON body, or the text fields of a multipart upload. */ + readonly body: Record; + readonly file?: File; + /** The id the stub gave the message it sent, if it sent one. */ + readonly messageId?: number; +} + +interface BotReply { + readonly status: number; + readonly body: unknown; +} + +/** + * Stands in for api.telegram.org. Sent messages get increasing ids, `getUpdates` hands out the + * queued batches in order, and a scripted reply for a method is used once, before the default. + */ +const makeTestBot = (blockEmptyPolls = false) => { + const calls: Array = []; + const updateBatches: Array> = []; + const scripted = new Map(); + let nextMessageId = 1000; + + const answer = ( + method: string, + body: Record, + ): { readonly reply: BotReply; readonly messageId?: number } => { + const chat = CHATS[Number(body.chat_id)]; + const reply = scripted.get(method); + if (reply !== undefined) { + scripted.delete(method); + return { reply }; + } + if (method === "getMe") { + return { reply: { status: 200, body: { ok: true, result: { username: "t3_test_bot" } } } }; + } + if (method === "getUpdates") { + return { reply: { status: 200, body: { ok: true, result: updateBatches.shift() ?? [] } } }; + } + if (method === "getChat") { + return chat === undefined + ? { + reply: { + status: 400, + body: { ok: false, error_code: 400, description: "Bad Request: chat not found" }, + }, + } + : { reply: { status: 200, body: { ok: true, result: chat } } }; + } + const messageId = nextMessageId++; + return { + reply: { + status: 200, + body: { + ok: true, + result: { message_id: messageId, ...(chat === undefined ? {} : { chat }) }, + }, + }, + messageId, + }; + }; + + const layer = Layer.succeed( + HttpClient.HttpClient, + HttpClient.make((request) => + Effect.suspend(() => { + const method = request.url.slice(request.url.lastIndexOf("/") + 1); + if ( + method === "getUpdates" && + blockEmptyPolls && + updateBatches.length === 0 && + !scripted.has(method) + ) + return Effect.never; + const body: Record = {}; + let file: File | undefined; + if (request.body._tag === "Uint8Array") { + Object.assign(body, decodeJsonRecord(new TextDecoder().decode(request.body.body))); + } else if (request.body._tag === "FormData") { + for (const [key, value] of request.body.formData.entries()) { + if (typeof value === "string") body[key] = value; + else file = value; + } + } + const { reply, messageId } = answer(method, body); + calls.push({ + method, + body, + ...(file === undefined ? {} : { file }), + ...(messageId === undefined ? {} : { messageId }), + }); + return Effect.succeed( + HttpClientResponse.fromWeb( + request, + new Response(encodeJson(reply.body), { + status: reply.status, + headers: { "content-type": "application/json" }, + }), + ), + ); + }), + ), + ); + + return { + calls, + layer, + queueUpdates: (batch: ReadonlyArray) => { + updateBatches.push(batch); + }, + script: (method: string, reply: BotReply) => { + scripted.set(method, reply); + }, + /** Everything the bot sent, leaving out its reads of the update queue. */ + sent: () => calls.filter((call) => call.method !== "getUpdates"), + }; +}; + +const thread = (input: { + readonly id: string; + readonly title: string; + readonly updatedAtMillis?: number; + readonly worktreePath?: string; + readonly archived?: boolean; + readonly deleted?: boolean; + readonly subagent?: boolean; +}) => + ({ + id: ThreadId.make(input.id), + projectId: PROJECT_ID, + title: input.title, + worktreePath: input.worktreePath ?? null, + activeRunId: null, + archivedAt: input.archived === true ? DateTime.makeUnsafe(1) : null, + deletedAt: input.deleted === true ? DateTime.makeUnsafe(1) : null, + updatedAt: DateTime.makeUnsafe(input.updatedAtMillis ?? 1), + lineage: { + parentThreadId: input.subagent === true ? ThreadId.make("parent") : null, + relationshipToParent: input.subagent === true ? "subagent" : null, + rootThreadId: ThreadId.make(input.id), + }, + }) as unknown as OrchestrationV2ThreadShell; + +/** An update carrying a text message from `from`, in their private chat unless `chatId` says. */ +const textUpdate = (input: { + readonly updateId: number; + readonly messageId: number; + readonly from: number; + readonly chatId?: number; + readonly text: string; + readonly replyTo?: number; +}) => { + const chatId = input.chatId ?? input.from; + return { + update_id: input.updateId, + message: { + message_id: input.messageId, + from: { id: input.from, is_bot: false, first_name: "Someone" }, + chat: { id: chatId, type: chatId < 0 ? "supergroup" : "private" }, + date: 1, + text: input.text, + ...(input.replyTo === undefined ? {} : { reply_to_message: { message_id: input.replyTo } }), + }, + }; +}; + +const makeHarness = (input: { + readonly telegram?: Partial; + readonly threads?: ReadonlyArray; + readonly failDeliveries?: number; + /** The thread refuses every message, which no retry can change. */ + readonly rejectDeliveries?: boolean; + readonly failRecording?: boolean; + readonly blockEmptyPolls?: boolean; + readonly onDelivered?: Effect.Effect; + readonly onDeliveryFailed?: Effect.Effect; +}) => { + const bot = makeTestBot(input.blockEmptyPolls); + const threads = [...(input.threads ?? [])]; + const delivered: Array = []; + const stored: Array = []; + let failuresLeft = input.failDeliveries ?? 0; + const repository = input.failRecording + ? Layer.mock(TelegramMessages.TelegramMessageRepository)({ + record: () => + Effect.fail( + new PersistenceSqlError({ + operation: "recordTelegramMessage", + cause: new Error("disk unavailable"), + }), + ), + }) + : TelegramMessages.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)); + const layer = TelegramChannel.layer.pipe( + Layer.provide(repository), + Layer.provideMerge( + ServerSettings.layerTest({ + telegram: { botToken: BOT_TOKEN, chatIds: [OWNER], ownerIds: [OWNER], ...input.telegram }, + }), + ), + Layer.provide( + Layer.mock(ThreadManagement.ThreadManagementService)({ + getThreadShell: (threadId) => + Effect.succeed(threads.find((candidate) => candidate.id === threadId) ?? null), + getShellSnapshot: () => + Effect.succeed({ schemaVersion: 1, snapshotSequence: 0, threads, archivedThreads: [] }), + getThreadRecords: (threadId, fields, filter) => + Effect.succeed({ + thread: threads.find((candidate) => candidate.id === threadId), + messages: stored.filter( + (message) => + message.threadId === threadId && + (filter?.messageIds === undefined || filter.messageIds.includes(message.id)) && + (filter?.messageRunIds === undefined || + (message.runId !== null && filter.messageRunIds.includes(message.runId))) && + (filter?.messageRoles === undefined || filter.messageRoles.includes(message.role)), + ), + } as unknown as ProjectionRecords<(typeof fields)[number]>), + sendToThread: (send) => + Effect.suspend( + (): ReturnType => { + if (input.rejectDeliveries === true) { + return Effect.fail( + new ThreadManagement.ThreadManagementNoSteerableRunError({ + threadId: send.threadId, + mode: "steer", + }), + ); + } + if (failuresLeft-- > 0) + return (input.onDeliveryFailed ?? Effect.void).pipe( + Effect.andThen( + Effect.fail( + new ThreadManagement.ThreadManagementProjectionLoadError({ + projectId: send.projectId, + threadId: send.threadId, + cause: new Error("temporary database failure"), + }), + ), + ), + ); + delivered.push(send); + const index = threads.findIndex((candidate) => candidate.id === send.threadId); + const runId = threads[index]?.activeRunId ?? RunId.make(`run:${send.threadId}`); + threads[index] = { ...threads[index]!, activeRunId: runId }; + stored.push({ + ...send, + id: send.messageId, + runId, + nodeId: null, + role: "user", + streaming: false, + createdAt: DateTime.makeUnsafe(delivered.length), + updatedAt: DateTime.makeUnsafe(delivered.length), + }); + return (input.onDelivered ?? Effect.void).pipe( + Effect.as({} as ThreadManagement.ThreadManagementSendResult), + ); + }, + ), + }), + ), + Layer.provide(Layer.mock(Project.ProjectService)({})), + Layer.provide(bot.layer), + Layer.provide(NodeServices.layer), + ); + return { bot, delivered, threads, layer }; +}; + +const FIX_LOGIN = thread({ id: "thread-fix-login", title: "Fix login" }); + +describe("TelegramChannel", () => { + it.effect("returns each owner's replies privately even with a group as the default", () => { + const secondOwner = 88; + const { bot, delivered, layer } = makeHarness({ + telegram: { chatIds: [GROUP], ownerIds: [OWNER, secondOwner] }, + threads: [FIX_LOGIN], + }); + return Effect.gen(function* () { + const channel = yield* TelegramChannel.TelegramChannel; + for (const [index, owner] of [OWNER, secondOwner].entries()) { + bot.queueUpdates([ + textUpdate({ + updateId: index * 2 + 1, + messageId: index * 2 + 1, + from: owner, + text: "/threads", + }), + ]); + yield* channel.receive; + const listed = bot.sent().at(-1)!; + bot.queueUpdates([ + textUpdate({ + updateId: index * 2 + 2, + messageId: index * 2 + 2, + from: owner, + text: "Explain the result", + replyTo: listed.messageId!, + }), + ]); + yield* channel.receive; + const sent = yield* channel.send({ threadId: FIX_LOGIN.id, text: "The result" }); + assert.equal(sent.chatId, owner); + assert.isTrue(sent.replyAvailable); + assert.include(delivered.at(-1)!.text, `chatId ${owner}`); + } + const explicit = yield* channel.send({ + threadId: FIX_LOGIN.id, + text: "Requested group update", + chatId: GROUP, + }); + assert.equal(explicit.chatId, GROUP); + assert.isFalse(explicit.replyAvailable); + }).pipe(Effect.provide(layer)); + }); + + it.effect("does not carry a Telegram destination into a later desktop run", () => { + const { bot, threads, layer } = makeHarness({ + telegram: { chatIds: [GROUP, OWNER] }, + threads: [FIX_LOGIN], + }); + return Effect.gen(function* () { + const channel = yield* TelegramChannel.TelegramChannel; + const sent = yield* channel.send({ threadId: FIX_LOGIN.id, chatId: OWNER, text: "Ready" }); + bot.queueUpdates([ + textUpdate({ + updateId: 1, + messageId: 2, + from: OWNER, + text: "Continue", + replyTo: sent.messageId, + }), + ]); + yield* channel.receive; + threads[0] = { ...threads[0]!, activeRunId: RunId.make("later-desktop-run") }; + assert.equal( + (yield* channel.send({ threadId: FIX_LOGIN.id, text: "New desktop request" })).chatId, + GROUP, + ); + }).pipe(Effect.provide(layer)); + }); + + it.effect( + "refuses a removed owner's reply destination instead of falling back to a group", + () => { + const { bot, layer } = makeHarness({ + telegram: { chatIds: [GROUP, OWNER] }, + threads: [FIX_LOGIN], + }); + return Effect.gen(function* () { + const channel = yield* TelegramChannel.TelegramChannel; + const settings = yield* ServerSettings.ServerSettingsService; + const sent = yield* channel.send({ threadId: FIX_LOGIN.id, chatId: OWNER, text: "Ready" }); + bot.queueUpdates([ + textUpdate({ + updateId: 1, + messageId: 2, + from: OWNER, + text: "Continue", + replyTo: sent.messageId, + }), + ]); + yield* channel.receive; + yield* settings.updateSettings({ telegram: { ownerIds: [] } }); + const before = bot.sent().length; + const error = yield* channel + .send({ threadId: FIX_LOGIN.id, text: "Private answer" }) + .pipe(Effect.flip); + assert.equal(error.reason, "reply-unavailable"); + assert.equal(bot.sent().length, before); + }).pipe(Effect.provide(layer)); + }, + ); + + it.effect( + "reports delivery without a reply link when recording fails, without uploading twice", + () => { + const { bot, layer } = makeHarness({ threads: [FIX_LOGIN], failRecording: true }); + return Effect.gen(function* () { + const channel = yield* TelegramChannel.TelegramChannel; + const sent = yield* channel.send({ threadId: FIX_LOGIN.id, text: "Report ready" }); + assert.equal(sent.messageId, 1000); + assert.isFalse(sent.replyAvailable); + assert.include(sent.warning!, "Do not resend"); + assert.equal(bot.sent().length, 1); + }).pipe(Effect.provide(layer)); + }, + ); + + it.effect("keeps a failed delivery unacknowledged and accepts it on the next read", () => { + const { bot, delivered, layer } = makeHarness({ threads: [FIX_LOGIN], failDeliveries: 1 }); + return Effect.gen(function* () { + const channel = yield* TelegramChannel.TelegramChannel; + yield* channel.send({ threadId: FIX_LOGIN.id, text: "Ready" }); + const update = textUpdate({ + updateId: 7, + messageId: 5, + from: OWNER, + text: "Continue", + replyTo: 1000, + }); + bot.queueUpdates([update]); + bot.queueUpdates([update]); + yield* channel.receive.pipe(Effect.flip); + assert.equal(delivered.length, 0); + yield* channel.receive; + yield* channel.receive; + assert.equal(delivered.length, 1); + assert.deepEqual( + bot.calls.filter((call) => call.method === "getUpdates").map((call) => call.body.offset), + [undefined, undefined, 8], + ); + }).pipe(Effect.provide(layer)); + }); + + it.effect("keeps listening after a delivery fails, and delivers it on the next read", () => + Effect.gen(function* () { + const failed = yield* Deferred.make(); + const accepted = yield* Deferred.make(); + const { bot, delivered, layer } = makeHarness({ + threads: [FIX_LOGIN], + failDeliveries: 1, + blockEmptyPolls: true, + onDeliveryFailed: Deferred.succeed(failed, undefined), + onDelivered: Deferred.succeed(accepted, undefined), + }); + const answer = textUpdate({ + updateId: 1, + messageId: 5, + from: OWNER, + text: "Ship it", + replyTo: 1000, + }); + yield* Effect.gen(function* () { + const channel = yield* TelegramChannel.TelegramChannel; + yield* channel.send({ threadId: FIX_LOGIN.id, text: "Build finished" }); + bot.queueUpdates([answer]); + yield* channel.start(); + yield* Deferred.await(failed); + assert.deepEqual(delivered, []); + + // Never acknowledged, so Telegram hands the same update out again once the wait is over. + bot.queueUpdates([answer]); + yield* TestClock.adjust("5 seconds"); + yield* Deferred.await(accepted); + assert.equal(delivered.length, 1); + assert.equal(delivered[0]?.threadId, FIX_LOGIN.id); + }).pipe(Effect.scoped, Effect.provide(layer)); + }), + ); + + it.effect("tells the owner when a thread refuses a message, and moves on to the next one", () => { + const { bot, delivered, layer } = makeHarness({ threads: [FIX_LOGIN], rejectDeliveries: true }); + return Effect.gen(function* () { + const channel = yield* TelegramChannel.TelegramChannel; + yield* channel.send({ threadId: FIX_LOGIN.id, text: "Build finished" }); + bot.queueUpdates([ + textUpdate({ updateId: 1, messageId: 5, from: OWNER, text: "Ship it", replyTo: 1000 }), + textUpdate({ updateId: 2, messageId: 6, from: OWNER, text: "/threads" }), + ]); + yield* channel.receive; + yield* channel.receive; + + assert.deepEqual(delivered, []); + const [, refusal, listed] = bot.sent(); + assert.include(String(refusal?.body.text), "could not take that message"); + // The refused message did not hold up the command sent after it. + assert.include(String(listed?.body.text), "Fix login"); + const offsets = bot.calls + .filter((call) => call.method === "getUpdates") + .map((call) => call.body.offset); + assert.deepEqual(offsets, [undefined, 3]); + }).pipe(Effect.provide(layer)); + }); + + it.effect( + "gives up on a message that keeps failing instead of holding up the rest forever", + () => { + const { bot, delivered, layer } = makeHarness({ threads: [FIX_LOGIN], failDeliveries: 99 }); + const answer = textUpdate({ + updateId: 1, + messageId: 5, + from: OWNER, + text: "Ship it", + replyTo: 1000, + }); + return Effect.gen(function* () { + const channel = yield* TelegramChannel.TelegramChannel; + yield* channel.send({ threadId: FIX_LOGIN.id, text: "Build finished" }); + for (let attempt = 1; attempt < 5; attempt++) { + bot.queueUpdates([answer]); + yield* channel.receive.pipe(Effect.flip); + } + bot.queueUpdates([answer]); + yield* channel.receive; + yield* channel.receive; + + assert.deepEqual(delivered, []); + assert.equal(bot.calls.findLast((call) => call.method === "getUpdates")?.body.offset, 2); + // The owner is not left thinking the message went through. + assert.include(String(bot.sent().at(-1)?.body.text), "could not be handled"); + }).pipe(Effect.provide(layer)); + }, + ); + + it.effect("does not redeliver a command after a lost Telegram confirmation", () => { + const { bot, delivered, threads, layer } = makeHarness({ threads: [FIX_LOGIN] }); + return Effect.gen(function* () { + const channel = yield* TelegramChannel.TelegramChannel; + yield* channel.send({ threadId: FIX_LOGIN.id, text: "Ready" }); + const update = textUpdate({ + updateId: 7, + messageId: 5, + from: OWNER, + text: "Continue", + replyTo: 1000, + }); + bot.queueUpdates([update]); + bot.script("sendMessage", { + status: 500, + body: { ok: false, error_code: 500, description: "Temporary failure" }, + }); + yield* channel.receive.pipe(Effect.flip); + assert.equal(delivered.length, 1); + threads[0] = { ...threads[0]!, activeRunId: null }; + bot.queueUpdates([update]); + yield* channel.receive; + assert.equal(delivered.length, 1); + assert.include(String(bot.sent().at(-1)?.body.text), "Already sent"); + }).pipe(Effect.provide(layer)); + }); + + it.effect("delivers an owner's reply to the thread that sent the message", () => { + const { bot, delivered, layer } = makeHarness({ threads: [FIX_LOGIN] }); + return Effect.gen(function* () { + const channel = yield* TelegramChannel.TelegramChannel; + const sent = yield* channel.send({ threadId: FIX_LOGIN.id, text: "Build finished" }); + assert.deepEqual(sent, { + chatId: OWNER, + chatName: "Olivia Owner", + messageId: 1000, + replyAvailable: true, + }); + assert.deepEqual(bot.sent()[0]?.body, { chat_id: OWNER, text: "Build finished" }); + + bot.queueUpdates([ + textUpdate({ updateId: 7, messageId: 5, from: OWNER, text: "Ship it", replyTo: 1000 }), + ]); + yield* channel.receive; + + assert.equal(delivered.length, 1); + const delivery = delivered[0]!; + assert.equal(delivery.threadId, FIX_LOGIN.id); + assert.equal(delivery.projectId, PROJECT_ID); + assert.match(delivery.text, /^Sent from Telegram\./); + assert.isTrue(delivery.text.endsWith("\n\nShip it")); + assert.equal(delivery.createdBy, "user"); + assert.equal(delivery.creationSource, "server"); + assert.equal(delivery.mode, "auto"); + // Derived from the bot, the chat, and the message, never from the token's secret half. + assert.include(delivery.commandId, `123:${OWNER}:5`); + assert.notInclude(delivery.commandId, "secret"); + + const confirmation = bot.sent().at(-1)!; + assert.equal(confirmation.method, "sendMessage"); + assert.deepEqual(confirmation.body, { + chat_id: OWNER, + text: "Sent to “Fix login”.", + reply_parameters: { message_id: 5, allow_sending_without_reply: true }, + }); + + // The confirmation can be answered too, and continues the same thread. + bot.queueUpdates([ + textUpdate({ + updateId: 8, + messageId: 6, + from: OWNER, + text: "And the tests", + replyTo: confirmation.messageId!, + }), + ]); + yield* channel.receive; + assert.equal(delivered.length, 2); + assert.equal(delivered[1]?.threadId, FIX_LOGIN.id); + }).pipe(Effect.provide(layer)); + }); + + it.effect("ignores an owner answering inside a group, and anybody else there", () => { + const { bot, delivered, layer } = makeHarness({ + telegram: { chatIds: [OWNER, GROUP] }, + threads: [FIX_LOGIN], + }); + return Effect.gen(function* () { + const channel = yield* TelegramChannel.TelegramChannel; + yield* channel.send({ threadId: FIX_LOGIN.id, text: "Build finished", chatId: GROUP }); + bot.queueUpdates([ + textUpdate({ + updateId: 1, + messageId: 2, + from: OWNER, + chatId: GROUP, + text: "Ship it", + replyTo: 1000, + }), + textUpdate({ + updateId: 2, + messageId: 3, + from: STRANGER, + chatId: GROUP, + text: "Delete everything", + replyTo: 1000, + }), + ]); + yield* channel.receive; + + assert.deepEqual(delivered, []); + assert.deepEqual( + bot.sent().map((call) => call.method), + ["sendMessage"], + ); + }).pipe(Effect.provide(layer)); + }); + + it.effect("ignores a reply in a private chat from somebody who is not an owner", () => { + const { bot, delivered, layer } = makeHarness({ + telegram: { ownerIds: [] }, + threads: [FIX_LOGIN], + }); + return Effect.gen(function* () { + const channel = yield* TelegramChannel.TelegramChannel; + yield* channel.send({ threadId: FIX_LOGIN.id, text: "Build finished" }); + bot.queueUpdates([ + textUpdate({ updateId: 1, messageId: 2, from: OWNER, text: "Ship it", replyTo: 1000 }), + ]); + yield* channel.receive; + + assert.deepEqual(delivered, []); + assert.equal(bot.sent().length, 1); + }).pipe(Effect.provide(layer)); + }); + + it.effect("keeps a redelivered message's ids stable and acknowledges what it read", () => { + const { bot, delivered, layer } = makeHarness({ threads: [FIX_LOGIN] }); + return Effect.gen(function* () { + const channel = yield* TelegramChannel.TelegramChannel; + yield* channel.send({ threadId: FIX_LOGIN.id, text: "Build finished" }); + const reply = textUpdate({ + updateId: 7, + messageId: 5, + from: OWNER, + text: "Ship it", + replyTo: 1000, + }); + bot.queueUpdates([reply]); + bot.queueUpdates([reply]); + bot.queueUpdates([ + textUpdate({ updateId: 8, messageId: 6, from: OWNER, text: "Ship it", replyTo: 1000 }), + ]); + yield* channel.receive; + yield* channel.receive; + yield* channel.receive; + + assert.equal(delivered.length, 2); + assert.notEqual(delivered[1]?.commandId, delivered[0]?.commandId); + assert.include(String(bot.sent()[2]?.body.text), "Already sent"); + const offsets = bot.calls + .filter((call) => call.method === "getUpdates") + .map((call) => call.body.offset); + assert.deepEqual(offsets, [undefined, 8, 8]); + }).pipe(Effect.provide(layer)); + }); + + it.effect("uploads a file from the thread's worktree as a document or a photo", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const worktree = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-telegram-" }); + yield* fileSystem.writeFileString(path.join(worktree, "report.pdf"), "%PDF-1.4 report"); + yield* fileSystem.writeFile(path.join(worktree, "chart.png"), new Uint8Array([137, 80, 78])); + const worktreeThread = thread({ + id: "thread-report", + title: "Report", + worktreePath: worktree, + }); + const { bot, layer } = makeHarness({ threads: [worktreeThread] }); + + yield* Effect.gen(function* () { + const channel = yield* TelegramChannel.TelegramChannel; + yield* channel.send({ + threadId: worktreeThread.id, + path: "report.pdf", + text: "Weekly report", + }); + yield* channel.send({ threadId: worktreeThread.id, path: "chart.png" }); + const missing = yield* channel + .send({ threadId: worktreeThread.id, path: "missing.pdf", text: "Where" }) + .pipe(Effect.flip); + + const [document, photo] = bot.sent(); + assert.equal(document?.method, "sendDocument"); + assert.deepEqual(document?.body, { chat_id: String(OWNER), caption: "Weekly report" }); + assert.equal(document?.file?.name, "report.pdf"); + assert.equal(yield* Effect.promise(() => document!.file!.text()), "%PDF-1.4 report"); + assert.equal(photo?.method, "sendPhoto"); + assert.deepEqual(photo?.body, { chat_id: String(OWNER) }); + assert.equal(photo?.file?.name, "chart.png"); + + assert.equal(missing.reason, "file-unreadable"); + assert.equal(bot.sent().length, 2); + }).pipe(Effect.provide(layer)); + }).pipe(Effect.provide(NodeServices.layer)), + ); + + it.effect("keeps a thread without full access inside its workspace", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const home = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-telegram-" }); + const worktree = path.join(home, "project"); + yield* fileSystem.makeDirectory(worktree); + const secret = path.join(home, "id_rsa"); + yield* fileSystem.writeFileString(secret, "private key"); + yield* fileSystem.symlink(secret, path.join(worktree, "notes.txt")); + const worktreeThread = thread({ + id: "thread-report", + title: "Report", + worktreePath: worktree, + }); + const { bot, layer } = makeHarness({ threads: [worktreeThread] }); + + yield* Effect.gen(function* () { + const channel = yield* TelegramChannel.TelegramChannel; + // An absolute path, a `..` and a symlink all point at the same file outside. + for (const outside of [secret, "../id_rsa", "notes.txt"]) { + const refused = yield* channel + .send({ threadId: worktreeThread.id, path: outside }) + .pipe(Effect.flip); + assert.equal(refused.reason, "file-outside-workspace"); + } + assert.deepEqual(bot.sent(), []); + + // Its own files go out whichever way the path is written. + yield* fileSystem.writeFileString(path.join(worktree, "report.txt"), "report"); + yield* channel.send({ + threadId: worktreeThread.id, + path: path.join(worktree, "report.txt"), + }); + assert.equal(bot.sent().at(-1)?.file?.name, "report.txt"); + + // A thread with full access can already read any of them, so it may send them. + for (const outside of [secret, "../id_rsa", "notes.txt"]) { + yield* channel.send({ threadId: worktreeThread.id, path: outside, hostFiles: true }); + } + assert.equal(bot.sent().length, 4); + }).pipe(Effect.provide(layer)); + }).pipe(Effect.provide(NodeServices.layer)), + ); + + it.effect.each([ + { + name: "no token", + telegram: { botToken: "" }, + send: { text: "hi" }, + reason: "not-configured", + }, + { name: "no chat", telegram: { chatIds: [] }, send: { text: "hi" }, reason: "not-configured" }, + { + name: "a chat that is not in Settings", + telegram: {}, + send: { text: "hi", chatId: 99 }, + reason: "chat-not-allowed", + }, + { name: "nothing to send", telegram: {}, send: { text: " " }, reason: "empty" }, + ] as const)("refuses $name without calling Telegram", ({ telegram, send, reason }) => { + const { bot, layer } = makeHarness({ telegram, threads: [FIX_LOGIN] }); + return Effect.gen(function* () { + const channel = yield* TelegramChannel.TelegramChannel; + const error = yield* channel.send({ threadId: FIX_LOGIN.id, ...send }).pipe(Effect.flip); + assert.equal(error.reason, reason); + assert.deepEqual(bot.calls, []); + }).pipe(Effect.provide(layer)); + }); + + it.effect("lists open threads one message each, and a reply to one picks that thread", () => { + const threads = [ + thread({ id: "thread-oldest", title: "Oldest", updatedAtMillis: 1_000 }), + thread({ id: "thread-newest", title: "Newest", updatedAtMillis: 3_000 }), + thread({ id: "thread-middle", title: "Middle", updatedAtMillis: 2_000 }), + thread({ id: "thread-archived", title: "Archived", updatedAtMillis: 4_000, archived: true }), + thread({ id: "thread-deleted", title: "Deleted", updatedAtMillis: 5_000, deleted: true }), + thread({ id: "thread-subagent", title: "Subagent", updatedAtMillis: 6_000, subagent: true }), + ]; + const { bot, delivered, layer } = makeHarness({ threads }); + return Effect.gen(function* () { + const channel = yield* TelegramChannel.TelegramChannel; + bot.queueUpdates([textUpdate({ updateId: 1, messageId: 2, from: OWNER, text: "/threads" })]); + yield* channel.receive; + + const listed = bot.sent(); + // The most recent thread is sent last, so it sits nearest the keyboard. + assert.deepEqual( + listed.map((call) => String(call.body.text).split("\n")[0]), + ["Oldest", "Middle", "Newest"], + ); + const middle = listed[1]!; + + bot.queueUpdates([ + textUpdate({ + updateId: 2, + messageId: 3, + from: OWNER, + text: "Carry on", + replyTo: middle.messageId!, + }), + ]); + yield* channel.receive; + assert.equal(delivered.length, 1); + assert.equal(delivered[0]?.threadId, ThreadId.make("thread-middle")); + }).pipe(Effect.provide(layer)); + }); + + it.effect("sends to the one group a name fits, and asks which when it fits several", () => { + const { bot, layer } = makeHarness({ + telegram: { chatIds: [OWNER, GROUP, -1002, -1003] }, + threads: [FIX_LOGIN], + }); + return Effect.gen(function* () { + const channel = yield* TelegramChannel.TelegramChannel; + + const sent = yield* channel.send({ + threadId: FIX_LOGIN.id, + text: "Report ready", + chat: "leads", + }); + assert.equal(sent.chatId, -1002); + assert.equal(sent.chatName, "Team leads"); + // A group cannot answer, so the agent is told where an answer can come from. + assert.isFalse(sent.replyAvailable); + assert.equal(bot.sent().at(-1)?.body.chat_id, -1002); + + const several = yield* channel + .send({ threadId: FIX_LOGIN.id, text: "Report ready", chat: "team" }) + .pipe(Effect.flip); + assert.equal(several.reason, "chat-not-found"); + assert.include(several.message, "“Team” (-1001), “Team leads” (-1002)"); + + // A lookup that merely failed must not make "team" look like it fits only one chat. + bot.script("getChat", { + status: 429, + body: { ok: false, error_code: 429, description: "Too Many Requests" }, + }); + const unsure = yield* channel + .send({ threadId: FIX_LOGIN.id, text: "Report ready", chat: "team" }) + .pipe(Effect.flip); + assert.equal(unsure.reason, "rejected"); + + // -1003 is a chat the bot was removed from: it has no name, and is not offered. + const none = yield* channel + .send({ threadId: FIX_LOGIN.id, text: "Report ready", chat: "sales" }) + .pipe(Effect.flip); + assert.equal(none.reason, "chat-not-found"); + assert.include(none.message, "“Olivia Owner” (42), “Team” (-1001), “Team leads” (-1002)"); + assert.equal(bot.sent().filter((call) => call.method === "sendMessage").length, 1); + }).pipe(Effect.provide(layer)); + }); + + it.effect( + "says a group's id there when an owner adds the bot, and stays quiet for anybody else", + () => { + const { bot, delivered, layer } = makeHarness({}); + const added = (updateId: number, by: number) => ({ + update_id: updateId, + my_chat_member: { + chat: { id: GROUP, type: "supergroup", title: "Team" }, + from: { id: by, is_bot: false }, + old_chat_member: { status: "left" }, + new_chat_member: { status: "member" }, + }, + }); + return Effect.gen(function* () { + const channel = yield* TelegramChannel.TelegramChannel; + bot.queueUpdates([added(3, STRANGER), added(4, OWNER)]); + yield* channel.receive; + yield* channel.receive; + + assert.deepEqual(delivered, []); + const introductions = bot.sent(); + assert.equal(introductions.length, 1); + assert.equal(introductions[0]?.body.chat_id, GROUP); + assert.include(String(introductions[0]?.body.text), String(GROUP)); + const offsets = bot.calls + .filter((call) => call.method === "getUpdates") + .map((call) => call.body.offset); + assert.deepEqual(offsets, [undefined, 5]); + }).pipe(Effect.provide(layer)); + }, + ); + + it.effect("tells a stranger who sends /start the id to add in Settings", () => { + const { bot, delivered, layer } = makeHarness({}); + return Effect.gen(function* () { + const channel = yield* TelegramChannel.TelegramChannel; + bot.queueUpdates([textUpdate({ updateId: 1, messageId: 9, from: STRANGER, text: "/start" })]); + yield* channel.receive; + + assert.deepEqual(delivered, []); + const [introduction] = bot.sent(); + assert.equal(introduction?.body.chat_id, STRANGER); + assert.include(String(introduction?.body.text), String(STRANGER)); + }).pipe(Effect.provide(layer)); + }); + + it.effect("tests the token by greeting every chat and naming the bot", () => { + const { bot, layer } = makeHarness({ telegram: { chatIds: [OWNER, GROUP] } }); + return Effect.gen(function* () { + const channel = yield* TelegramChannel.TelegramChannel; + const result = yield* channel.sendTest(); + assert.deepEqual(result, { + botUsername: "t3_test_bot", + chats: [ + { chatId: OWNER, name: "Olivia Owner" }, + { chatId: GROUP, name: "Team" }, + ], + }); + assert.deepEqual( + bot.sent().map((call) => [call.method, call.body.chat_id]), + [ + ["getMe", undefined], + ["sendMessage", OWNER], + ["sendMessage", GROUP], + ], + ); + }).pipe(Effect.provide(layer)); + }); + + it.effect("tries every chat in a test and says which one it could not reach", () => { + const { bot, layer } = makeHarness({ telegram: { chatIds: [GROUP, OWNER] } }); + return Effect.gen(function* () { + bot.script("sendMessage", { + status: 403, + body: { ok: false, error_code: 403, description: "Forbidden: bot was kicked" }, + }); + const channel = yield* TelegramChannel.TelegramChannel; + const result = yield* channel.sendTest(); + assert.deepEqual(result.chats, [ + { chatId: GROUP, error: "Forbidden: bot was kicked" }, + { chatId: OWNER, name: "Olivia Owner" }, + ]); + }).pipe(Effect.provide(layer)); + }); + + it.effect("passes on Telegram's reason when it refuses the token", () => { + const { bot, layer } = makeHarness({}); + return Effect.gen(function* () { + bot.script("getMe", { + status: 401, + body: { ok: false, error_code: 401, description: "Unauthorized" }, + }); + const channel = yield* TelegramChannel.TelegramChannel; + const error = yield* channel.sendTest().pipe(Effect.flip); + assert.equal(error.reason, "rejected"); + assert.include(error.message, "Unauthorized"); + assert.equal(bot.sent().length, 1); + }).pipe(Effect.provide(layer)); + }); + + it.effect("does not let one unreadable update hold up the ones after it", () => { + const { bot, delivered, layer } = makeHarness({ threads: [FIX_LOGIN] }); + return Effect.gen(function* () { + const channel = yield* TelegramChannel.TelegramChannel; + yield* channel.send({ threadId: FIX_LOGIN.id, text: "Build finished" }); + bot.queueUpdates([ + { update_id: 7, message: { message_id: "not a number", chat: null } }, + textUpdate({ updateId: 8, messageId: 5, from: OWNER, text: "Ship it", replyTo: 1000 }), + { update_id: 9, message: { text: 42 } }, + ]); + yield* channel.receive; + yield* channel.receive; + + assert.equal(delivered.length, 1); + assert.equal(delivered[0]?.threadId, FIX_LOGIN.id); + // Unreadable updates are acknowledged too, or Telegram would send them again forever. + const offsets = bot.calls + .filter((call) => call.method === "getUpdates") + .map((call) => call.body.offset); + assert.deepEqual(offsets, [undefined, 10]); + }).pipe(Effect.provide(layer)); + }); +}); diff --git a/apps/server/src/telegram/TelegramChannel.ts b/apps/server/src/telegram/TelegramChannel.ts new file mode 100644 index 0000000000000..e47562ddcc969 --- /dev/null +++ b/apps/server/src/telegram/TelegramChannel.ts @@ -0,0 +1,688 @@ +import { + CommandId, + DEFAULT_SERVER_SETTINGS, + MessageId, + TelegramSendError, + type TelegramSendResult, + type TelegramSendTestResult, + type TelegramSettings, + type ThreadId, +} from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as DateTime from "effect/DateTime"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as Ref from "effect/Ref"; +import * as Schema from "effect/Schema"; +import type * as Scope from "effect/Scope"; +import * as Stream from "effect/Stream"; + +import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts"; +import * as TelegramMessages from "../persistence/TelegramMessages.ts"; +import * as Project from "../project/ProjectService.ts"; +import { forkParked } from "../serverActivation.ts"; +import * as ServerSettings from "../serverSettings.ts"; +import * as TelegramBotApi from "./TelegramBotApi.ts"; +import { + classifyIncomingMessage, + decodeIncomingMessage, + type IncomingMessage, + joinedGroupChatId, + TELEGRAM_CAPTION_MAX_CHARS, + TELEGRAM_FILE_MAX_BYTES, + TELEGRAM_HELP, + TELEGRAM_TEXT_MAX_CHARS, + TELEGRAM_TEXT_ONLY, + telegramBotId, + telegramChatName, + telegramChatsNamed, + telegramFileKind, + telegramFileMimeType, + telegramIntroduction, + telegramReplyAddress, + telegramThreadLine, + telegramThreadMessage, + telegramThreadTitle, +} from "./telegramRules.ts"; + +/** How long an answer to a message can still find its thread. */ +const MESSAGE_RETENTION_DAYS = 30; +/** How many threads `/threads` lists, most recent first. Each is a message of its own. */ +const LISTED_THREADS = 8; +const RETRY_MIN_SECONDS = 5; +const RETRY_MAX_SECONDS = 300; +/** Reads of the same update that may fail before it is dropped, so one cannot hold up the rest. */ +const UPDATE_ATTEMPTS = 5; + +export interface TelegramSendInput { + /** + * The thread the message is from. An answer to it in Telegram is delivered to this thread; + * a message from no thread cannot be answered. + */ + readonly threadId: ThreadId | undefined; + /** The message, or the caption when there is a file. */ + readonly text?: string | undefined; + /** A file to upload. A relative path is read from the thread's workspace. */ + readonly path?: string | undefined; + /** An explicit destination; otherwise the current Telegram sender, or the first configured chat. */ + readonly chatId?: number | undefined; + /** A configured chat by name, as the user said it: "the team group". `chatId` wins over it. */ + readonly chat?: string | undefined; + /** + * Whether `path` may leave the thread's workspace. Only a caller with full access gets this: + * the tool is not one a supervised agent is asked about, so it must not read past the + * workspace on its own. + */ + readonly hostFiles?: boolean | undefined; +} + +class TelegramReceiveError extends Schema.TaggedError()( + "TelegramReceiveError", + { + cause: Schema.Defect(), + }, +) {} + +/** + * The environment's Telegram bot: agents send through it, and the people allowed to answer + * reach their threads through it. + */ +export class TelegramChannel extends Context.Service< + TelegramChannel, + { + readonly send: ( + input: TelegramSendInput, + ) => Effect.Effect; + /** Checks the saved token and says hello in every configured chat. */ + readonly sendTest: () => Effect.Effect; + /** Handles the messages waiting for the bot, once. `start` keeps doing this. */ + readonly receive: Effect.Effect; + /** Listens for as long as the scope lives, following the token in Settings. */ + readonly start: () => Effect.Effect; + } +>()("t3/telegram/TelegramChannel") {} + +const make = Effect.gen(function* () { + const settingsService = yield* ServerSettings.ServerSettingsService; + const threads = yield* ThreadManagement.ThreadManagementService; + const projects = yield* Project.ProjectService; + const messages = yield* TelegramMessages.TelegramMessageRepository; + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const api = yield* TelegramBotApi.make; + // Which update to ask for next, and the token that is true of. Kept in memory only: after a + // restart Telegram resends what was never acknowledged, and delivery is keyed by the message, + // so a thread does not receive it twice. + const cursor = yield* Ref.make<{ readonly token: string; readonly offset: number } | null>(null); + /** Failed reads in a row, which is what spaces out the retries. */ + const failures = yield* Ref.make(0); + /** The update that last failed, and how many times, so a hopeless one is given up on. */ + const stuck = yield* Ref.make<{ readonly updateId: number; readonly attempts: number } | null>( + null, + ); + + // Read on every use so a token saved in Settings applies without a restart. + const currentSettings = settingsService.getSettings.pipe( + Effect.map((settings) => settings.telegram), + Effect.catch((error) => + // No cause: a settings decode error can quote a hand-edited token. + Effect.logWarning("failed to read Telegram settings", { operation: error.operation }).pipe( + Effect.as(DEFAULT_SERVER_SETTINGS.telegram), + ), + ), + ); + + /** Links a message the bot sent to a thread, so an answer to it has somewhere to go. */ + const remember = Effect.fn("TelegramChannel.remember")(function* (input: { + readonly token: string; + readonly chatId: number; + readonly messageId: number; + readonly threadId: ThreadId; + }) { + const now = yield* DateTime.now; + yield* messages.record({ + botId: telegramBotId(input.token), + chatId: input.chatId, + messageId: input.messageId, + threadId: input.threadId, + sentAt: DateTime.formatIso(now), + dropBefore: DateTime.formatIso(DateTime.subtract(now, { days: MESSAGE_RETENTION_DAYS })), + }); + }); + + // Use the durable input of the active run. A sticky per-thread destination could send a + // later desktop request to the previous Telegram owner, and an in-memory map loses restarts. + const replyAddress = Effect.fn("TelegramChannel.replyAddress")(function* (threadId: ThreadId) { + const thread = yield* threads.getThreadShell(threadId); + if (thread === null || thread.activeRunId === null) return null; + const records = yield* threads.getThreadRecords(threadId, ["messages"], { + messageRoles: ["user"], + messageRunIds: [thread.activeRunId], + }); + const latest = records.messages.at(-1); + return latest?.creationSource === "server" ? telegramReplyAddress(latest.id) : null; + }); + + const workspaceOf = Effect.fn("TelegramChannel.workspaceOf")(function* (threadId: ThreadId) { + const thread = yield* threads.getThreadShell(threadId); + if (thread === null) return null; + if (thread.worktreePath !== null) return thread.worktreePath; + const project = yield* projects.getById(thread.projectId); + return Option.isSome(project) ? project.value.workspaceRoot : null; + }); + + const readUpload = Effect.fn("TelegramChannel.readUpload")(function* ( + input: Pick, + requestedPath: string, + ) { + const unreadable = new TelegramSendError({ reason: "file-unreadable" }); + const outside = new TelegramSendError({ reason: "file-outside-workspace" }); + const realPath = (target: string) => + fileSystem.realPath(target).pipe(Effect.mapError(() => unreadable)); + // An absolute path is a host file read in place, such as a report an agent wrote to a temp + // directory; anything else is read from the thread's workspace. + const workspace = + input.threadId === undefined + ? null + : yield* workspaceOf(input.threadId).pipe(Effect.mapError(() => unreadable)); + if (workspace === null && !path.isAbsolute(requestedPath)) return yield* unreadable; + const filePath = yield* realPath(path.resolve(workspace ?? "", requestedPath)); + if (input.hostFiles !== true) { + if (workspace === null) return yield* outside; + // Compared after links are followed, so neither `..` nor a symlink leaves the workspace. + const inside = path.relative(yield* realPath(workspace), filePath); + if (inside === ".." || inside.startsWith(`..${path.sep}`) || path.isAbsolute(inside)) { + return yield* outside; + } + } + const info = yield* fileSystem.stat(filePath).pipe(Effect.mapError(() => unreadable)); + const sizeBytes = Number(info.size); + if (info.type !== "File" || sizeBytes === 0) return yield* unreadable; + if (sizeBytes > TELEGRAM_FILE_MAX_BYTES) { + return yield* new TelegramSendError({ reason: "file-too-large" }); + } + const bytes = yield* fileSystem.readFile(filePath).pipe(Effect.mapError(() => unreadable)); + const fileName = path.basename(requestedPath); + return { + kind: telegramFileKind(fileName, sizeBytes), + file: new File([bytes], fileName, { type: telegramFileMimeType(fileName) }), + }; + }); + + /** Telegram said no and will say no again: not a rate limit, a server error or a lost call. */ + const refusedForGood = (error: TelegramBotApi.TelegramApiError) => + error.reason === "rejected" && + error.errorCode !== undefined && + error.errorCode >= 400 && + error.errorCode < 500 && + error.errorCode !== 429; + + const sendFailure = (error: TelegramBotApi.TelegramApiError) => + new TelegramSendError({ + reason: error.reason, + ...(error.description === undefined ? {} : { description: error.description }), + }); + + /** + * The configured chat a user named. Settings hold ids only, so the names are read from + * Telegram; a chat the bot was removed from has none and cannot be picked this way. + */ + const chatNamed = Effect.fn("TelegramChannel.chatNamed")(function* ( + settings: TelegramSettings, + requested: string, + ) { + const chats = yield* Effect.forEach(settings.chatIds, (chatId) => + api.getChat(settings.botToken, chatId).pipe( + Effect.map((chat) => ({ chatId, name: telegramChatName(chat) })), + Effect.catchTags({ + // A chat the bot can no longer see has no name. A call that merely failed must not + // drop a chat from the choice, or an ambiguous name would look like it fits one. + TelegramApiError: (error) => + refusedForGood(error) + ? Effect.succeed({ chatId, name: null }) + : Effect.fail(sendFailure(error)), + }), + ), + ); + const fits = telegramChatsNamed(chats, requested); + if (fits.length === 1 && fits[0] !== undefined) return fits[0].chatId; + // With the ids, so the agent can ask which one and then send by `chatId`. + const listed = (fits.length > 1 ? fits : chats) + .flatMap((chat) => (chat.name === null ? [] : [`“${chat.name}” (${chat.chatId})`])) + .join(", "); + return yield* new TelegramSendError({ + reason: "chat-not-found", + description: + fits.length > 1 + ? `Several fit: ${listed}. Ask which one, then send with its chatId.` + : listed.length > 0 + ? `The chats there are ${listed}.` + : "None of the chats there has a name the bot can read.", + }); + }); + + const send: TelegramChannel["Service"]["send"] = Effect.fn("TelegramChannel.send")( + function* (input) { + const settings = yield* currentSettings; + if (settings.botToken.length === 0) { + return yield* new TelegramSendError({ reason: "not-configured" }); + } + const requestedChat = input.chat?.trim() ?? ""; + const explicitChatId = + input.chatId ?? + (requestedChat.length > 0 ? yield* chatNamed(settings, requestedChat) : undefined); + const reply = + input.threadId === undefined + ? null + : yield* replyAddress(input.threadId).pipe( + Effect.mapError(() => new TelegramSendError({ reason: "context-unavailable" })), + ); + const canReply = + reply !== null && + reply.botId === telegramBotId(settings.botToken) && + settings.ownerIds.includes(reply.chatId); + if (explicitChatId === undefined && reply !== null && !canReply) { + return yield* new TelegramSendError({ reason: "reply-unavailable" }); + } + const chatId = explicitChatId ?? reply?.chatId ?? settings.chatIds[0]; + if (chatId === undefined) return yield* new TelegramSendError({ reason: "not-configured" }); + // An owner can be answered without also being listed as a chat: they reach threads + // through /threads, and the agent is told their id with every message they send. + if (!settings.chatIds.includes(chatId) && !settings.ownerIds.includes(chatId)) { + return yield* new TelegramSendError({ reason: "chat-not-allowed" }); + } + const text = input.text?.trim() ?? ""; + const requestedPath = input.path?.trim() ?? ""; + const hasFile = requestedPath.length > 0; + if (text.length === 0 && !hasFile) return yield* new TelegramSendError({ reason: "empty" }); + if (text.length > (hasFile ? TELEGRAM_CAPTION_MAX_CHARS : TELEGRAM_TEXT_MAX_CHARS)) { + return yield* new TelegramSendError({ reason: "text-too-long" }); + } + const request = hasFile + ? api.sendFile(settings.botToken, { + chatId, + caption: text, + ...(yield* readUpload(input, requestedPath)), + }) + : api.sendMessage(settings.botToken, { chatId, text }); + const sent = yield* request.pipe(Effect.mapError(sendFailure)); + const chatName = sent.chat === undefined ? null : telegramChatName(sent.chat); + const linked = + input.threadId !== undefined && + (yield* remember({ + token: settings.botToken, + chatId, + messageId: sent.message_id, + threadId: input.threadId, + }).pipe( + Effect.as(true), + Effect.catch((cause) => + Effect.logWarning("failed to link a Telegram message to its thread", { + threadId: input.threadId, + cause, + }).pipe(Effect.as(false)), + ), + )); + const replyAvailable = linked && chatId > 0 && settings.ownerIds.includes(chatId); + return { + chatId, + ...(chatName === null ? {} : { chatName }), + messageId: sent.message_id, + replyAvailable, + ...(replyAvailable + ? {} + : { + warning: linked + ? "The message was delivered. Replies are enabled only in a configured owner's private chat. Use /threads there to select a thread." + : "The message was delivered, but replies cannot reach this thread. Do not resend it. Use /threads in the bot's private chat to select a thread.", + }), + }; + }, + ); + + const sendTest: TelegramChannel["Service"]["sendTest"] = Effect.fn("TelegramChannel.sendTest")( + function* () { + const settings = yield* currentSettings; + if (settings.botToken.length === 0) { + return yield* new TelegramSendError({ reason: "not-configured" }); + } + const bot = yield* api.getMe(settings.botToken).pipe(Effect.mapError(sendFailure)); + // Every chat is tried, so one the bot was removed from does not hide the rest. + const chats = yield* Effect.forEach(settings.chatIds, (chatId) => + api.sendMessage(settings.botToken, { chatId, text: "T3 Code can reach this chat." }).pipe( + Effect.map((sent) => { + const name = sent.chat === undefined ? null : telegramChatName(sent.chat); + return { chatId, ...(name === null ? {} : { name }) }; + }), + Effect.catchTags({ + TelegramApiError: (error) => + Effect.succeed({ chatId, error: error.description ?? "Telegram did not answer." }), + }), + ), + ); + return { botUsername: bot.username, chats }; + }, + ); + + /** Delivers an owner's answer to the thread it answers, and says in the chat how that went. */ + const deliver = Effect.fn("TelegramChannel.deliver")(function* (input: { + readonly token: string; + readonly message: IncomingMessage; + readonly replyToMessageId: number; + readonly text: string; + }) { + const botId = telegramBotId(input.token); + const chatId = input.message.chat.id; + const threadId = yield* messages.threadFor({ + botId, + chatId, + messageId: input.replyToMessageId, + }); + if (Option.isNone(threadId)) { + return { + text: "I don't know which thread that message is from. Send /threads to pick one.", + threadId: null, + }; + } + const thread = yield* threads.getThreadShell(threadId.value); + if (thread === null || thread.deletedAt !== null) { + return { text: "That thread no longer exists.", threadId: null }; + } + // Keyed by the Telegram message, so one Telegram resends after a restart is not sent twice. + const key = `${botId}:${chatId}:${input.message.message_id}`; + const messageId = MessageId.make(`telegram-message:${key}`); + const existing = yield* threads.getThreadRecords(thread.id, ["messages"], { + messageIds: [messageId], + }); + // A lost Telegram acknowledgement can replay this after the original run has finished. + // Do not ask sendToThread to choose a new dispatch mode for an already accepted input. + if (existing.messages.length > 0) { + return { + text: `Already sent to “${telegramThreadTitle(thread.title)}”.`, + threadId: thread.id, + }; + } + const text = yield* threads + .sendToThread({ + projectId: thread.projectId, + threadId: thread.id, + commandId: CommandId.make(`telegram:${key}`), + messageId, + text: telegramThreadMessage(input.text, chatId), + attachments: [], + mode: "auto", + createdBy: "user", + creationSource: "server", + }) + .pipe( + Effect.as(`Sent to “${telegramThreadTitle(thread.title)}”.`), + Effect.catchTags({ + ThreadManagementThreadArchivedError: () => + Effect.succeed("That thread is archived. Unarchive it in T3 Code to message it."), + }), + // A thread that could not be read may be readable on the next try. Anything the thread + // itself refused stays refused for this message, so the owner is told instead of + // Telegram being asked for it again. + Effect.catchIf( + (error) => + error._tag !== "ThreadManagementProjectionLoadError" && + error._tag !== "ThreadManagementDurableRunProjectionError", + (error) => + Effect.logWarning("a thread did not take a Telegram message", { + threadId: thread.id, + error: error._tag, + }).pipe( + Effect.as(`“${telegramThreadTitle(thread.title)}” could not take that message.`), + ), + ), + ); + return { text, threadId: thread.id }; + }); + + const listThreads = Effect.fn("TelegramChannel.listThreads")(function* () { + const snapshot = yield* threads.getShellSnapshot({ location: "active" }); + return ( + snapshot.threads + .filter( + (thread) => + thread.deletedAt === null && + thread.archivedAt === null && + thread.lineage.relationshipToParent !== "subagent", + ) + .toSorted( + (left, right) => + DateTime.toEpochMillis(right.updatedAt) - DateTime.toEpochMillis(left.updatedAt), + ) + .slice(0, LISTED_THREADS) + // Sent oldest first, so the most recent thread ends up nearest the keyboard. + .toReversed() + ); + }); + + const handleMessage = Effect.fn("TelegramChannel.handleMessage")(function* ( + settings: TelegramSettings, + message: IncomingMessage, + ) { + const token = settings.botToken; + const chatId = message.chat.id; + const action = classifyIncomingMessage(settings.ownerIds, message); + const answer = (text: string) => + api.sendMessage(token, { chatId, text, replyToMessageId: message.message_id }); + switch (action.type) { + case "ignore": + return; + case "introduce": { + const introduced = answer( + telegramIntroduction({ + chatId, + privateChat: message.chat.type === "private", + owner: action.owner, + }), + ); + // Anybody can send /start. If Telegram is slow to take the answer, an owner's + // messages must not wait behind a stranger's. + yield* action.owner + ? introduced + : introduced.pipe(Effect.catchTags({ TelegramApiError: () => Effect.void })); + return; + } + case "help": + yield* answer(TELEGRAM_HELP); + return; + case "text-only": + yield* answer(TELEGRAM_TEXT_ONLY); + return; + case "list-threads": { + const open = yield* listThreads(); + if (open.length === 0) { + yield* answer("There are no open threads."); + return; + } + // One message per thread: answering it is how that thread is picked. + for (const thread of open) { + const sent = yield* api.sendMessage(token, { + chatId, + text: telegramThreadLine({ title: thread.title, working: thread.activeRunId !== null }), + }); + yield* remember({ token, chatId, messageId: sent.message_id, threadId: thread.id }); + } + return; + } + case "deliver": { + const outcome = yield* deliver({ + token, + message, + replyToMessageId: action.replyToMessageId, + text: action.text, + }); + const sent = yield* answer(outcome.text); + if (outcome.threadId !== null) { + // The confirmation can be answered too, which continues the same thread. + yield* remember({ + token, + chatId, + messageId: sent.message_id, + threadId: outcome.threadId, + }); + } + return; + } + } + }); + + const fetchUpdates = Effect.fn("TelegramChannel.fetchUpdates")(function* (token: string) { + const seen = yield* Ref.get(cursor); + return yield* api.getUpdates(token, seen?.token === token ? seen.offset : undefined); + }); + + const handleUpdates = Effect.fn("TelegramChannel.handleUpdates")(function* ( + settings: TelegramSettings, + updates: ReadonlyArray<{ + readonly update_id: number; + readonly message?: unknown; + readonly my_chat_member?: unknown; + }>, + ) { + // A blocked chat or deleted reply cannot recover by retrying. Transport errors, rate + // limits, server errors and persistence failures leave this update unacknowledged. + const refusedOrRetry = (error: TelegramBotApi.TelegramApiError) => + refusedForGood(error) + ? Effect.logWarning("Telegram refused an incoming message response", { + errorCode: error.errorCode, + }) + : Effect.fail(error); + for (const update of updates) { + const current = yield* currentSettings; + if (current.botToken !== settings.botToken) return; + const joinedGroup = joinedGroupChatId(current.ownerIds, update.my_chat_member); + const message = decodeIncomingMessage(update.message); + const handled = Effect.gen(function* () { + if (joinedGroup !== null) { + // Says the group's id where it was added, so it can be copied into Settings. + yield* api.sendMessage(current.botToken, { + chatId: joinedGroup, + text: telegramIntroduction({ chatId: joinedGroup, privateChat: false, owner: false }), + }); + } + if (Option.isSome(message)) yield* handleMessage(current, message.value); + }).pipe(Effect.catchTags({ TelegramApiError: refusedOrRetry })); + yield* handled.pipe( + Effect.catch((error) => + Effect.gen(function* () { + const seen = yield* Ref.get(stuck); + const attempts = seen?.updateId === update.update_id ? seen.attempts + 1 : 1; + if (attempts < UPDATE_ATTEMPTS) { + yield* Ref.set(stuck, { updateId: update.update_id, attempts }); + return yield* Effect.fail(error); + } + // Acknowledged below: one message that cannot be handled must not stop the rest. + yield* Effect.logWarning("gave up on a Telegram update that kept failing", { + updateId: update.update_id, + error: error._tag, + }); + const lost = + Option.isSome(message) && + classifyIncomingMessage(current.ownerIds, message.value).type !== "ignore" + ? message.value + : null; + if (lost !== null) { + // Best effort: whatever broke the handling may break this too. + yield* api + .sendMessage(current.botToken, { + chatId: lost.chat.id, + text: "That message could not be handled. Send it again.", + replyToMessageId: lost.message_id, + }) + .pipe(Effect.catchTags({ TelegramApiError: () => Effect.void })); + } + }), + ), + ); + yield* Ref.set(cursor, { token: settings.botToken, offset: update.update_id + 1 }); + } + }); + + const receive: TelegramChannel["Service"]["receive"] = Effect.gen(function* () { + const settings = yield* currentSettings; + if (settings.botToken.length === 0) return; + yield* handleUpdates(settings, yield* fetchUpdates(settings.botToken)); + }).pipe( + Effect.mapError((cause) => + cause._tag === "TelegramApiError" ? cause : new TelegramReceiveError({ cause }), + ), + Effect.withSpan("TelegramChannel.receive"), + ); + + /** + * One wait for messages. The wait ends early when Settings change, so a new token is picked + * up at once rather than when the request Telegram is holding open runs out. Messages already + * read are handled to the end: a save must not cut a `/threads` list in half. + */ + const listenOnce = Effect.gen(function* () { + // Subscribed before the read, so a save that lands in between is not missed. + const changes = yield* settingsService.subscribeChanges; + // A stream that ends without a change is not a change: treating it as one would end every + // wait at once and turn this into a busy loop. + const settingsChanged = changes.pipe( + Stream.runHead, + Effect.flatMap((change) => (Option.isSome(change) ? Effect.void : Effect.never)), + ); + const settings = yield* currentSettings; + if (settings.botToken.length === 0) { + yield* settingsChanged; + return; + } + const failed = yield* Ref.get(failures); + yield* fetchUpdates(settings.botToken).pipe( + Effect.raceFirst(settingsChanged.pipe(Effect.as(null))), + Effect.flatMap((updates) => + updates === null ? Effect.void : handleUpdates(settings, updates), + ), + Effect.mapError((cause) => + cause._tag === "TelegramApiError" ? cause : new TelegramReceiveError({ cause }), + ), + Effect.withSpan("TelegramChannel.receive"), + Effect.andThen(Ref.set(failures, 0)), + Effect.catch((error) => + Effect.gen(function* () { + yield* Ref.set(failures, failed + 1); + yield* Effect.logWarning("Telegram will retry receiving messages", { + error: error._tag, + // Never a Bot API error, which is the only one that could quote the token's URL. + ...(error._tag === "TelegramApiError" + ? { errorCode: error.errorCode } + : { cause: error.cause }), + }); + const backoff = Math.min(RETRY_MAX_SECONDS, RETRY_MIN_SECONDS * 2 ** failed); + const retryAfter = error._tag === "TelegramApiError" ? (error.retryAfterSeconds ?? 0) : 0; + // The wait Telegram asks for is not ours to cut short; the rest of it is. + yield* Effect.sleep(Duration.seconds(retryAfter)); + yield* Effect.sleep(Duration.seconds(Math.max(0, backoff - retryAfter))).pipe( + Effect.raceFirst(settingsChanged), + ); + }), + ), + ); + }).pipe(Effect.scoped); + + const start: TelegramChannel["Service"]["start"] = () => + forkParked( + Effect.forever( + listenOnce.pipe( + Effect.catchCause((cause) => + Effect.logError("the Telegram listener failed and will start again", { cause }).pipe( + Effect.andThen(Effect.sleep(Duration.seconds(RETRY_MIN_SECONDS))), + ), + ), + ), + ), + ); + + return TelegramChannel.of({ send, sendTest, receive, start }); +}); + +export const layer = Layer.effect(TelegramChannel, make); diff --git a/apps/server/src/telegram/telegramRules.test.ts b/apps/server/src/telegram/telegramRules.test.ts new file mode 100644 index 0000000000000..02483f0d200bf --- /dev/null +++ b/apps/server/src/telegram/telegramRules.test.ts @@ -0,0 +1,176 @@ +import { describe, expect, it } from "vite-plus/test"; + +import { + classifyIncomingMessage, + type IncomingMessage, + joinedGroupChatId, + telegramBotId, + telegramChatName, + telegramChatsNamed, + telegramFileKind, +} from "./telegramRules.ts"; + +const OWNER = 42; +const STRANGER = 77; +const GROUP = -1001; + +const message = (input: { + readonly from: number; + readonly chat?: { readonly id: number; readonly type: string }; + readonly text?: string; + readonly replyTo?: number; + readonly isBot?: boolean; +}): IncomingMessage => ({ + message_id: 500, + chat: input.chat ?? { id: input.from, type: "private" }, + from: { id: input.from, is_bot: input.isBot ?? false }, + ...(input.text === undefined ? {} : { text: input.text }), + ...(input.replyTo === undefined ? {} : { reply_to_message: { message_id: input.replyTo } }), +}); + +const classify = (input: Parameters[0]) => + classifyIncomingMessage([OWNER], message(input)); + +describe("classifyIncomingMessage", () => { + it("ignores a stranger's text, even when it answers a thread's message", () => { + expect(classify({ from: STRANGER, text: "hello" })).toEqual({ type: "ignore" }); + expect(classify({ from: STRANGER, text: "run rm -rf", replyTo: 10 })).toEqual({ + type: "ignore", + }); + }); + + it("introduces a stranger who sends /start without treating them as an owner", () => { + expect(classify({ from: STRANGER, text: "/start" })).toEqual({ + type: "introduce", + owner: false, + }); + }); + + it("ignores an owner answering a thread's message inside a group", () => { + expect( + classify({ from: OWNER, chat: { id: GROUP, type: "supergroup" }, text: "go", replyTo: 10 }), + ).toEqual({ type: "ignore" }); + }); + + it("lets an owner in their private chat list threads and start", () => { + expect(classify({ from: OWNER, text: "/threads" })).toEqual({ type: "list-threads" }); + expect(classify({ from: OWNER, text: "/start@some_bot" })).toEqual({ + type: "introduce", + owner: true, + }); + }); + + it("delivers an owner's reply to the message it answers", () => { + expect(classify({ from: OWNER, text: " ship it ", replyTo: 10 })).toEqual({ + type: "deliver", + replyToMessageId: 10, + text: "ship it", + }); + }); + + it("answers an owner's plain text with help instead of picking a thread", () => { + expect(classify({ from: OWNER, text: "ship it" })).toEqual({ type: "help" }); + }); + + it("tells an owner that only text reaches a thread", () => { + expect(classify({ from: OWNER, replyTo: 10 })).toEqual({ type: "text-only" }); + }); + + it("answers an unknown command with help, even as a reply", () => { + expect(classify({ from: OWNER, text: "/send 2 ship it", replyTo: 10 })).toEqual({ + type: "help", + }); + }); + + it("ignores other bots", () => { + expect(classify({ from: OWNER, isBot: true, text: "/start" })).toEqual({ type: "ignore" }); + }); +}); + +describe("telegramFileKind", () => { + const MB = 1024 * 1024; + + it("sends small images as photos and large ones as documents", () => { + expect(telegramFileKind("chart.jpg", 2 * MB)).toBe("photo"); + expect(telegramFileKind("CHART.PNG", 2 * MB)).toBe("photo"); + expect(telegramFileKind("chart.jpg", 11 * MB)).toBe("document"); + }); + + it("sends mp4 as video and anything else as a document", () => { + expect(telegramFileKind("demo.mp4", 20 * MB)).toBe("video"); + expect(telegramFileKind("report.pdf", MB)).toBe("document"); + }); +}); + +describe("telegramBotId", () => { + it("is the public half of a token", () => { + expect(telegramBotId("123:abc")).toBe("123"); + }); + + it("never returns any part of a malformed token", () => { + for (const token of ["secret-only", ":secret", "secret:123"]) { + expect(telegramBotId(token)).toBe(""); + } + }); +}); + +describe("joinedGroupChatId", () => { + const change = (chatType: string, from: string, to: string, addedBy = OWNER) => ({ + chat: { id: GROUP, type: chatType, title: "Team" }, + from: { id: addedBy, is_bot: false }, + old_chat_member: { status: from }, + new_chat_member: { status: to }, + }); + const joined = (update: unknown) => joinedGroupChatId([OWNER], update); + + it("is the group an owner just added the bot to", () => { + expect(joined(change("group", "left", "member"))).toBe(GROUP); + expect(joined(change("supergroup", "kicked", "administrator"))).toBe(GROUP); + }); + + it("is nothing when somebody else added the bot, so a stranger cannot make it speak", () => { + expect(joined(change("group", "left", "member", STRANGER))).toBeNull(); + }); + + it("is nothing when the bot was already there, left, or joined a channel", () => { + expect(joined(change("supergroup", "member", "administrator"))).toBeNull(); + expect(joined(change("group", "member", "left"))).toBeNull(); + expect(joined(change("channel", "left", "administrator"))).toBeNull(); + expect(joined(undefined)).toBeNull(); + }); +}); + +describe("telegramChatName", () => { + it("is a group's title or a person's name", () => { + expect(telegramChatName({ id: GROUP, type: "supergroup", title: "Team" })).toBe("Team"); + expect( + telegramChatName({ id: OWNER, type: "private", first_name: "Ada", last_name: "Diaz" }), + ).toBe("Ada Diaz"); + expect(telegramChatName({ id: OWNER, type: "private", username: "ada" })).toBe("ada"); + expect(telegramChatName({ id: OWNER, type: "private" })).toBeNull(); + }); +}); + +describe("telegramChatsNamed", () => { + const chats = [ + { chatId: OWNER, name: "Ada Diaz" }, + { chatId: GROUP, name: "Team" }, + { chatId: -1002, name: "Team leads" }, + { chatId: -1003, name: null }, + ]; + + it("finds the one chat a name or part of a name fits", () => { + expect(telegramChatsNamed(chats, " LEADS ")).toEqual([chats[2]]); + expect(telegramChatsNamed(chats, "ada")).toEqual([chats[0]]); + }); + + it("returns every chat a name fits, so a renamed group cannot take another's place", () => { + // Group members set the title. "Team" must not win over "Team leads" by being shorter. + expect(telegramChatsNamed(chats, "team")).toEqual([chats[1], chats[2]]); + }); + + it("finds nothing for a name no chat has, or an empty one", () => { + expect(telegramChatsNamed(chats, "sales")).toEqual([]); + expect(telegramChatsNamed(chats, " ")).toEqual([]); + }); +}); diff --git a/apps/server/src/telegram/telegramRules.ts b/apps/server/src/telegram/telegramRules.ts new file mode 100644 index 0000000000000..bb6513553f2d5 --- /dev/null +++ b/apps/server/src/telegram/telegramRules.ts @@ -0,0 +1,182 @@ +import * as Schema from "effect/Schema"; + +import type { TelegramChat, TelegramFileKind } from "./TelegramBotApi.ts"; + +/** Telegram's own limits for a bot. */ +export const TELEGRAM_TEXT_MAX_CHARS = 4096; +export const TELEGRAM_CAPTION_MAX_CHARS = 1024; +export const TELEGRAM_FILE_MAX_BYTES = 50 * 1024 * 1024; +const TELEGRAM_PHOTO_MAX_BYTES = 10 * 1024 * 1024; + +const IncomingMessage = Schema.Struct({ + message_id: Schema.Int, + chat: Schema.Struct({ id: Schema.Int, type: Schema.String }), + from: Schema.optional(Schema.Struct({ id: Schema.Int, is_bot: Schema.Boolean })), + text: Schema.optional(Schema.String), + reply_to_message: Schema.optional(Schema.Struct({ message_id: Schema.Int })), +}); +export type IncomingMessage = typeof IncomingMessage.Type; + +/** Reads the parts of a Telegram message the channel acts on. */ +export const decodeIncomingMessage = Schema.decodeUnknownOption(IncomingMessage); + +const MembershipChange = Schema.Struct({ + chat: Schema.Struct({ id: Schema.Int, type: Schema.String }), + from: Schema.Struct({ id: Schema.Int }), + old_chat_member: Schema.Struct({ status: Schema.String }), + new_chat_member: Schema.Struct({ status: Schema.String }), +}); + +const decodeMembershipChange = Schema.decodeUnknownOption(MembershipChange); + +const isMember = (status: string) => status === "member" || status === "administrator"; + +/** + * The group an owner just added the bot to, read from a `my_chat_member` update. The bot says + * the group's id there, which is the number Settings asks for. Anybody can add a bot to a + * group, so only an owner's doing makes it speak; a channel is left alone, because a post + * there reaches every subscriber. + */ +export function joinedGroupChatId(ownerIds: ReadonlyArray, update: unknown): number | null { + const change = decodeMembershipChange(update); + if (change._tag === "None") return null; + const { chat, from, old_chat_member, new_chat_member } = change.value; + if (chat.type !== "group" && chat.type !== "supergroup") return null; + if (!ownerIds.includes(from.id)) return null; + return !isMember(old_chat_member.status) && isMember(new_chat_member.status) ? chat.id : null; +} + +/** What to call a chat: a group's title, or a person's name. */ +export function telegramChatName(chat: TelegramChat): string | null { + const person = [chat.first_name, chat.last_name] + .filter((part) => part !== undefined && part.trim().length > 0) + .join(" "); + const name = chat.title?.trim() || person.trim() || chat.username?.trim() || ""; + return name.length > 0 ? name : null; +} + +/** + * The configured chats a user could mean by a name, as in "send it to the team group". Group + * titles are set by their members, so a name is only good when it fits one chat: the caller + * sends to a single match and asks again for anything else. + */ +export function telegramChatsNamed( + chats: ReadonlyArray, + requested: string, +): ReadonlyArray { + const wanted = requested.trim().toLowerCase(); + if (wanted.length === 0) return []; + return chats.filter((chat) => chat.name !== null && chat.name.toLowerCase().includes(wanted)); +} + +export type IncomingAction = + /** Not for the channel: a bot, a stranger, or an owner writing outside their private chat. */ + | { readonly type: "ignore" } + /** `/start` from anybody, answered with the id Settings asks for. */ + | { readonly type: "introduce"; readonly owner: boolean } + | { readonly type: "list-threads" } + | { readonly type: "help" } + /** An owner sent something other than text, which a thread cannot receive yet. */ + | { readonly type: "text-only" } + | { readonly type: "deliver"; readonly replyToMessageId: number; readonly text: string }; + +function commandName(text: string): string | null { + // In a group Telegram appends the bot's name: `/start@my_bot`. + const match = /^\/([a-z_]+)(?:@\w+)?(?:\s|$)/i.exec(text); + return match?.[1]?.toLowerCase() ?? null; +} + +/** + * Decides what one incoming message means. Only an owner writing in their own private chat can + * reach a thread: what they type becomes a user message for an agent that can run commands, so + * a group, where anybody can answer, never counts, even when an owner is the one answering. + */ +export function classifyIncomingMessage( + ownerIds: ReadonlyArray, + message: IncomingMessage, +): IncomingAction { + const { chat, from } = message; + if (from === undefined || from.is_bot) return { type: "ignore" }; + const text = message.text?.trim() ?? ""; + const command = commandName(text); + const owner = chat.type === "private" && chat.id === from.id && ownerIds.includes(from.id); + if (command === "start") return { type: "introduce", owner }; + if (!owner) return { type: "ignore" }; + if (command === "threads") return { type: "list-threads" }; + if (command !== null) return { type: "help" }; + if (text.length === 0) return { type: "text-only" }; + if (message.reply_to_message === undefined) return { type: "help" }; + return { type: "deliver", replyToMessageId: message.reply_to_message.message_id, text }; +} + +/** The Bot API upload a file goes through, chosen by what Telegram can show inline. */ +export function telegramFileKind(fileName: string, sizeBytes: number): TelegramFileKind { + const extension = fileName.slice(fileName.lastIndexOf(".") + 1).toLowerCase(); + if (["jpg", "jpeg", "png", "webp"].includes(extension)) { + // A larger image is refused as a photo but accepted as a file. + return sizeBytes <= TELEGRAM_PHOTO_MAX_BYTES ? "photo" : "document"; + } + return extension === "mp4" ? "video" : "document"; +} + +const MIME_TYPES: Record = { + jpg: "image/jpeg", + jpeg: "image/jpeg", + png: "image/png", + webp: "image/webp", + mp4: "video/mp4", + pdf: "application/pdf", +}; + +export function telegramFileMimeType(fileName: string): string { + const extension = fileName.slice(fileName.lastIndexOf(".") + 1).toLowerCase(); + return MIME_TYPES[extension] ?? "application/octet-stream"; +} + +/** + * The bot a token belongs to. The id is the public half of the token, and is what keeps one + * bot's message ids apart from another's after the token in Settings is replaced. + */ +export function telegramBotId(token: string): string { + return /^(\d+):/.exec(token)?.[1] ?? ""; +} + +export const TELEGRAM_HELP = + "Reply to a message from a thread to answer it. Send /threads to see the open threads and reply to the one you want."; + +export const TELEGRAM_TEXT_ONLY = "Only text reaches a thread for now."; + +/** What an agent reads above a message that came in from Telegram. */ +export function telegramThreadMessage(text: string, chatId: number): string { + return `Sent from Telegram. Answer with telegram_send using chatId ${chatId}.\n\n${text}`; +} + +/** Read the server-assigned message id, never user-written text, to choose a reply destination. */ +export function telegramReplyAddress(messageId: string): { botId: string; chatId: number } | null { + const match = /^telegram-message:(\d+):(\d+):\d+$/.exec(messageId); + if (match === null) return null; + const chatId = Number(match[2]); + return Number.isSafeInteger(chatId) && chatId > 0 ? { botId: match[1]!, chatId } : null; +} + +/** A thread as the chat names it. */ +export function telegramThreadTitle(title: string): string { + return title.trim() || "Untitled thread"; +} + +export function telegramThreadLine(thread: { + readonly title: string; + readonly working: boolean; +}): string { + return `${telegramThreadTitle(thread.title)}\n${thread.working ? "Working" : "Idle"} · reply here to message it`; +} + +export function telegramIntroduction(input: { + readonly chatId: number; + readonly privateChat: boolean; + readonly owner: boolean; +}): string { + if (input.owner) return `T3 Code is connected to this chat. ${TELEGRAM_HELP}`; + const subject = input.privateChat ? "Your Telegram id" : "This chat's id"; + return `${subject} is ${input.chatId}. Add it in T3 Code under Settings → Integrations → Telegram.`; +} diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index af3fc819a4c71..4bd9d84451aea 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -123,6 +123,7 @@ import * as ThreadMessageIntake from "./orchestration-v2/ThreadMessageIntake.ts" import * as IdAllocator from "./orchestration-v2/IdAllocator.ts"; import * as ScheduledTasks from "./scheduledTasks/ScheduledTaskService.ts"; import * as SecretRequests from "./secrets/SecretRequests.ts"; +import * as TelegramChannel from "./telegram/TelegramChannel.ts"; import { archivedShellStreamItemFromThreadShell, buildActiveShellSnapshot, @@ -1225,6 +1226,7 @@ const layerWsRpc = ( const pullRequests = yield* PullRequestService.PullRequestService; const pullRequestSync = yield* PullRequestSyncReactor.PullRequestSyncReactor; const deviceService = yield* DeviceService.DeviceService; + const telegram = yield* TelegramChannel.TelegramChannel; const deviceHostContext = yield* Effect.context>>(); const orchestrationEngine = yield* Orchestrator.OrchestratorV2; @@ -2386,6 +2388,7 @@ const layerWsRpc = ( : serverSettings.updateProviderInstance(providerInstanceMutation, nextPatch); return ServerSettings.redactServerSettingsForClient(settings); }), + [WS_METHODS.serverTestTelegram]: (_input) => telegram.sendTest(), [WS_METHODS.serverDiscoverSourceControl]: (_input) => sourceControlDiscovery.discover, [WS_METHODS.serverGetTraceDiagnostics]: (_input) => TraceDiagnostics.readTraceDiagnostics({ @@ -3116,6 +3119,7 @@ export const layer = Layer.unwrap( const serverBrowser = yield* ServerBrowser.ServerBrowser; const serverSelfUpdate = yield* ServerSelfUpdate.ServerSelfUpdate; const pullRequests = yield* PullRequestService.PullRequestService; + const telegram = yield* TelegramChannel.TelegramChannel; const sql = yield* SqlClient.SqlClient; return HttpRouter.add( "GET", @@ -3182,6 +3186,8 @@ export const layer = Layer.unwrap( // One server-lifetime service means clients share the same PR caches, and a WS // mutation invalidates the HTTP diff cache that every client reads from. Layer.provide(Layer.succeed(PullRequestService.PullRequestService, pullRequests)), + // The Settings test reaches the same bot the listener and the agents' tool use. + Layer.provide(Layer.succeed(TelegramChannel.TelegramChannel, telegram)), Layer.provide( SourceControlDiscovery.layer.pipe( Layer.provide( diff --git a/apps/web/src/components/settings/BitbucketCredentialsSettings.tsx b/apps/web/src/components/settings/BitbucketCredentialsSettings.tsx index a423334d1c7f9..983d1b5788a2c 100644 --- a/apps/web/src/components/settings/BitbucketCredentialsSettings.tsx +++ b/apps/web/src/components/settings/BitbucketCredentialsSettings.tsx @@ -44,7 +44,7 @@ function savedMethod(saved: BitbucketSettings): CredentialMethod | null { } /** A write-only token field. It never shows the saved token; typing a new one replaces it. */ -function TokenInput({ +export function TokenInput({ id, isSaved, draft, diff --git a/apps/web/src/components/settings/IntegrationsSettings.tsx b/apps/web/src/components/settings/IntegrationsSettings.tsx index a0c0a79bebc6e..a4933363d75c8 100644 --- a/apps/web/src/components/settings/IntegrationsSettings.tsx +++ b/apps/web/src/components/settings/IntegrationsSettings.tsx @@ -3,6 +3,7 @@ import { DeviceToolVersions } from "../device/DeviceToolVersions"; import { useScopedSettings, useUpdateScopedSettings } from "./useScopedSettings"; import { ScopedSwitch } from "./ScopedSwitch"; import { DeviceHostsSettings } from "./DeviceHostsSettings"; +import { TelegramSettingsSection } from "./TelegramSettings"; /** * Integrations settings - preferences for surfaces T3 Code embeds rather than * owns. Browser is the first section: the defaults a preview tab opens at, @@ -1537,6 +1538,7 @@ export function IntegrationsSettingsPanel() { {/* Server-authoritative agent access is scoped by the header selection; the preview defaults below are device-local and ignore it. */} + {previewDefaultsDisabled ? ( diff --git a/apps/web/src/components/settings/TelegramSettings.logic.test.ts b/apps/web/src/components/settings/TelegramSettings.logic.test.ts new file mode 100644 index 0000000000000..a5cf1ffc08a04 --- /dev/null +++ b/apps/web/src/components/settings/TelegramSettings.logic.test.ts @@ -0,0 +1,133 @@ +import type { TelegramSettings } from "@t3tools/contracts"; +import { describe, expect, it } from "vite-plus/test"; + +import { + parseTelegramIdList, + resolveTelegramDraft, + telegramSettingsSaved, + telegramTestMessage, +} from "./TelegramSettings.logic"; + +const empty: TelegramSettings = { botToken: "", chatIds: [], ownerIds: [] }; +const saved: TelegramSettings = { + botToken: "", + chatIds: [111, -1002233], + ownerIds: [111], +}; +const untouched = { botToken: "", chatIds: null, ownerIds: null }; + +describe("parseTelegramIdList", () => { + it("accepts ids separated by commas, spaces, or both, and keeps their order", () => { + expect(parseTelegramIdList(" 42,-1001234\n7 , 9 ", "chat")).toEqual({ + ok: true, + ids: [42, -1001234, 7, 9], + }); + expect(parseTelegramIdList(" ", "chat")).toEqual({ ok: true, ids: [] }); + }); + + it("drops repeats so the first chat stays the default", () => { + expect(parseTelegramIdList("5, 3, 5", "chat")).toEqual({ ok: true, ids: [5, 3] }); + }); + + it.each(["0", "-0", "1.5", "1e3", "abc", "+12", "12a", "99999999999999999999"])( + "rejects %s", + (entry) => { + expect(parseTelegramIdList(`1, ${entry}`, "chat")).toEqual({ + ok: false, + error: `${entry} is not a Telegram id.`, + }); + }, + ); + + it("rejects a group or channel where only people fit", () => { + expect(parseTelegramIdList("-1001234", "chat").ok).toBe(true); + expect(parseTelegramIdList("7, -1001234", "person")).toEqual({ + ok: false, + error: "-1001234 is a group or channel. Only people can answer.", + }); + }); +}); + +describe("resolveTelegramDraft", () => { + it("has nothing to save until something differs from the server", () => { + expect(resolveTelegramDraft(saved, untouched)).toMatchObject({ dirty: false, patch: null }); + // Retyping the same ids with other separators is not a change. + expect( + resolveTelegramDraft(saved, { botToken: " ", chatIds: "111 -1002233", ownerIds: "111," }), + ).toMatchObject({ dirty: false, patch: null }); + }); + + it("keeps the saved token unless a new one is typed", () => { + expect(resolveTelegramDraft(saved, { ...untouched, chatIds: "-1002233, 111" }).patch).toEqual({ + chatIds: [-1002233, 111], + ownerIds: [111], + }); + expect(resolveTelegramDraft(empty, { ...untouched, botToken: " 123:abc " }).patch).toEqual({ + botToken: "123:abc", + chatIds: [], + ownerIds: [], + }); + }); + + it("blocks saving while a field is invalid and says which", () => { + expect( + resolveTelegramDraft(saved, { botToken: "123:abc", chatIds: "1.5", ownerIds: "-5" }), + ).toEqual({ + chatIdsError: "1.5 is not a Telegram id.", + ownerIdsError: "-5 is a group or channel. Only people can answer.", + dirty: true, + patch: null, + }); + }); + + it("saves cleared lists", () => { + expect(resolveTelegramDraft(saved, { ...untouched, ownerIds: "" }).patch).toEqual({ + chatIds: [111, -1002233], + ownerIds: [], + }); + }); +}); + +describe("telegramSettingsSaved", () => { + it("is true while a token or any id is saved", () => { + expect(telegramSettingsSaved(empty)).toBe(false); + expect(telegramSettingsSaved({ ...empty, ownerIds: [1] })).toBe(true); + expect(telegramSettingsSaved({ ...empty, botToken: "" })).toBe(true); + }); +}); + +describe("telegramTestMessage", () => { + it("names the chats the test reached, so a group's id can be told from a person's", () => { + expect( + telegramTestMessage({ botUsername: "t3_bot", chats: [{ chatId: 1, name: "Ada Diaz" }] }), + ).toEqual({ tone: "info", message: "@t3_bot reached Ada Diaz." }); + expect( + telegramTestMessage({ + botUsername: "t3_bot", + chats: [{ chatId: 1, name: "Ada Diaz" }, { chatId: -2, name: "Team" }, { chatId: -3 }], + }).message, + ).toBe("@t3_bot reached Ada Diaz, Team, and -3."); + }); + + it("says which chat it could not reach, and why, next to the ones it did", () => { + expect( + telegramTestMessage({ + botUsername: "t3_bot", + chats: [ + { chatId: 1, name: "Ada Diaz" }, + { chatId: -2, error: "Forbidden: bot was kicked" }, + ], + }), + ).toEqual({ + tone: "error", + message: "@t3_bot reached Ada Diaz. @t3_bot could not reach -2 (Forbidden: bot was kicked).", + }); + }); + + it("points to /start when no chat is set up yet", () => { + expect(telegramTestMessage({ botUsername: "t3_bot", chats: [] })).toEqual({ + tone: "info", + message: "@t3_bot is connected. Send /start to it in Telegram to get your chat id.", + }); + }); +}); diff --git a/apps/web/src/components/settings/TelegramSettings.logic.ts b/apps/web/src/components/settings/TelegramSettings.logic.ts new file mode 100644 index 0000000000000..50341a7c026b8 --- /dev/null +++ b/apps/web/src/components/settings/TelegramSettings.logic.ts @@ -0,0 +1,119 @@ +import type { + ServerSettingsPatch, + TelegramSendTestResult, + TelegramSettings, +} from "@t3tools/contracts"; + +type TelegramPatch = NonNullable; + +export type TelegramIdListParse = + | { readonly ok: true; readonly ids: ReadonlyArray } + | { readonly ok: false; readonly error: string }; + +/** + * Reads ids typed into one field, separated by commas or spaces. Chats may be + * negative (groups and channels); people never are. Repeats are dropped and + * the order is kept, because the first chat is the default. + */ +export function parseTelegramIdList(text: string, kind: "chat" | "person"): TelegramIdListParse { + const ids: number[] = []; + for (const entry of text.split(/[\s,]+/)) { + if (entry === "") continue; + const id = /^-?\d+$/.test(entry) ? Number(entry) : Number.NaN; + if (!Number.isSafeInteger(id) || id === 0) { + return { ok: false, error: `${entry} is not a Telegram id.` }; + } + if (kind === "person" && id < 0) { + return { ok: false, error: `${entry} is a group or channel. Only people can answer.` }; + } + if (!ids.includes(id)) ids.push(id); + } + return { ok: true, ids }; +} + +export function formatTelegramIdList(ids: ReadonlyArray): string { + return ids.join(", "); +} + +export interface TelegramDraft { + /** A new token, or empty to keep the saved one. */ + readonly botToken: string; + /** Typed chat ids, or null while the field still shows the saved list. */ + readonly chatIds: string | null; + /** Typed owner ids, or null while the field still shows the saved list. */ + readonly ownerIds: string | null; +} + +export interface TelegramDraftState { + readonly chatIdsError: string | null; + readonly ownerIdsError: string | null; + /** Something differs from what the server has, valid or not. */ + readonly dirty: boolean; + /** What Save sends, or null when there is nothing valid to save. */ + readonly patch: TelegramPatch | null; +} + +function sameIds(left: ReadonlyArray, right: ReadonlyArray): boolean { + return left.length === right.length && left.every((id, index) => id === right[index]); +} + +/** Compares the form with the saved settings. The saved token is never shown, only replaced. */ +export function resolveTelegramDraft( + saved: TelegramSettings, + draft: TelegramDraft, +): TelegramDraftState { + const botToken = draft.botToken.trim(); + const chats = parseTelegramIdList(draft.chatIds ?? formatTelegramIdList(saved.chatIds), "chat"); + const owners = parseTelegramIdList( + draft.ownerIds ?? formatTelegramIdList(saved.ownerIds), + "person", + ); + const chatsChanged = !chats.ok || !sameIds(chats.ids, saved.chatIds); + const ownersChanged = !owners.ok || !sameIds(owners.ids, saved.ownerIds); + const dirty = botToken !== "" || chatsChanged || ownersChanged; + return { + chatIdsError: chats.ok ? null : chats.error, + ownerIdsError: owners.ok ? null : owners.error, + dirty, + patch: + dirty && chats.ok && owners.ok + ? { + ...(botToken === "" ? {} : { botToken }), + chatIds: [...chats.ids], + ownerIds: [...owners.ids], + } + : null, + }; +} + +export function telegramSettingsSaved(saved: TelegramSettings): boolean { + return saved.botToken !== "" || saved.chatIds.length > 0 || saved.ownerIds.length > 0; +} + +/** What a test found, said in one line. Any chat it missed makes it an error. */ +export function telegramTestMessage(result: TelegramSendTestResult): { + readonly tone: "info" | "error"; + readonly message: string; +} { + const bot = `@${result.botUsername}`; + if (result.chats.length === 0) { + return { + tone: "info", + message: `${bot} is connected. Send /start to it in Telegram to get your chat id.`, + }; + } + const list = new Intl.ListFormat("en", { type: "conjunction" }); + // Names show which id is which chat, which matters once a group is in the list. + const reached = result.chats + .filter((chat) => chat.error === undefined) + .map((chat) => chat.name ?? String(chat.chatId)); + const missed = result.chats.flatMap((chat) => + chat.error === undefined ? [] : [`${chat.chatId} (${chat.error})`], + ); + const reachedLine = reached.length > 0 ? `${bot} reached ${list.format(reached)}.` : ""; + if (missed.length === 0) return { tone: "info", message: reachedLine }; + return { + tone: "error", + message: `${reachedLine} ${bot} could not reach ${list.format(missed)}.`.trim(), + }; +} diff --git a/apps/web/src/components/settings/TelegramSettings.tsx b/apps/web/src/components/settings/TelegramSettings.tsx new file mode 100644 index 0000000000000..21f4e1c7ea27a --- /dev/null +++ b/apps/web/src/components/settings/TelegramSettings.tsx @@ -0,0 +1,257 @@ +import type { EnvironmentId, ServerSettingsPatch } from "@t3tools/contracts"; +import { ExternalLinkIcon } from "lucide-react"; +import { useState } from "react"; + +import { useEnvironmentSettings } from "../../hooks/useSettings"; +import { cn } from "../../lib/utils"; +import { formatEnvironmentQueryError } from "../../state/query"; +import { serverEnvironment } from "../../state/server"; +import { useAtomCommand } from "../../state/use-atom-command"; +import { Button, InlineButton } from "../ui/button"; +import { Input } from "../ui/input"; +import { Label } from "../ui/label"; +import { TokenInput } from "./BitbucketCredentialsSettings"; +import { SettingsRow, SettingsSection } from "./settingsLayout"; +import { searchableSetting } from "./settingsSearch"; +import { useSettingsScope } from "./SettingsScopeContext"; +import { + formatTelegramIdList, + resolveTelegramDraft, + telegramSettingsSaved, + telegramTestMessage, + type TelegramDraft, +} from "./TelegramSettings.logic"; + +const BOT_TUTORIAL_URL = "https://core.telegram.org/bots/tutorial#obtain-your-bot-token"; +const UNTOUCHED: TelegramDraft = { botToken: "", chatIds: null, ownerIds: null }; + +/** + * The Telegram channel of the selected environment. Like Source Control, it + * edits the representative environment and names it when several are selected. + */ +export function TelegramSettingsSection() { + const { scope, environment, connectedEnvironments } = useSettingsScope(); + const environmentId = environment?.environmentId ?? null; + const aggregate = scope.environmentIds.length !== 1 && connectedEnvironments.length > 1; + const { id, title } = searchableSetting("telegram"); + + return ( + + {environmentId === null ? ( +

+ Connect an environment to set up Telegram. +

+ ) : ( + + )} + + ); +} + +type Status = { readonly tone: "info" | "error"; readonly message: string }; + +function IdListField({ + id, + label, + hint, + placeholder, + value, + error, + onChange, +}: { + readonly id: string; + readonly label: string; + readonly hint: string; + readonly placeholder: string; + readonly value: string; + readonly error: string | null; + readonly onChange: (value: string) => void; +}) { + return ( +
+ + onChange(event.target.value)} + /> +

+ {error ?? hint} +

+
+ ); +} + +/** + * Bot token, chats agents may send to, and the people who may answer. The + * token is write-only: the server keeps it in its secret store and reports + * only whether one is set. A test goes to the saved settings, so it waits for + * pending edits to be saved. + */ +function TelegramSettings({ + environmentId, + environmentWide, +}: { + readonly environmentId: EnvironmentId; + readonly environmentWide: boolean; +}) { + const saved = useEnvironmentSettings(environmentId, (settings) => settings.telegram); + const updateSettings = useAtomCommand(serverEnvironment.updateSettings, { + label: "save Telegram settings", + }); + const testTelegram = useAtomCommand(serverEnvironment.testTelegram, { reportFailure: false }); + const [draft, setDraft] = useState(UNTOUCHED); + const [pending, setPending] = useState<"save" | "test" | null>(null); + const [status, setStatus] = useState(null); + const { chatIdsError, ownerIdsError, dirty, patch } = resolveTelegramDraft(saved, draft); + const tokenSaved = saved.botToken !== ""; + const fieldId = (name: string) => `telegram-${name}-${environmentId}`; + + const edit = (next: Partial) => { + setDraft((current) => ({ ...current, ...next })); + // A result describes the saved settings; an edit makes it stale. + setStatus(null); + }; + + const save = async (telegram: NonNullable) => { + setPending("save"); + setStatus(null); + try { + const result = await updateSettings({ environmentId, input: { patch: { telegram } } }); + if (result._tag === "Success") setDraft(UNTOUCHED); + else setStatus({ tone: "error", message: "Could not save the Telegram settings." }); + } finally { + setPending(null); + } + }; + + const sendTest = async () => { + setPending("test"); + setStatus(null); + try { + const result = await testTelegram({ environmentId, input: {} }); + setStatus( + result._tag === "Success" + ? telegramTestMessage(result.value) + : { tone: "error", message: formatEnvironmentQueryError(result.cause) }, + ); + } finally { + setPending(null); + } + }; + + const footer: Status | null = environmentWide + ? { tone: "info", message: "Environment-wide setting. Select an environment to change it." } + : (status ?? (tokenSaved && dirty ? { tone: "info", message: "Save to send a test." } : null)); + + return ( + + Agents can send you messages and files on Telegram, and you can answer by replying. Create + a separate bot with @BotFather, paste its token and save first. Then send /start to that + bot to get your id. Add your id to both fields below and save again.{" "} + } + > + How to create a bot + + + + } + > +
{ + event.preventDefault(); + if (patch && pending === null) void save(patch); + }} + > + {/* Locked while saving: a successful save clears the drafts, which would drop edits made mid-request. */} +
+
+ + edit({ botToken })} + /> +
+ edit({ chatIds })} + /> + edit({ ownerIds })} + /> +
+

+ {footer?.message} +

+
+ {telegramSettingsSaved(saved) ? ( + + ) : null} + + +
+
+
+
+ + ); +} diff --git a/apps/web/src/components/settings/settingsSearch.ts b/apps/web/src/components/settings/settingsSearch.ts index 853886d81fcd9..78212c8f29d9c 100644 --- a/apps/web/src/components/settings/settingsSearch.ts +++ b/apps/web/src/components/settings/settingsSearch.ts @@ -623,6 +623,13 @@ export const SETTINGS_SEARCH_ITEMS = [ searchTerms: ["refresh availability versions auth state models background probes seconds off"], providerSettingsOnly: true, }, + { + id: "telegram", + title: "Telegram", + to: "/settings/integrations", + searchTerms: ["bot token chat owner reply threads send pdf image video document message"], + environmentOnly: true, + }, { id: "agent-browser-access", title: "Agent browser access", diff --git a/docs/README.md b/docs/README.md index ead465ccce039..1414212a942e9 100644 --- a/docs/README.md +++ b/docs/README.md @@ -15,6 +15,7 @@ - [Visual replies](./user/html-renders.md) - [Import browser sessions](./user/browser-import.md) - [Devices](./user/devices.md) +- [Telegram](./user/telegram.md) - [Usage and limits](./user/usage.md) - [Product usage data](./user/telemetry.md) - [Remote access](./user/remote-access.md) diff --git a/docs/user/telegram.md b/docs/user/telegram.md new file mode 100644 index 0000000000000..0f4d04dc256bd --- /dev/null +++ b/docs/user/telegram.md @@ -0,0 +1,36 @@ +# Telegram + +A thread can send you text, an image, a video, or a file such as a PDF on Telegram, and you can answer that message from your phone. The answer comes back into the same thread. + +## Set it up + +Open **Settings → Integrations → Telegram**. This belongs to the environment you have selected. + +1. In Telegram, talk to [@BotFather](https://t.me/BotFather) and create a new bot for this T3 environment. Do not share it with another running bot service or another T3 environment. +2. Paste its token into **Bot token** and choose **Save**. You can leave the ids empty for this first save. The token is stored on the server and is not shown again. +3. Open a private chat with that bot and send `/start`. It replies with your Telegram id. Put that number in **Chats** and **Who can answer**, then save again. +4. Choose **Send test**. The bot says hello in each chat you listed. This checks sending; complete the reply check below to check both directions. + +Set this up on desktop or web. The mobile app shows what a thread sent to Telegram, but has no screen for the bot. + +## Groups + +A group can receive what agents send. It cannot answer them. + +1. Add the bot to the group. It posts the group's id there, a negative number. It only does this when someone in **Who can answer** adds it; otherwise send `/start@your_bot` in the group. +2. Add that id to **Chats** and save. **Send test** then names every chat it reached, so you can tell the ids apart. +3. Ask for the group by name: "send the report to the team group". The name has to fit one chat in **Chats**; if it fits several, the agent asks which. With no chat named, an answer to your Telegram message goes back to you, and any other send goes to the first chat in the list. + +The bot finds no groups on its own and adds none to the list. It sends only to the ids in **Chats** and to the people in **Who can answer**, so a group someone else put the bot in receives nothing until you add it. + +Answering works only from a private chat, and only for the people in **Who can answer**. What they type becomes a message to an agent that can run commands on this machine, so a reply inside a group is ignored, even when one of those people writes it. Leave **Who can answer** empty to turn answering off. Everyone listed can reach all open threads in this environment. + +## Send and answer + +Ask the agent to send something on Telegram. A JPG, PNG, or WebP image, and an MP4 video, show in the chat. Anything else, including a PDF, arrives as a file. A file can be up to 50 MB. A thread sends files from its own workspace; only a thread with full access can send a file from elsewhere on the machine. + +To check replies, ask a thread to send you a short message, then reply to it in your private chat with the bot. Your text appears in that thread, the agent continues, and its answer comes back to your private chat. + +Send `/threads` to see the eight most recently updated open threads. Each one arrives as its own message. Reply to the one you want. `/start` gives a short reminder of these commands, or your id if you are not in **Who can answer** yet. A message older than 30 days may have lost its thread; use `/threads` to get a fresh one. + +The environment server has to be running for an answer to arrive, and Telegram keeps a message for it for up to 24 hours. Nothing is sent on its own: a result reaches Telegram when you ask the agent to send it, and only text comes back in. diff --git a/packages/client-runtime/src/state/server.ts b/packages/client-runtime/src/state/server.ts index 17037af3c9868..91db8dc048348 100644 --- a/packages/client-runtime/src/state/server.ts +++ b/packages/client-runtime/src/state/server.ts @@ -1173,6 +1173,10 @@ export function createServerEnvironmentAtoms( concurrency: configConcurrency, }), updateSettings, + testTelegram: createEnvironmentRpcCommand(runtime, { + label: "environment-data:server:test-telegram", + tag: WS_METHODS.serverTestTelegram, + }), // Provider-instance mutations share the settings command and its // environment-serial scheduler. The named boundary keeps clients on the // atomic map-entry payload instead of rebuilding a stale whole map. diff --git a/packages/client-runtime/src/t3ToolSummary.test.ts b/packages/client-runtime/src/t3ToolSummary.test.ts index 8ae001ea77691..5aafad9879343 100644 --- a/packages/client-runtime/src/t3ToolSummary.test.ts +++ b/packages/client-runtime/src/t3ToolSummary.test.ts @@ -201,6 +201,24 @@ describe("summarizeT3ToolCalls", () => { }); }); + it("counts delivered Telegram messages and keeps rejected sends out of the count", () => { + const rejected = completed( + { text: "Report ready" }, + { _tag: "TelegramSendError", reason: "chat-not-allowed" }, + ); + expect( + summarizeT3ToolCalls("telegram-send", [ + completed({ text: "Report ready" }, { chatId: 42, messageId: 1 }), + completed({ path: "/tmp/report.pdf" }, { chatId: 42, messageId: 2 }), + rejected, + ]), + ).toEqual({ label: "Sent 2 Telegram messages", failedCount: 1 }); + expect(summarizeT3ToolCalls("telegram-send", [rejected])).toEqual({ + label: "Tried to send 1 Telegram message", + failedCount: 1, + }); + }); + it("does not confuse a child's failure or wait timeout with failure of the orchestration call", () => { const failedChild = { taskId: "task-1", status: "failed", summary: "command not found" }; expect( diff --git a/packages/client-runtime/src/t3ToolSummary.ts b/packages/client-runtime/src/t3ToolSummary.ts index b9a46fa65287f..30061cfc428fe 100644 --- a/packages/client-runtime/src/t3ToolSummary.ts +++ b/packages/client-runtime/src/t3ToolSummary.ts @@ -404,6 +404,9 @@ export function summarizeT3ToolCalls( case "html-render": label = phrase("Rendered", "render", quantity(selected.length, "HTML page")); break; + case "telegram-send": + label = phrase("Sent", "send", quantity(selected.length, "Telegram message")); + break; case "capabilities": label = phrase("Checked", "check", `orchestration capabilities ${times}`); break; diff --git a/packages/client-runtime/src/work-log/presentation.ts b/packages/client-runtime/src/work-log/presentation.ts index f5e9daab74901..ece15e275da4a 100644 --- a/packages/client-runtime/src/work-log/presentation.ts +++ b/packages/client-runtime/src/work-log/presentation.ts @@ -632,6 +632,7 @@ function summaryActionPriority(action: ToolGroupAction | T3McpToolSummaryAction) case "attachment-prepare": case "attachment-discard": case "attachment-send": + case "telegram-send": return 0; case "other": case "update": diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index d738ef79909e2..b63eb33122c35 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -58,6 +58,7 @@ export * from "./previewAutomation.ts"; export * from "./resourceTelemetry.ts"; export * from "./usage.ts"; export * from "./scheduledTask.ts"; +export * from "./telegram.ts"; export * from "./worktreeMcp.ts"; export * from "./resourceTelemetry.ts"; export * from "./rpc.ts"; diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 20645ac4755b6..0da667e7c9fbb 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -312,6 +312,7 @@ import { } from "./providerUsageLimits.ts"; import { UsagePricing, UsageReadError, UsageSummary, UsageSummaryInput } from "./usage.ts"; import { ServerSettings, ServerSettingsError, ServerSettingsPatch } from "./settings.ts"; +import { TelegramSendError, TelegramSendTestResult } from "./telegram.ts"; import { ScheduledTaskDeleteInput, ScheduledTaskDeleteResult, @@ -463,6 +464,7 @@ export const WS_METHODS = { serverRemoveKeybinding: "server.removeKeybinding", serverGetSettings: "server.getSettings", serverUpdateSettings: "server.updateSettings", + serverTestTelegram: "server.testTelegram", serverDiscoverSourceControl: "server.discoverSourceControl", serverSearchAcpRegistry: "server.searchAcpRegistry", serverPrepareAcpRegistryAgent: "server.prepareAcpRegistryAgent", @@ -738,6 +740,12 @@ const WsServerUpdateSettingsRpc = Rpc.make(WS_METHODS.serverUpdateSettings, { error: Schema.Union([ServerSettingsError, EnvironmentAuthorizationError]), }); +const WsServerTestTelegramRpc = Rpc.make(WS_METHODS.serverTestTelegram, { + payload: Schema.Struct({}), + success: TelegramSendTestResult, + error: Schema.Union([TelegramSendError, EnvironmentAuthorizationError]), +}); + const WsServerDiscoverSourceControlRpc = Rpc.make(WS_METHODS.serverDiscoverSourceControl, { payload: Schema.Struct({}), success: SourceControlDiscoveryResult, @@ -1807,6 +1815,7 @@ export const WsRpcGroup = RpcGroup.make( WsServerRemoveKeybindingRpc, WsServerGetSettingsRpc, WsServerUpdateSettingsRpc, + WsServerTestTelegramRpc, WsServerDiscoverSourceControlRpc, WsServerSearchAcpRegistryRpc, WsServerPrepareAcpRegistryAgentRpc, diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index 942fbb915ffe3..47b6688bc4b34 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -17,6 +17,7 @@ import { TrimmedString, } from "./baseSchemas.ts"; import { UsageLimitSourceId } from "./usageLimitSourceId.ts"; +import { TelegramId } from "./telegram.ts"; import { EnvironmentMachineKind, ThreadEnvMode, WorktreeSubmodules } from "./environment.ts"; import { KeybindingShortcut } from "./keybindings.ts"; import { @@ -1049,6 +1050,21 @@ export const GitHubSettings = Schema.Struct({ }); export type GitHubSettings = typeof GitHubSettings.Type; +/** + * The Telegram channel of this environment. Agents send text and files to + * `chatIds`, the first being the default. People in `ownerIds` can answer from + * their private chat with the bot, and their text enters the thread as a user + * message, so an empty list turns answering off. The bot token lives in the + * server's secret store; settings and clients only see a redaction marker when + * one is set. + */ +export const TelegramSettings = Schema.Struct({ + botToken: TrimmedString.pipe(Schema.withDecodingDefault(Effect.succeed(""))), + chatIds: Schema.Array(TelegramId).pipe(Schema.withDecodingDefault(Effect.succeed([]))), + ownerIds: Schema.Array(TelegramId).pipe(Schema.withDecodingDefault(Effect.succeed([]))), +}); +export type TelegramSettings = typeof TelegramSettings.Type; + export const ObservabilitySettings = Schema.Struct({ otlpTracesUrl: TrimmedString.pipe(Schema.withDecodingDefault(Effect.succeed(""))), otlpMetricsUrl: TrimmedString.pipe(Schema.withDecodingDefault(Effect.succeed(""))), @@ -1460,6 +1476,7 @@ export const ServerSettings = Schema.Struct({ observability: ObservabilitySettings.pipe(Schema.withDecodingDefault(Effect.succeed({}))), bitbucket: BitbucketSettings.pipe(Schema.withDecodingDefault(Effect.succeed({}))), github: GitHubSettings.pipe(Schema.withDecodingDefault(Effect.succeed({}))), + telegram: TelegramSettings.pipe(Schema.withDecodingDefault(Effect.succeed({}))), // Keyed by a user-chosen id so a source keeps its rows across edits. Entries // this build cannot decode round-trip untouched, as provider instances do. usageLimitSources: Schema.Record(UsageLimitSourceId, UsageLimitSourceConfig).pipe( @@ -1761,6 +1778,14 @@ export const ServerSettingsPatch = Schema.Struct({ tokens: Schema.optionalKey(Schema.Record(GitHubHost, TrimmedString)), }), ), + /** An empty token clears it; an omitted one keeps what the server has. */ + telegram: Schema.optionalKey( + Schema.Struct({ + botToken: Schema.optionalKey(TrimmedString), + chatIds: Schema.optionalKey(Schema.Array(TelegramId)), + ownerIds: Schema.optionalKey(Schema.Array(TelegramId)), + }), + ), providers: Schema.optionalKey( Schema.Struct({ codex: Schema.optionalKey(CodexSettingsPatch), diff --git a/packages/contracts/src/telegram.ts b/packages/contracts/src/telegram.ts new file mode 100644 index 0000000000000..1cf357d9174a4 --- /dev/null +++ b/packages/contracts/src/telegram.ts @@ -0,0 +1,80 @@ +import * as Schema from "effect/Schema"; + +/** A Telegram chat or user id. Groups and channels are negative. */ +export const TelegramId = Schema.Int.check( + Schema.makeFilter((id) => id !== 0, { expected: "a non-zero Telegram id" }), +); +export type TelegramId = typeof TelegramId.Type; + +const TELEGRAM_SEND_FAILURES = { + "not-configured": + "Telegram is not set up. Add a bot token and a chat under Settings → Integrations → Telegram.", + "chat-not-allowed": "That chat is not one of the Telegram chats in Settings.", + "chat-not-found": "No Telegram chat in Settings goes by that name.", + "reply-unavailable": + "The Telegram reply destination is no longer available. Check the bot and owner ids in Settings; the reply was not sent to another chat.", + "context-unavailable": "The thread's Telegram reply destination could not be read. Try again.", + empty: "There is nothing to send. Give a message, a file, or both.", + "text-too-long": + "The text is over Telegram's limit: 4096 characters for a message, 1024 next to a file.", + "file-unreadable": "The file could not be read.", + "file-outside-workspace": + "That file is outside this thread's workspace. Only a thread with full access can send it.", + "file-too-large": "The file is over Telegram's 50 MB upload limit.", + rejected: "Telegram rejected the message.", + unreachable: + "Telegram did not answer. The message may or may not have been sent; check the chat before sending it again.", +} as const; + +/** Why a message did not reach Telegram. */ +export class TelegramSendError extends Schema.TaggedError()( + "TelegramSendError", + { + reason: Schema.Literals([ + "not-configured", + "chat-not-allowed", + "chat-not-found", + "reply-unavailable", + "context-unavailable", + "empty", + "text-too-long", + "file-unreadable", + "file-outside-workspace", + "file-too-large", + "rejected", + "unreachable", + ]), + /** What Telegram answered, in its own words, when it rejected the request. */ + description: Schema.optionalKey(Schema.String), + }, +) { + override get message(): string { + const base = TELEGRAM_SEND_FAILURES[this.reason]; + return this.description === undefined ? base : `${base} ${this.description}`; + } +} + +export const TelegramSendResult = Schema.Struct({ + chatId: TelegramId, + /** The group's title or the person's name, when Telegram gave one. */ + chatName: Schema.optionalKey(Schema.String), + messageId: Schema.Int, + /** Delivery and reply routing are separate: do not resend an already delivered file. */ + replyAvailable: Schema.Boolean, + warning: Schema.optionalKey(Schema.String), +}); +export type TelegramSendResult = typeof TelegramSendResult.Type; + +export const TelegramSendTestResult = Schema.Struct({ + /** The bot the saved token belongs to, without the `@`. */ + botUsername: Schema.String, + /** Every configured chat, by name, and why the test did not reach the ones it missed. */ + chats: Schema.Array( + Schema.Struct({ + chatId: TelegramId, + name: Schema.optionalKey(Schema.String), + error: Schema.optionalKey(Schema.String), + }), + ), +}); +export type TelegramSendTestResult = typeof TelegramSendTestResult.Type; diff --git a/packages/shared/src/t3McpToolPresentation.ts b/packages/shared/src/t3McpToolPresentation.ts index 11abf77f704be..7e5533ee3a112 100644 --- a/packages/shared/src/t3McpToolPresentation.ts +++ b/packages/shared/src/t3McpToolPresentation.ts @@ -61,7 +61,8 @@ export type T3McpToolSummaryAction = | "browser" | "device" | "html-preview" - | "html-render"; + | "html-render" + | "telegram-send"; export interface T3McpToolDefinition { readonly displayName: string; @@ -315,6 +316,7 @@ const T3_MCP_TOOLS: Readonly> = { t3_thread_send_attachments: tool(["Send", "Sending", "Sent", "attachments"], "attachment-send"), html_preview: tool(["Preview", "Previewing", "Previewed", "an HTML page"], "html-preview"), html_render: tool(["Render", "Rendering", "Rendered", "an HTML page"], "html-render"), + telegram_send: tool(["Send", "Sending", "Sent", "a Telegram message"], "telegram-send"), }; /**