diff --git a/.github/workflows/helm-boundary-check.yml b/.github/workflows/helm-boundary-check.yml index 78603fc..489f171 100644 --- a/.github/workflows/helm-boundary-check.yml +++ b/.github/workflows/helm-boundary-check.yml @@ -2,12 +2,14 @@ name: HELM Boundary Check on: pull_request: - push: - branches: - - main + +concurrency: + group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true jobs: validate: + timeout-minutes: 20 runs-on: ubuntu-latest steps: - uses: actions/checkout@v5 @@ -15,6 +17,8 @@ jobs: - uses: actions/setup-node@v5 with: node-version: "22" + cache: "npm" + cache-dependency-path: "packages/js/helm-tool-wrapper/package-lock.json" - uses: actions/setup-python@v6 with: @@ -28,6 +32,14 @@ jobs: run: npm test working-directory: packages/js/helm-tool-wrapper + - name: Install channel bridge dependencies + run: npm install + working-directory: packages/js/helm-channel-bridge + + - name: Test channel bridge + run: npm test + working-directory: packages/js/helm-channel-bridge + - name: Test Python wrapper run: python -m unittest discover packages/python/helm_tool_wrapper/tests diff --git a/Makefile b/Makefile index a2d6455..19c24cc 100644 --- a/Makefile +++ b/Makefile @@ -4,6 +4,7 @@ validate: test-js test-python samples verify-samples test-js: cd packages/js/helm-tool-wrapper && npm install && npm test + cd packages/js/helm-channel-bridge && npm install && npm test test-python: python3 -m unittest discover packages/python/helm_tool_wrapper/tests diff --git a/integrations/telegram/README.md b/integrations/telegram/README.md new file mode 100644 index 0000000..a1c1412 --- /dev/null +++ b/integrations/telegram/README.md @@ -0,0 +1,19 @@ +# Telegram + HELM + +Telegram gives operators a phone surface for human-in-the-loop control of +governed agent sessions. HELM evaluates every inbound command before it may +execute — fail closed, receipted, deny by default. + +Use `ChannelBridge`, `TelegramTransport`, and `createKernelEvaluator(...)` +from `@mindburn/helm-channel-bridge` (`packages/js/helm-channel-bridge`). + +- Bot token via the `HELM_TELEGRAM_BOT_TOKEN` environment variable only. +- DMs only, explicit chat-ID allowlist, empty allowlist denies everyone. +- Every command materializes as a Kernel-evaluated turn; unknown commands are + denied. `autoPermission` only for explicitly allowlisted routine read-only + commands (`help`/`list`/`status` by default). +- `ask_human` questions are relayed to the chat; answers are Kernel-evaluated + before being routed back into the suspended turn. + +WhatsApp/baileys is intentionally not supported (unofficial protocol, +account-ban and ToS risk). diff --git a/packages/js/helm-channel-bridge/README.md b/packages/js/helm-channel-bridge/README.md new file mode 100644 index 0000000..f3f0a97 --- /dev/null +++ b/packages/js/helm-channel-bridge/README.md @@ -0,0 +1,99 @@ +# @mindburn/helm-channel-bridge + +HELM-governed channel bridge: human-in-the-loop by phone. Inbound chat +commands (Telegram today) become Kernel-evaluated governed turns; pending +`ask_human` questions are relayed to the phone and the answers routed back — +every hop receipted by the HELM AI Kernel. + +This is a HELM-compatible example adapter. Command-bridge and transport +mechanisms are adapted from the Apache-2.0 +[Rowboat](https://github.com/rowboatlabs/rowboat) project's ChannelBridge and +Telegram transport; the implementation here is original and adds HELM +governance semantics Rowboat does not have (Rowboat runs channel turns with +`autoPermission: true` and no per-command policy evaluation). + +## Governance model (fail closed) + +- **Every inbound command is Kernel-evaluated.** `help`, `list`, `status`, + `resume`, `new`, `stop`, chat turns, and `ask_human` answers each produce a + `/api/v1/evaluate` preflight with a distinct action URN + (`channel..command.`, `channel..turn.run`, + `channel..ask_human.answer`). Only an explicit `ALLOW` dispatches. +- **Deny by default.** Unknown verdicts, `ESCALATE`, evaluator outages, and + malformed responses are treated as denials. Unknown slash-commands are + denied locally without evaluation or dispatch. +- **autoPermission is allowlist-only.** It is granted only to commands in the + operator's explicit allowlist (default: the routine read-only commands + `help`/`list`/`status`). Chat turns run with `autoPermission: false` so tool + effects inside the turn still need Kernel/permission approval. Adding + `"chat"` to the allowlist restores Rowboat-style permission-less turns — + a deliberate, risky operator choice. +- **Transport is fail closed.** Telegram DMs only (group chats are ignored — + any member could otherwise drive the bridge), an explicit chat-and-sender-ID + allowlist (empty = deny everyone), an offset confirmed only after the + inbound handler settles (at-least-once delivery after a persistence failure), + and terminal handling for revoked tokens (401/404). + +## Credentials + +The Telegram bot token is read from the `HELM_TELEGRAM_BOT_TOKEN` environment +variable **only** — never from config files, command arguments, or inbound +messages. It is never logged or embedded in message text. + +```bash +export HELM_TELEGRAM_BOT_TOKEN=... # from @BotFather +``` + +## Usage + +```ts +import { + ChannelBridge, + TelegramTransport, + createKernelEvaluator, + telegramOptionsFromEnv, +} from "@mindburn/helm-channel-bridge"; + +const bridge = new ChannelBridge({ + transportName: "telegram", + evaluator: createKernelEvaluator({ + tenantId: process.env.HELM_TENANT_ID!, + apiKey: process.env.HELM_API_KEY!, + }), + sessions: myGovernedSessions, // ChannelSessions implementation + turnEvents: myTurnEventBus, // ChannelTurnEventSource implementation +}); + +const transport = new TelegramTransport( + telegramOptionsFromEnv(process.env, { + allowFrom: ["123456789"], // your Telegram chat ID + stateFile: ".helm/telegram-offset.json", + onInbound: (senderKey, chatId, text) => + bridge.handleInbound(senderKey, text, (msg) => transport.send(chatId, msg)), + }), +); + +await transport.start(); +``` + +`ChannelSessions` / `ChannelTurnEventSource` are minimal interfaces your +governed runtime implements (principal-scoped session listing, create session, +send message, stop turn, respond to ask_human, settle-event stream). The +bridge only ever calls them after a Kernel `ALLOW`. + +## Demo of a non-dispatching path + +See `src/bridge.test.ts`: a `DENY` verdict (with receipt ID) blocks dispatch +and is reported to the sender; `ESCALATE`, unknown verdicts, evaluator +outages, and unknown slash-commands are all denied without touching the turn +engine. + +## Development + +```bash +npm install +npm test +``` + +Tests use a fake transport, fake session engine, fake evaluator, and an +in-memory event bus — no network, no credentials. diff --git a/packages/js/helm-channel-bridge/package-lock.json b/packages/js/helm-channel-bridge/package-lock.json new file mode 100644 index 0000000..4a10f3f --- /dev/null +++ b/packages/js/helm-channel-bridge/package-lock.json @@ -0,0 +1,48 @@ +{ + "name": "@mindburn/helm-channel-bridge", + "version": "0.1.0", + "lockfileVersion": 3, + "requires": true, + "packages": { + "": { + "name": "@mindburn/helm-channel-bridge", + "version": "0.1.0", + "license": "Apache-2.0", + "devDependencies": { + "@types/node": "^22.0.0", + "typescript": "^6.0.3" + } + }, + "node_modules/@types/node": { + "version": "22.20.1", + "resolved": "https://registry.npmjs.org/@types/node/-/node-22.20.1.tgz", + "integrity": "sha512-EANqOCF9QFyra+4pfxUcX9STKJpCLjMbObVzljIJomAWSnuSIEAvyzEU53GaajbXJEgdh0iEcPL+DGvpUd4k1Q==", + "dev": true, + "license": "MIT", + "dependencies": { + "undici-types": "~6.21.0" + } + }, + "node_modules/typescript": { + "version": "6.0.3", + "resolved": "https://registry.npmjs.org/typescript/-/typescript-6.0.3.tgz", + "integrity": "sha512-y2TvuxSZPDyQakkFRPZHKFm+KKVqIisdg9/CZwm9ftvKXLP8NRWj38/ODjNbr43SsoXqNuAisEf1GdCxqWcdBw==", + "dev": true, + "license": "Apache-2.0", + "bin": { + "tsc": "bin/tsc", + "tsserver": "bin/tsserver" + }, + "engines": { + "node": ">=14.17" + } + }, + "node_modules/undici-types": { + "version": "6.21.0", + "resolved": "https://registry.npmjs.org/undici-types/-/undici-types-6.21.0.tgz", + "integrity": "sha512-iwDZqg0QAGrg9Rav5H4n0M64c3mkR59cJ6wQp+7C4nI0gsmExaedaYLNO44eT4AtBBwjbTiGPMlt2Md0T9H9JQ==", + "dev": true, + "license": "MIT" + } + } +} diff --git a/packages/js/helm-channel-bridge/package.json b/packages/js/helm-channel-bridge/package.json new file mode 100644 index 0000000..c934284 --- /dev/null +++ b/packages/js/helm-channel-bridge/package.json @@ -0,0 +1,42 @@ +{ + "name": "@mindburn/helm-channel-bridge", + "version": "0.1.0", + "description": "HELM-governed channel bridge: human-in-the-loop by phone with Kernel-evaluated turns (Telegram transport)", + "type": "module", + "main": "dist/index.js", + "types": "dist/index.d.ts", + "files": [ + "dist/index.js", + "dist/index.d.ts", + "dist/bridge.js", + "dist/bridge.d.ts", + "dist/evaluator.js", + "dist/evaluator.d.ts", + "dist/telegram.js", + "dist/telegram.d.ts", + "README.md" + ], + "scripts": { + "build": "tsc -p tsconfig.json", + "test": "npm run build && node --test \"dist/**/*.test.js\"" + }, + "keywords": [ + "helm", + "ai", + "agents", + "hitl", + "telegram", + "channel", + "receipts" + ], + "license": "Apache-2.0", + "repository": { + "type": "git", + "url": "https://github.com/Mindburn-Labs/helm-agent-integrations", + "directory": "packages/js/helm-channel-bridge" + }, + "devDependencies": { + "@types/node": "^22.0.0", + "typescript": "^6.0.3" + } +} diff --git a/packages/js/helm-channel-bridge/src/bridge.test.ts b/packages/js/helm-channel-bridge/src/bridge.test.ts new file mode 100644 index 0000000..ce1d45c --- /dev/null +++ b/packages/js/helm-channel-bridge/src/bridge.test.ts @@ -0,0 +1,418 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { + ChannelBridge, + type ChannelBridgeConfig, + type ChannelSessionSummary, + type ChannelTurnEvent, +} from "./bridge.js"; +import type { + ChannelDecision, + ChannelEvaluationRequest, +} from "./evaluator.js"; + +class FakeEvaluator { + requests: ChannelEvaluationRequest[] = []; + decision: ChannelDecision = { verdict: "ALLOW", receiptId: "rcpt-fake" }; + decide?: (req: ChannelEvaluationRequest) => ChannelDecision; + + async evaluate(req: ChannelEvaluationRequest): Promise { + this.requests.push(structuredClone(req)); + return this.decide ? this.decide(req) : this.decision; + } +} + +class FakeBus { + private listeners = new Set<(e: { turnId: string; event: ChannelTurnEvent }) => void>(); + + subscribeAll(listener: (e: { turnId: string; event: ChannelTurnEvent }) => void): () => void { + this.listeners.add(listener); + return () => this.listeners.delete(listener); + } + + emit(turnId: string, event: ChannelTurnEvent): void { + for (const listener of [...this.listeners]) { + listener({ turnId, event }); + } + } + + get size(): number { + return this.listeners.size; + } +} + +interface SentMessage { + sessionId: string; + text: string; + options: { autoPermission: boolean; principal: string }; +} + +class FakeSessions { + summaries: ChannelSessionSummary[] = []; + sent: SentMessage[] = []; + stopped: Array<{ turnId: string; reason: string }> = []; + answered: Array<{ turnId: string; toolCallId: string; answer: string }> = []; + listCalls: string[] = []; + onlyPrincipal?: string; + created = 0; + turnCounter = 0; + /** Optional hook fired synchronously inside sendMessage (bus is already subscribed). */ + onSend?: (turnId: string) => void; + /** Optional hook fired synchronously inside respondToAskHuman. */ + onAnswer?: (turnId: string) => void; + + // Deliberately UNscoped: returns every session regardless of principal, to + // prove the bridge itself enforces per-principal scoping even when the + // engine does not. + listSessions(principal: string): ChannelSessionSummary[] { + this.listCalls.push(principal); + return this.onlyPrincipal === undefined || principal === this.onlyPrincipal ? this.summaries : []; + } + + async createSession(): Promise { + this.created += 1; + const sessionId = `sess-${this.created}`; + this.summaries.push({ + sessionId, + title: `Session ${this.created}`, + updatedAt: new Date().toISOString(), + }); + return sessionId; + } + + async sendMessage( + sessionId: string, + text: string, + options: { autoPermission: boolean; principal: string }, + ): Promise<{ turnId: string }> { + this.turnCounter += 1; + const turnId = `turn-${this.turnCounter}`; + this.sent.push({ sessionId, text, options }); + const entry = this.summaries.find((s) => s.sessionId === sessionId); + if (entry) { + entry.latestTurnId = turnId; + entry.latestTurnStatus = "running"; + } + this.onSend?.(turnId); + return { turnId }; + } + + async stopTurn(turnId: string, reason: string): Promise { + this.stopped.push({ turnId, reason }); + const entry = this.summaries.find((s) => s.latestTurnId === turnId); + if (entry) entry.latestTurnStatus = "cancelled"; + } + + async respondToAskHuman(turnId: string, toolCallId: string, answer: string): Promise { + this.answered.push({ turnId, toolCallId, answer }); + this.onAnswer?.(turnId); + } +} + +function harness(overrides: Partial = {}) { + const evaluator = new FakeEvaluator(); + const sessions = new FakeSessions(); + const bus = new FakeBus(); + const replies: string[] = []; + const bridge = new ChannelBridge({ + transportName: "telegram", + evaluator, + sessions, + turnEvents: bus, + ...overrides, + }); + const reply = async (text: string) => { + replies.push(text); + }; + return { bridge, evaluator, sessions, bus, replies, reply }; +} + +test("read-only command is Kernel-evaluated before execution", async () => { + const { bridge, evaluator, sessions, replies, reply } = harness(); + sessions.summaries.push({ + sessionId: "sess-x", + title: "Quarterly audit", + updatedAt: new Date().toISOString(), + }); + + await bridge.handleInbound("telegram:42", "list", reply); + + assert.equal(evaluator.requests.length, 1); + const req = evaluator.requests[0]; + assert.equal(req.actionUrn, "channel.telegram.command.list"); + assert.equal(req.senderKey, "telegram:42"); + assert.equal(req.effectClass, "E0"); + assert.equal(req.metadata?.auto_permission, true); + assert.equal(sessions.sent.length, 0); + assert.ok(replies.some((r) => r.includes("Quarterly audit"))); +}); + +test("resumed sessions retain the sender principal for every lookup", async () => { + const { bridge, sessions, replies, reply } = harness(); + sessions.onlyPrincipal = "telegram:42"; + sessions.summaries.push({ + sessionId: "sess-1", + title: "Private audit", + updatedAt: new Date().toISOString(), + latestTurnId: "turn-private", + latestTurnStatus: "running", + }); + + await bridge.handleInbound("telegram:42", "resume 1", reply); + await bridge.handleInbound("telegram:42", "status", reply); + await bridge.handleInbound("telegram:42", "stop", reply); + + assert.ok(replies.some((r) => r.includes('Resumed "Private audit"'))); + assert.ok(replies.some((r) => r.includes('Current session: "Private audit"'))); + assert.deepEqual(sessions.stopped, [{ turnId: "turn-private", reason: "stopped from governed channel" }]); + assert.ok(sessions.listCalls.every((principal) => principal === "telegram:42")); +}); + +test("a bridge-owned session is hidden from another sender even with an unscoped engine", async () => { + const { bridge, sessions, bus, replies, reply } = harness(); + sessions.onSend = (turnId) => bus.emit(turnId, { type: "turn_completed", text: "done" }); + + await bridge.handleInbound("telegram:42", "start private work", reply); + replies.length = 0; + await bridge.handleInbound("telegram:1337", "list", reply); + await bridge.handleInbound("telegram:1337", "resume 1", reply); + + assert.equal(sessions.sent.length, 1); + assert.ok(replies.some((r) => r.includes("No governed sessions yet"))); + assert.ok(replies.some((r) => r.includes("No session #1"))); +}); + +test("new with a message evaluates the fresh session and embedded turn separately", async () => { + const { bridge, evaluator, sessions, replies, reply } = harness(); + evaluator.decide = (request) => + request.actionUrn === "channel.telegram.turn.run" + ? { verdict: "DENY", reason: "turns disabled" } + : { verdict: "ALLOW" }; + + await bridge.handleInbound("telegram:42", "new delete the audit", reply); + + assert.deepEqual( + evaluator.requests.map((request) => request.actionUrn), + ["channel.telegram.command.new", "channel.telegram.turn.run"], + ); + assert.equal(sessions.created, 0); + assert.equal(sessions.sent.length, 0); + assert.ok(replies.some((r) => r.includes('HELM denied "chat"'))); +}); + +test("a timeout keeps the sender busy without discarding the active session", async () => { + const { bridge, sessions, bus, replies, reply } = harness({ turnTimeoutMs: 1 }); + + await bridge.handleInbound("telegram:42", "first", reply); + await bridge.handleInbound("telegram:42", "new second", reply); + await bridge.handleInbound("telegram:42", "status", reply); + + assert.equal(sessions.sent.length, 1); + assert.ok(replies.some((r) => r.includes('Current session: "Session 1"'))); + + bus.emit("turn-1", { type: "turn_completed", text: "first complete" }); + await new Promise((resolve) => setImmediate(resolve)); + sessions.onSend = (turnId) => bus.emit(turnId, { type: "turn_completed", text: "second complete" }); + await bridge.handleInbound("telegram:42", "second", reply); + + assert.equal(sessions.sent.length, 2); + assert.equal(sessions.sent[1].sessionId, "sess-1"); +}); + +test("chat message becomes a Kernel-evaluated turn and dispatches on ALLOW", async () => { + const { bridge, evaluator, sessions, bus, replies, reply } = harness(); + sessions.onSend = (turnId) => bus.emit(turnId, { type: "turn_completed", text: "All clear." }); + + await bridge.handleInbound("telegram:42", "summarize the audit", reply); + + assert.equal(evaluator.requests.length, 1); + const req = evaluator.requests[0]; + assert.equal(req.actionUrn, "channel.telegram.turn.run"); + assert.equal(req.effectClass, "E3"); + assert.equal(req.metadata?.auto_permission, false); + + assert.equal(sessions.sent.length, 1); + assert.equal(sessions.sent[0].text, "summarize the audit"); + assert.equal(sessions.sent[0].options.autoPermission, false); + assert.equal(sessions.sent[0].options.principal, "telegram:42"); + assert.ok(replies.includes("All clear.")); + // Watcher unsubscribes after the turn settles. + assert.equal(bus.size, 0); +}); + +test("DENY verdict blocks dispatch and reports the receipt", async () => { + const { bridge, evaluator, sessions, replies, reply } = harness(); + evaluator.decision = { + verdict: "DENY", + reason: "channel turns disabled by policy", + reasonCode: "POLICY_DENY", + receiptId: "rcpt-deny-1", + }; + + await bridge.handleInbound("telegram:42", "delete everything", reply); + + assert.equal(evaluator.requests.length, 1); + assert.equal(sessions.sent.length, 0); + assert.equal(sessions.created, 0); + const denial = replies.find((r) => r.startsWith("⛔")); + assert.ok(denial); + assert.ok(denial.includes("channel turns disabled by policy")); + assert.ok(denial.includes("rcpt-deny-1")); +}); + +test("non-ALLOW verdicts fail closed (ESCALATE, unknown verdict, evaluator outage)", async () => { + const { bridge, evaluator, sessions, replies, reply } = harness(); + + evaluator.decision = { verdict: "ESCALATE", reason: "needs operator approval" }; + await bridge.handleInbound("telegram:42", "run the payroll", reply); + assert.equal(sessions.sent.length, 0); + + evaluator.decision = { verdict: "SOMETHING_UNEXPECTED" }; + await bridge.handleInbound("telegram:42", "status please", reply); + assert.equal(sessions.sent.length, 0); + + evaluator.decision = { + verdict: "DENY", + reason: "HELM Kernel evaluation unavailable: connection refused", + reasonCode: "CHANNEL_EVALUATOR_UNAVAILABLE", + }; + await bridge.handleInbound("telegram:42", "list", reply); + assert.equal(sessions.sent.length, 0); + + assert.equal(evaluator.requests.length, 3); + assert.equal(replies.filter((r) => r.startsWith("⛔")).length, 3); +}); + +test("unknown slash-commands are denied without evaluation or dispatch", async () => { + const { bridge, evaluator, sessions, replies, reply } = harness(); + + await bridge.handleInbound("telegram:42", "/rm -rf /", reply); + + assert.equal(evaluator.requests.length, 0); + assert.equal(sessions.sent.length, 0); + const denial = replies.find((r) => r.startsWith("⛔")); + assert.ok(denial); + assert.ok(denial.includes("Unknown command")); +}); + +test("ask_human question is relayed and the answer round-trips through the Kernel", async () => { + const { bridge, evaluator, sessions, bus, replies, reply } = harness(); + sessions.onSend = (turnId) => + bus.emit(turnId, { + type: "turn_suspended", + pendingAskHuman: { + toolCallId: "call-99", + question: "Deploy to production?", + options: ["yes", "no"], + }, + pendingPermissions: 0, + }); + sessions.onAnswer = (turnId) => + bus.emit(turnId, { type: "turn_completed", text: "Deployed and receipted." }); + + await bridge.handleInbound("telegram:42", "ship it", reply); + assert.ok(replies.some((r) => r.includes("❓ Deploy to production?"))); + assert.ok(replies.some((r) => r.includes("1. yes"))); + + replies.length = 0; + await bridge.handleInbound("telegram:42", "yes", reply); + + // The answer was evaluated as its own governed command. + const answerReq = evaluator.requests.find((r) => + r.actionUrn === "channel.telegram.ask_human.answer" + ); + assert.ok(answerReq); + assert.equal(answerReq.input.answer, "yes"); + assert.equal(answerReq.metadata?.auto_permission, false); + + assert.equal(sessions.answered.length, 1); + assert.deepEqual(sessions.answered[0], { + turnId: "turn-1", + toolCallId: "call-99", + answer: "yes", + }); + assert.ok(replies.includes("Deployed and receipted.")); +}); + +test("denied ask_human answer is never routed back into the turn", async () => { + const { bridge, evaluator, sessions, bus, replies, reply } = harness(); + sessions.onSend = (turnId) => + bus.emit(turnId, { + type: "turn_suspended", + pendingAskHuman: { toolCallId: "call-1", question: "Proceed?" }, + }); + + await bridge.handleInbound("telegram:42", "start", reply); + assert.ok(replies.some((r) => r.includes("❓ Proceed?"))); + + evaluator.decide = (req) => + req.actionUrn.endsWith("ask_human.answer") + ? { verdict: "DENY", reason: "answers require MFA step-up", receiptId: "rcpt-mfa" } + : { verdict: "ALLOW" }; + replies.length = 0; + await bridge.handleInbound("telegram:42", "yes", reply); + + assert.equal(sessions.answered.length, 0); + const denial = replies.find((r) => r.startsWith("⛔")); + assert.ok(denial); + assert.ok(denial.includes("rcpt-mfa")); +}); + +test("autoPermission is granted only to explicitly allowlisted commands", async () => { + const { bridge, sessions, bus, reply } = harness(); + sessions.onSend = (turnId) => bus.emit(turnId, { type: "turn_completed", text: "ok" }); + + // Defaults: routine read-only commands are allowlisted, chat is not. + assert.equal(bridge.isAutoPermissionAllowed("help"), true); + assert.equal(bridge.isAutoPermissionAllowed("list"), true); + assert.equal(bridge.isAutoPermissionAllowed("status"), true); + assert.equal(bridge.isAutoPermissionAllowed("chat"), false); + assert.equal(bridge.isAutoPermissionAllowed("stop"), false); + + await bridge.handleInbound("telegram:42", "hello", reply); + assert.equal(sessions.sent[0].options.autoPermission, false); + + // Operator explicitly allowlists chat turns (Rowboat-style); documented risk. + const permissive = harness({ autoPermissionAllowlist: ["help", "list", "status", "chat"] }); + permissive.sessions.onSend = (turnId) => + permissive.bus.emit(turnId, { type: "turn_completed", text: "ok" }); + await permissive.bridge.handleInbound("telegram:42", "hello", permissive.reply); + assert.equal(permissive.sessions.sent[0].options.autoPermission, true); +}); + +test("stop cancels the running turn only after Kernel ALLOW", async () => { + const { bridge, evaluator, sessions, replies, reply } = harness(); + sessions.summaries.push({ + sessionId: "sess-1", + title: "Long task", + updatedAt: new Date().toISOString(), + latestTurnId: "turn-77", + latestTurnStatus: "running", + }); + + await bridge.handleInbound("telegram:42", "resume 1", reply); + await bridge.handleInbound("telegram:42", "stop", reply); + + const stopReq = evaluator.requests.find((r) => r.actionUrn === "channel.telegram.command.stop"); + assert.ok(stopReq); + assert.equal(stopReq.effectClass, "E2"); + assert.deepEqual(sessions.stopped, [{ turnId: "turn-77", reason: "stopped from governed channel" }]); + assert.ok(replies.includes("🛑 Stop requested.")); +}); + +test("denied stop never touches the engine", async () => { + const { bridge, evaluator, sessions, reply } = harness(); + sessions.summaries.push({ + sessionId: "sess-1", + title: "Long task", + updatedAt: new Date().toISOString(), + latestTurnId: "turn-77", + latestTurnStatus: "running", + }); + evaluator.decision = { verdict: "DENY", reason: "no remote stops" }; + + await bridge.handleInbound("telegram:42", "resume 1", reply); + await bridge.handleInbound("telegram:42", "stop", reply); + + assert.equal(sessions.stopped.length, 0); +}); diff --git a/packages/js/helm-channel-bridge/src/bridge.ts b/packages/js/helm-channel-bridge/src/bridge.ts new file mode 100644 index 0000000..1f7c2b6 --- /dev/null +++ b/packages/js/helm-channel-bridge/src/bridge.ts @@ -0,0 +1,704 @@ +// Transport-agnostic, HELM-governed channel bridge. +// +// Structure adapted from the Apache-2.0 Rowboat project's ChannelBridge +// (rowboatlabs/rowboat, apps/x/packages/core/src/channels/bridge.ts): +// per-sender session state, a command layer, a turn-settle watcher, and an +// ask_human relay. This is an original implementation with HELM-governed +// semantics instead of Rowboat's autoPermission-by-default behavior: +// +// - EVERY inbound command is evaluated by the HELM Kernel before it may +// execute. Non-ALLOW verdicts, unknown verdicts, and evaluator failures +// are all fail-closed denials. +// - Unknown slash-commands are denied locally without dispatch. +// - autoPermission is granted only to commands in an explicit operator +// allowlist (default: the routine read-only commands help/list/status). +// Chat turns run with autoPermission=false unless the operator +// deliberately allowlists "chat". + +import type { ChannelDecision, ChannelEvaluator } from "./evaluator.js"; + +export type ReplyFn = (text: string) => Promise; + +export interface ChannelSessionSummary { + sessionId: string; + title?: string; + updatedAt: string; + latestTurnId?: string; + latestTurnStatus?: string; + error?: boolean; +} + +export interface ChannelTurnSendOptions { + /** True only when the command is in the operator autoPermission allowlist. */ + autoPermission: boolean; + /** HELM principal that owns this turn (the channel sender). */ + principal: string; + metadata?: Record; +} + +/** + * Minimal governed-session engine the bridge drives. Implementations are + * expected to route sendMessage/stopTurn/respondToAskHuman through the host + * agent runtime; the bridge only ever calls them after a Kernel ALLOW. + * + * Session listing is principal-scoped: implementations MUST return only the + * sessions the given principal is authorized to see. The bridge additionally + * checks recorded ownership for sessions it created, so another sender cannot + * list, resume, or drive them even if an in-process engine responds + * incorrectly. After a bridge restart, the session engine remains the source + * of truth for pre-existing session visibility. + */ +export interface ChannelSessions { + /** Return only sessions visible to this principal (the channel sender). */ + listSessions(principal: string): ChannelSessionSummary[]; + createSession(): Promise; + sendMessage( + sessionId: string, + text: string, + options: ChannelTurnSendOptions, + ): Promise<{ turnId: string }>; + stopTurn(turnId: string, reason: string): Promise; + respondToAskHuman(turnId: string, toolCallId: string, answer: string): Promise; +} + +export type ChannelTurnEvent = + | { type: "turn_completed"; text: string | null } + | { type: "turn_failed"; error: string } + | { type: "turn_cancelled" } + | { + type: "turn_suspended"; + pendingAskHuman?: { toolCallId: string; question: string; options?: string[] } | null; + pendingPermissions?: number; + }; + +export interface ChannelTurnEventSource { + /** Subscribe to settle-relevant events for every turn. Returns unsubscribe. */ + subscribeAll( + listener: (event: { turnId: string; event: ChannelTurnEvent }) => void, + ): () => void; +} + +export interface ChannelBridgeConfig { + /** Transport name used in action URNs, e.g. "telegram". */ + transportName: string; + evaluator: ChannelEvaluator; + sessions: ChannelSessions; + turnEvents: ChannelTurnEventSource; + /** Defaults to 30 minutes, matching the desktop turn budget. */ + turnTimeoutMs?: number; + /** + * Commands that may run with autoPermission. Default: ["help", "list", + * "status"] — routine read-only commands. Adding "chat" restores + * Rowboat-style permission-less turns and is a deliberate, risky operator + * choice; turns otherwise always run with autoPermission=false so tool + * effects still need Kernel/permission approval. + */ + autoPermissionAllowlist?: string[]; + riskClass?: string; +} + +const DEFAULT_TURN_TIMEOUT_MS = 30 * 60 * 1000; +const DEFAULT_AUTO_PERMISSION_ALLOWLIST = ["help", "list", "status"]; +const LIST_LIMIT = 10; +// Telegram caps messages at 4096 chars; long replies are chunked, then +// truncated — the governed session keeps the full text. +const REPLY_CHUNK_SIZE = 3500; +const MAX_REPLY_CHUNKS = 3; + +const HELP_TEXT = [ + "🤖 HELM channel commands:", + "• help — show this help", + "• list — recent governed sessions", + "• resume N — continue session N from the list", + "• new [message] — start a fresh governed session", + "• status — current session and what it is doing", + "• stop — cancel the running turn", + "", + "Anything else is evaluated by the HELM Kernel and, on ALLOW, sent to your current session.", +].join("\n"); + +/** Per-command effect classification for Kernel evaluation. */ +const COMMAND_EFFECT_CLASS: Record = { + help: "E0", + list: "E0", + status: "E0", + resume: "E1", + new: "E1", + stop: "E2", + chat: "E3", + ask_human_answer: "E3", +}; + +interface SenderState { + activeSessionId: string | null; + // sessionIds as last shown by `list` (1-based indexing for `resume N`). + lastList: string[]; + pendingAsk: { turnId: string; toolCallId: string } | null; + busy: boolean; + // Turn currently occupying the sender. Cleared only by a real settle event + // (or by the turn never starting); a watcher timeout alone NEVER clears it, + // so a still-running turn keeps the sender busy. + activeTurnId: string | null; +} + +type Settled = + | { kind: "completed"; text: string | null } + | { kind: "failed"; error: string } + | { kind: "cancelled" } + | { kind: "ask_human"; toolCallId: string; question: string; options?: string[] } + | { kind: "suspended" } + | { kind: "timeout" }; + +interface ParsedCommand { + name: string; + arg?: string; +} + +function parseCommand(trimmed: string): ParsedCommand { + const slash = /^\/([a-zA-Z]+)(?:\s+([\s\S]+))?$/.exec(trimmed); + if (slash) { + return { name: slash[1].toLowerCase(), arg: slash[2]?.trim() }; + } + const lower = trimmed.toLowerCase(); + if (lower === "help" || lower === "?") return { name: "help" }; + if (lower === "list" || lower === "chats") return { name: "list" }; + if (lower === "status") return { name: "status" }; + if (lower === "stop") return { name: "stop" }; + if (lower === "new") return { name: "new" }; + const newWithText = /^new\s+([\s\S]+)$/i.exec(trimmed); + if (newWithText) return { name: "new", arg: newWithText[1].trim() }; + const resume = /^(?:resume|open)\s+(\d+)$/i.exec(trimmed); + if (resume) return { name: "resume", arg: resume[1] }; + return { name: "chat", arg: trimmed }; +} + +const KNOWN_COMMANDS = new Set(["help", "list", "status", "stop", "new", "resume", "chat"]); + +function settleOf(event: ChannelTurnEvent): Settled | null { + switch (event.type) { + case "turn_completed": + return { kind: "completed", text: event.text }; + case "turn_failed": + return { kind: "failed", error: event.error }; + case "turn_cancelled": + return { kind: "cancelled" }; + case "turn_suspended": { + const ask = event.pendingAskHuman; + if (ask) { + return { + kind: "ask_human", + toolCallId: ask.toolCallId, + question: ask.question || "The agent needs your input.", + options: ask.options, + }; + } + // Suspended without an ask_human means a permission approval is + // waiting somewhere else (e.g. an operator console); report it. + if ((event.pendingPermissions ?? 0) > 0) { + return { kind: "suspended" }; + } + return null; + } + default: + return null; + } +} + +function relativeTime(iso: string, now: number): string { + const then = Date.parse(iso); + if (!Number.isFinite(then)) return ""; + const diffSec = Math.round((now - then) / 1000); + if (diffSec < 60) return "just now"; + const diffMin = Math.round(diffSec / 60); + if (diffMin < 60) return `${diffMin}m ago`; + const diffHr = Math.round(diffMin / 60); + if (diffHr < 24) return `${diffHr}h ago`; + return `${Math.round(diffHr / 24)}d ago`; +} + +function chunkReply(text: string): string[] { + if (text.length <= REPLY_CHUNK_SIZE) return [text]; + const parts: string[] = []; + let rest = text; + while (rest.length > 0 && parts.length < MAX_REPLY_CHUNKS) { + parts.push(rest.slice(0, REPLY_CHUNK_SIZE)); + rest = rest.slice(REPLY_CHUNK_SIZE); + } + if (rest.length > 0) { + parts[parts.length - 1] += "\n… (truncated — open the governed session for the full reply)"; + } + return parts; +} + +interface TurnWatcher { + waitFor(turnId: string, timeoutMs: number): Promise; + dispose(): void; +} + +export class ChannelBridge { + private senders = new Map(); + // Ownership record for sessions created through this bridge. Used to + // enforce per-principal scoping even if the session engine ever answers + // with an unscoped listing (defense in depth — the engine scopes too). + private sessionOwners = new Map(); + private readonly turnTimeoutMs: number; + private readonly autoPermissionAllowlist: Set; + + constructor(private readonly config: ChannelBridgeConfig) { + this.turnTimeoutMs = config.turnTimeoutMs ?? DEFAULT_TURN_TIMEOUT_MS; + this.autoPermissionAllowlist = new Set( + config.autoPermissionAllowlist ?? DEFAULT_AUTO_PERMISSION_ALLOWLIST, + ); + } + + /** True when the command may run with autoPermission (explicit allowlist only). */ + isAutoPermissionAllowed(commandName: string): boolean { + return this.autoPermissionAllowlist.has(commandName); + } + + async handleInbound(senderKey: string, text: string, reply: ReplyFn): Promise { + const trimmed = text.trim(); + if (!trimmed) return; + const state = this.senderState(senderKey); + const command = parseCommand(trimmed); + + try { + // Fail closed: unknown slash-commands are denied without evaluation or + // dispatch. Bare text always parses to a known command ("chat"). + if (!KNOWN_COMMANDS.has(command.name)) { + await reply( + `⛔ Unknown command "/${command.name}" — denied by default. Send "help" for the command list.`, + ); + return; + } + + if (command.name === "chat" && state.pendingAsk) { + await this.answerPendingAsk(state, senderKey, command.arg ?? trimmed, reply); + return; + } + + const decision = await this.evaluate(senderKey, state, command); + if (decision.verdict !== "ALLOW") { + await reply(denialText(command.name, decision)); + return; + } + + switch (command.name) { + case "help": + await reply(HELP_TEXT); + return; + case "list": + await reply(this.renderList(state, senderKey)); + return; + case "resume": + await reply(this.resumeSession(state, senderKey, Number(command.arg))); + return; + case "status": + await reply(this.renderStatus(state, senderKey)); + return; + case "stop": + await reply(await this.stopActive(state, senderKey)); + return; + case "new": { + // A fresh-session request must never discard the active session + // while its turn is still running. The Kernel has authorized this + // command, but no local state change is safe until the sender is + // free to start the new turn. + if (state.busy) { + await reply('⏳ Still working on the previous message — send "stop" to cancel it.'); + return; + } + state.activeSessionId = null; + state.pendingAsk = null; + if (!command.arg) { + await reply("🆕 Fresh governed session — send your first message."); + return; + } + // "new " embeds a turn: command.new (E1, evaluated above) + // only authorizes starting a fresh session. The message itself MUST + // pass the full turn evaluation chain (E3 turn.run) before dispatch, + // exactly like a bare chat message — otherwise policies that deny + // channel turns would be bypassed by prefixing "new". + const turnDecision = await this.evaluate(senderKey, state, { name: "chat", arg: command.arg }); + if (turnDecision.verdict !== "ALLOW") { + await reply(denialText("chat", turnDecision)); + return; + } + await this.runChatTurn(state, senderKey, command.arg, reply); + return; + } + case "chat": + await this.runChatTurn(state, senderKey, command.arg ?? trimmed, reply); + return; + } + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + await reply(`❌ ${message}`).catch(() => undefined); + } + } + + private senderState(senderKey: string): SenderState { + let state = this.senders.get(senderKey); + if (!state) { + state = { activeSessionId: null, lastList: [], pendingAsk: null, busy: false, activeTurnId: null }; + this.senders.set(senderKey, state); + } + return state; + } + + private evaluate( + senderKey: string, + state: SenderState, + command: ParsedCommand, + ): Promise { + const autoPermission = this.isAutoPermissionAllowed(command.name); + const actionUrn = command.name === "chat" + ? `channel.${this.config.transportName}.turn.run` + : `channel.${this.config.transportName}.command.${command.name}`; + return this.config.evaluator.evaluate({ + actionUrn, + senderKey, + sessionId: state.activeSessionId ?? `channel:${senderKey}`, + input: { command: command.name, arg: command.arg }, + riskClass: this.config.riskClass ?? "T2", + effectClass: COMMAND_EFFECT_CLASS[command.name] ?? "E3", + metadata: { + framework: "helm-channel-bridge", + transport: this.config.transportName, + command: command.name, + auto_permission: autoPermission, + }, + }); + } + + private sessionEntry(senderKey: string, sessionId: string): ChannelSessionSummary | undefined { + const owner = this.sessionOwners.get(sessionId); + if (owner !== undefined && owner !== senderKey) return undefined; + return this.config.sessions + .listSessions(senderKey) + .find((e) => e.sessionId === sessionId); + } + + /** + * Sessions visible to one sender. The engine is asked for the + * principal-scoped list; anything this bridge recorded as owned by a + * different principal is dropped regardless, so cross-principal session + * discovery or control is denied at the bridge boundary too. + */ + private visibleSessions(senderKey: string): ChannelSessionSummary[] { + return this.config.sessions + .listSessions(senderKey) + .filter((e) => { + const owner = this.sessionOwners.get(e.sessionId); + return owner === undefined || owner === senderKey; + }); + } + + private recentSessions(senderKey: string): ChannelSessionSummary[] { + return this.visibleSessions(senderKey) + .filter((e) => !e.error) + .sort((a, b) => b.updatedAt.localeCompare(a.updatedAt)) + .slice(0, LIST_LIMIT); + } + + private renderList(state: SenderState, senderKey: string): string { + const entries = this.recentSessions(senderKey); + if (entries.length === 0) { + return "No governed sessions yet — just send a message to start one."; + } + state.lastList = entries.map((e) => e.sessionId); + const now = Date.now(); + const lines = entries.map((e, i) => { + const marker = e.latestTurnStatus === "suspended" + ? " ⚠️" + : e.latestTurnStatus === "idle" + ? " ⏳" + : ""; + const active = e.sessionId === state.activeSessionId ? " ← current" : ""; + return `${i + 1}. ${e.title ?? "Untitled"}${marker} (${relativeTime(e.updatedAt, now)})${active}`; + }); + return ["Recent governed sessions:", ...lines, "", `Reply "resume N" to continue one.`].join("\n"); + } + + private resumeSession(state: SenderState, senderKey: string, index: number): string { + if (state.lastList.length === 0) { + state.lastList = this.recentSessions(senderKey).map((e) => e.sessionId); + } + const sessionId = state.lastList[index - 1]; + if (!sessionId) { + return `No session #${index} — send "list" to see recent governed sessions.`; + } + // Defense in depth: never resume a session recorded under another + // principal, even if it somehow surfaced in this sender's list. + const owner = this.sessionOwners.get(sessionId); + if (owner !== undefined && owner !== senderKey) { + return "⛔ That session belongs to a different sender — access denied."; + } + const entry = this.sessionEntry(senderKey, sessionId); + if (!entry) { + state.lastList = []; + return "⛔ That session is no longer available — send \"list\" to see your governed sessions."; + } + state.activeSessionId = sessionId; + state.pendingAsk = null; + return `▶️ Resumed "${entry?.title ?? "Untitled"}" — send a message to continue.`; + } + + private renderStatus(state: SenderState, senderKey: string): string { + if (!state.activeSessionId) { + return "No current session — your next message starts a new governed one."; + } + const entry = this.sessionEntry(senderKey, state.activeSessionId); + if (!entry) return "The current session no longer exists — send a message to start fresh."; + const status = state.busy + ? "working" + : entry.latestTurnStatus === "suspended" + ? "waiting on input" + : entry.latestTurnStatus ?? "unknown"; + return `Current session: "${entry.title ?? "Untitled"}" — ${status}.`; + } + + private async stopActive(state: SenderState, senderKey: string): Promise { + state.pendingAsk = null; + if (!state.activeSessionId) return "Nothing to stop."; + const entry = this.sessionEntry(senderKey, state.activeSessionId); + if (!entry?.latestTurnId) return "Nothing to stop."; + if ( + entry.latestTurnStatus === "completed" + || entry.latestTurnStatus === "failed" + || entry.latestTurnStatus === "cancelled" + ) { + return "Nothing running in the current session."; + } + await this.config.sessions.stopTurn(entry.latestTurnId, "stopped from governed channel"); + return "🛑 Stop requested."; + } + + private async runChatTurn( + state: SenderState, + senderKey: string, + text: string, + reply: ReplyFn, + ): Promise { + if (state.busy) { + await reply('⏳ Still working on the previous message — send "stop" to cancel it.'); + return; + } + state.busy = true; + const watcher = this.watchBus(); + try { + await reply("⏳ Working on it…"); + if (!state.activeSessionId) { + state.activeSessionId = await this.config.sessions.createSession(); + // Record ownership so no other sender can list/resume/drive it. + this.sessionOwners.set(state.activeSessionId, senderKey); + } + const sent = await this.config.sessions.sendMessage(state.activeSessionId, text, { + autoPermission: this.isAutoPermissionAllowed("chat"), + principal: senderKey, + metadata: { + framework: "helm-channel-bridge", + transport: this.config.transportName, + }, + }); + state.activeTurnId = sent.turnId; + const settled = await watcher.waitFor(sent.turnId, this.turnTimeoutMs); + if (settled.kind === "timeout") { + // The turn is still running. Keep the sender busy and reconcile the + // busy flag from the actual settle event — a timeout alone must never + // free the sender, or later messages would start concurrent turns in + // the same session. + this.reconcileBusyOnSettle(state, sent.turnId, reply); + } else { + state.activeTurnId = null; + state.busy = false; + } + await this.deliverSettled(state, sent.turnId, settled, reply); + } finally { + watcher.dispose(); + if (state.activeTurnId === null) { + state.busy = false; + } + } + } + + private async answerPendingAsk( + state: SenderState, + senderKey: string, + text: string, + reply: ReplyFn, + ): Promise { + const ask = state.pendingAsk; + if (!ask) return; + // The answer itself is an inbound command: Kernel-evaluated before it is + // routed back into the suspended turn. + const decision = await this.config.evaluator.evaluate({ + actionUrn: `channel.${this.config.transportName}.ask_human.answer`, + senderKey, + sessionId: state.activeSessionId ?? `channel:${senderKey}`, + input: { command: "ask_human_answer", turnId: ask.turnId, toolCallId: ask.toolCallId, answer: text }, + riskClass: this.config.riskClass ?? "T2", + effectClass: COMMAND_EFFECT_CLASS.ask_human_answer, + metadata: { + framework: "helm-channel-bridge", + transport: this.config.transportName, + command: "ask_human_answer", + auto_permission: false, + }, + }); + if (decision.verdict !== "ALLOW") { + await reply(denialText("ask_human.answer", decision)); + return; + } + if (state.busy) { + // Keep pendingAsk: the answer was NOT accepted for routing, so the + // sender must be able to retry once the current turn frees up. + await reply('⏳ Still working on the previous message — send "stop" to cancel it.'); + return; + } + state.busy = true; + const watcher = this.watchBus(); + try { + const settledPromise = watcher.waitFor(ask.turnId, this.turnTimeoutMs); + // The answer is accepted for routing only now — clear pendingAsk at the + // point of acceptance, not before the busy check or evaluation. + state.pendingAsk = null; + state.activeTurnId = ask.turnId; + // respondToAskHuman may resolve only when the resumed turn settles, so + // race it against the watcher rather than awaiting it first; a stale + // ask (already answered elsewhere) rejects and is re-routed as chat. + const settled = await Promise.race([ + settledPromise, + this.config.sessions + .respondToAskHuman(ask.turnId, ask.toolCallId, text) + .then(() => settledPromise), + ]); + if (settled.kind === "timeout") { + // Resumed turn still running: stay busy until its real settle event. + this.reconcileBusyOnSettle(state, ask.turnId, reply); + } else { + state.activeTurnId = null; + state.busy = false; + } + await this.deliverSettled(state, ask.turnId, settled, reply); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + await reply(`❌ Could not deliver your answer: ${message}`); + } finally { + watcher.dispose(); + if (state.activeTurnId === null) { + state.busy = false; + } + } + } + + /** + * After a watcher timeout the turn is still running. Subscribe for its real + * settle event and only then free the sender (and deliver the outcome), so + * the busy flag always reflects an actual turn completion/failure/cancel + * rather than an arbitrary clock. + */ + private reconcileBusyOnSettle(state: SenderState, turnId: string, reply: ReplyFn): void { + const unsubscribe = this.config.turnEvents.subscribeAll((event) => { + if (event.turnId !== turnId) return; + const settled = settleOf(event.event); + if (!settled) return; + unsubscribe(); + if (state.activeTurnId === turnId) { + state.activeTurnId = null; + state.busy = false; + } + void this.deliverSettled(state, turnId, settled, reply).catch(() => undefined); + }); + } + + private async deliverSettled( + state: SenderState, + turnId: string, + settled: Settled, + reply: ReplyFn, + ): Promise { + switch (settled.kind) { + case "completed": + for (const chunk of chunkReply(settled.text ?? "✅ Done (no text reply).")) { + await reply(chunk); + } + return; + case "failed": + await reply(`❌ Turn failed: ${settled.error}`); + return; + case "cancelled": + await reply("🛑 Stopped."); + return; + case "ask_human": { + state.pendingAsk = { turnId, toolCallId: settled.toolCallId }; + const lines = [`❓ ${settled.question}`]; + if (settled.options?.length) { + lines.push(...settled.options.map((o, i) => `${i + 1}. ${o}`)); + } + lines.push("", "Reply with your answer — it will be Kernel-evaluated before delivery."); + await reply(lines.join("\n")); + return; + } + case "suspended": + await reply( + "⚠️ The agent is waiting for a permission approval — continue from your governed console.", + ); + return; + case "timeout": + await reply( + "⏱️ Still running — this chat stays busy until the turn actually finishes; check the governed console for progress.", + ); + return; + } + } + + // Buffers settle-relevant events from the moment of subscription so a + // settle firing between sendMessage and waitFor() is never lost. One + // watcher per in-flight message; mechanism adapted from Rowboat's + // ChannelBridge (Apache-2.0), original implementation. + private watchBus(): TurnWatcher { + const buffered: Array<{ turnId: string; settled: Settled }> = []; + let waiter: { turnId: string; resolve: (settled: Settled) => void } | null = null; + let cancelTimer: (() => void) | null = null; + const unsubscribe = this.config.turnEvents.subscribeAll((event) => { + const settled = settleOf(event.event); + if (!settled) return; + if (waiter) { + if (event.turnId === waiter.turnId) waiter.resolve(settled); + return; + } + buffered.push({ turnId: event.turnId, settled }); + }); + return { + waitFor: (turnId: string, timeoutMs: number): Promise => + new Promise((resolve) => { + const hit = buffered.find((b) => b.turnId === turnId); + if (hit) { + resolve(hit.settled); + return; + } + buffered.length = 0; + const timer = setTimeout(() => resolve({ kind: "timeout" }), timeoutMs); + cancelTimer = () => clearTimeout(timer); + waiter = { + turnId, + resolve: (settled) => { + clearTimeout(timer); + resolve(settled); + }, + }; + }), + dispose: () => { + unsubscribe(); + cancelTimer?.(); + }, + }; + } +} + +function denialText(commandName: string, decision: ChannelDecision): string { + const reason = decision.reason ?? decision.reasonCode ?? "policy denial"; + const receipt = decision.receiptId ? ` (receipt ${decision.receiptId})` : ""; + return `⛔ HELM denied "${commandName}": ${reason}${receipt}`; +} diff --git a/packages/js/helm-channel-bridge/src/evaluator.ts b/packages/js/helm-channel-bridge/src/evaluator.ts new file mode 100644 index 0000000..bfba5e2 --- /dev/null +++ b/packages/js/helm-channel-bridge/src/evaluator.ts @@ -0,0 +1,175 @@ +// Kernel evaluator for channel commands. +// +// Every inbound channel command is evaluated by the HELM AI Kernel before it +// may execute. This module deliberately mirrors the tenant-scoped +// /api/v1/evaluate contract used by @mindburn/helm-tool-wrapper, but stays +// self-contained so this package has no cross-package build dependency. +// +// Fail-closed rule: any transport failure, malformed response, or non-ALLOW +// verdict is treated as a denial. The bridge never dispatches without an +// explicit ALLOW. + +export type ChannelVerdict = "ALLOW" | "DENY" | "ESCALATE" | "PENDING" | string; + +export interface ChannelDecision { + verdict: ChannelVerdict; + reason?: string; + reasonCode?: string; + receiptId?: string; + decisionId?: string; +} + +export interface ChannelEvaluationRequest { + /** Action URN, e.g. "channel.telegram.turn.run". */ + actionUrn: string; + /** Stable channel sender identity, used as the HELM principal. */ + senderKey: string; + /** Session the evaluation belongs to (active session or channel identity). */ + sessionId: string; + /** Command payload placed under context.args. */ + input: Record; + riskClass?: string; + effectClass?: string; + metadata?: Record; +} + +export interface ChannelEvaluator { + evaluate(request: ChannelEvaluationRequest): Promise; +} + +export type FetchLike = ( + input: string, + init?: { + method?: string; + headers?: Record; + body?: string; + signal?: AbortSignal; + }, +) => Promise<{ + ok: boolean; + status: number; + statusText?: string; + headers: { get(name: string): string | null }; + json(): Promise; + text(): Promise; +}>; + +export interface KernelEvaluatorConfig { + tenantId: string; + apiKey: string; + helmUrl?: string; + timeoutMs?: number; + fetch?: FetchLike; +} + +const DEFAULT_HELM_URL = "http://127.0.0.1:7714"; +const DEFAULT_TIMEOUT_MS = 30_000; + +/** Synthetic decision used when the Kernel cannot be reached. Fail closed. */ +export function evaluatorUnavailableDecision(detail: string): ChannelDecision { + return { + verdict: "DENY", + reason: `HELM Kernel evaluation unavailable: ${detail}`, + reasonCode: "CHANNEL_EVALUATOR_UNAVAILABLE", + }; +} + +function readRecord(value: unknown): Record { + return typeof value === "object" && value !== null ? value as Record : {}; +} + +function readString(value: unknown): string | undefined { + return typeof value === "string" && value.trim() !== "" ? value : undefined; +} + +function extractDecision(payload: unknown, headers: { get(name: string): string | null }): ChannelDecision { + const body = readRecord(payload); + const nested = readRecord(body.decision ?? body.record ?? body.result ?? payload); + const rawVerdict = nested.verdict ?? nested.status ?? body.verdict ?? body.status; + // Unknown or missing verdicts fail closed to DENY. + const verdict = typeof rawVerdict === "string" ? rawVerdict.toUpperCase() : "DENY"; + return { + verdict, + reason: readString(nested.reason) ?? readString(body.reason), + reasonCode: readString(nested.reason_code) ?? readString(body.reason_code) + ?? headers.get("x-helm-reason-code") ?? undefined, + receiptId: readString(nested.receipt_id) ?? readString(body.receipt_id) + ?? headers.get("x-helm-receipt-id") ?? undefined, + decisionId: readString(nested.decision_id) ?? readString(body.decision_id) + ?? readString(nested.id) ?? headers.get("x-helm-decision-id") ?? undefined, + }; +} + +/** + * Create a ChannelEvaluator backed by a live HELM AI Kernel. + * + * The evaluator never throws for governance outcomes: HTTP errors and network + * failures are converted into fail-closed DENY decisions so the bridge can + * report them to the sender without dispatching. + */ +export function createKernelEvaluator(config: KernelEvaluatorConfig): ChannelEvaluator { + const baseUrl = (config.helmUrl ?? DEFAULT_HELM_URL).replace(/\/$/, ""); + const tenantId = config.tenantId.trim(); + const apiKey = config.apiKey.trim(); + if (tenantId === "") { + throw new Error("HELM tenantId is required for the channel evaluator"); + } + if (apiKey === "") { + throw new Error("HELM apiKey is required for the channel evaluator"); + } + + return { + async evaluate(request: ChannelEvaluationRequest): Promise { + const fetchImpl = config.fetch ?? globalThis.fetch as FetchLike | undefined; + if (!fetchImpl) { + return evaluatorUnavailableDecision("no fetch implementation is available"); + } + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), config.timeoutMs ?? DEFAULT_TIMEOUT_MS); + const payload = { + principal: request.senderKey, + action: "EXECUTE_TOOL", + resource: request.actionUrn, + context: { + tool: request.actionUrn, + args: request.input, + arguments: request.input, + agent_id: request.senderKey, + session_id: request.sessionId, + action_urn: request.actionUrn, + risk_class: request.riskClass ?? "T2", + effect_class: request.effectClass ?? "E3", + metadata: request.metadata ?? {}, + }, + }; + try { + const response = await fetchImpl(`${baseUrl}/api/v1/evaluate`, { + method: "POST", + headers: { + "Authorization": `Bearer ${apiKey}`, + "Content-Type": "application/json", + "X-Helm-Tenant-ID": tenantId, + "X-Helm-Principal-ID": request.senderKey, + }, + body: JSON.stringify(payload), + signal: controller.signal, + }); + let body: unknown; + try { + body = await response.json(); + } catch { + body = undefined; + } + if (!response.ok) { + return evaluatorUnavailableDecision(`HTTP ${response.status} from /api/v1/evaluate`); + } + return extractDecision(body, response.headers); + } catch (error) { + const detail = error instanceof Error ? error.message : String(error); + return evaluatorUnavailableDecision(detail); + } finally { + clearTimeout(timer); + } + }, + }; +} diff --git a/packages/js/helm-channel-bridge/src/index.ts b/packages/js/helm-channel-bridge/src/index.ts new file mode 100644 index 0000000..32c5441 --- /dev/null +++ b/packages/js/helm-channel-bridge/src/index.ts @@ -0,0 +1,31 @@ +export { + createKernelEvaluator, + evaluatorUnavailableDecision, + type ChannelDecision, + type ChannelEvaluationRequest, + type ChannelEvaluator, + type ChannelVerdict, + type FetchLike, + type KernelEvaluatorConfig, +} from "./evaluator.js"; + +export { + ChannelBridge, + type ChannelBridgeConfig, + type ChannelSessionSummary, + type ChannelSessions, + type ChannelTurnEvent, + type ChannelTurnEventSource, + type ChannelTurnSendOptions, + type ReplyFn, +} from "./bridge.js"; + +export { + TelegramApiError, + TelegramTransport, + TELEGRAM_BOT_TOKEN_ENV, + telegramOptionsFromEnv, + type TelegramTransportOptions, + type TelegramTransportStatus, + type TelegramUpdate, +} from "./telegram.js"; diff --git a/packages/js/helm-channel-bridge/src/telegram.test.ts b/packages/js/helm-channel-bridge/src/telegram.test.ts new file mode 100644 index 0000000..e329787 --- /dev/null +++ b/packages/js/helm-channel-bridge/src/telegram.test.ts @@ -0,0 +1,393 @@ +import assert from "node:assert/strict"; +import { mkdir, mkdtemp, readFile, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import test from "node:test"; +import type { FetchLike } from "./evaluator.js"; +import { + TELEGRAM_BOT_TOKEN_ENV, + TelegramApiError, + TelegramTransport, + telegramOptionsFromEnv, + type TelegramTransportStatus, + type TelegramUpdate, +} from "./telegram.js"; + +interface RecordedCall { + url: string; + body?: unknown; +} + +function fakeFetch(handler: (method: string, body?: unknown) => unknown) { + const calls: RecordedCall[] = []; + const fetchImpl: FetchLike = async (input, init) => { + const url = input; + const method = url.slice(url.lastIndexOf("/") + 1); + const body = init?.body ? JSON.parse(init.body) : undefined; + calls.push({ url, body }); + const result = handler(method, body); + const isError = result instanceof TelegramApiError; + return { + ok: true, + status: 200, + headers: { get: () => null }, + json: async () => + isError + ? { ok: false, description: result.message, error_code: result.code } + : { ok: true, result }, + text: async () => "", + }; + }; + return { calls, fetchImpl }; +} + +function dmUpdate(chatId: number, text: string, updateId = 1): TelegramUpdate { + return { + update_id: updateId, + message: { + message_id: updateId, + text, + chat: { id: chatId, type: "private" }, + from: { id: chatId, is_bot: false }, + }, + }; +} + +test("constructor requires a bot token (env-var only)", () => { + assert.throws( + () => + new TelegramTransport({ + botToken: " ", + allowFrom: [], + stateFile: "/tmp/x.json", + onInbound: () => undefined, + }), + /HELM_TELEGRAM_BOT_TOKEN/, + ); +}); + +test("telegramOptionsFromEnv reads the token from the environment only", () => { + assert.throws( + () => + telegramOptionsFromEnv({}, { + allowFrom: [], + stateFile: "/tmp/x.json", + onInbound: () => undefined, + }), + /HELM_TELEGRAM_BOT_TOKEN/, + ); + const opts = telegramOptionsFromEnv( + { [TELEGRAM_BOT_TOKEN_ENV]: "test-token-from-env" }, + { allowFrom: ["42"], stateFile: "/tmp/x.json", onInbound: () => undefined }, + ); + assert.equal(opts.botToken, "test-token-from-env"); +}); + +test("allowlisted DM is routed to the bridge as telegram:", async () => { + const inbound: Array<{ senderKey: string; chatId: string; text: string }> = []; + const transport = new TelegramTransport({ + botToken: "t", + allowFrom: ["42"], + stateFile: "/tmp/x.json", + onInbound: (senderKey, chatId, text) => { + inbound.push({ senderKey, chatId, text }); + }, + }); + + await transport.processUpdate(dmUpdate(42, "list", 7)); + + assert.deepEqual(inbound, [{ senderKey: "telegram:42", chatId: "42", text: "list" }]); +}); + +test("group chats, bots, and non-allowlisted chats are rejected fail-closed", async () => { + const inbound: unknown[] = []; + const { calls, fetchImpl } = fakeFetch(() => ({})); + const transport = new TelegramTransport({ + botToken: "t", + allowFrom: ["42"], + stateFile: "/tmp/x.json", + onInbound: () => { + inbound.push(1); + }, + fetch: fetchImpl, + }); + + // Group chat: silently ignored (any member could drive the bridge). + await transport.processUpdate({ + update_id: 1, + message: { + message_id: 1, + text: "list", + chat: { id: -100, type: "group" }, + from: { id: 42, is_bot: false }, + }, + }); + // Bot-authored message: ignored. + await transport.processUpdate({ + update_id: 2, + message: { + message_id: 2, + text: "list", + chat: { id: 42, type: "private" }, + from: { id: 999, is_bot: true }, + }, + }); + // Non-allowlisted DM: silently dropped — no pairing hint (spam/discovery + // vector), never routed inbound. + await transport.processUpdate(dmUpdate(1337, "list", 3)); + await new Promise((resolve) => setImmediate(resolve)); + + assert.equal(inbound.length, 0); + assert.equal(calls.filter((c) => c.url.endsWith("/sendMessage")).length, 0); +}); + +test("an allowlisted chat is still dropped when the sender user ID is not allowlisted", async () => { + const inbound: unknown[] = []; + const { calls, fetchImpl } = fakeFetch(() => ({})); + const transport = new TelegramTransport({ + botToken: "t", + allowFrom: ["42"], + stateFile: "/tmp/x.json", + onInbound: () => { + inbound.push(1); + }, + fetch: fetchImpl, + }); + + // chat.id is allowlisted but from.id is not: drop silently. + await transport.processUpdate({ + update_id: 1, + message: { + message_id: 1, + text: "list", + chat: { id: 42, type: "private" }, + from: { id: 1337, is_bot: false }, + }, + }); + // Missing sender identity: drop. + await transport.processUpdate({ + update_id: 2, + message: { + message_id: 2, + text: "list", + chat: { id: 42, type: "private" }, + }, + }); + + assert.equal(inbound.length, 0); + assert.equal(calls.filter((c) => c.url.endsWith("/sendMessage")).length, 0); +}); + +test("pollOnce advances and persists the update offset", async () => { + const dir = await mkdtemp(path.join(tmpdir(), "helm-channel-bridge-")); + try { + const stateFile = path.join(dir, "telegram-offset.json"); + const inbound: string[] = []; + const { calls, fetchImpl } = fakeFetch((method) => + method === "getUpdates" ? [dmUpdate(42, "hello", 101), dmUpdate(42, "status", 102)] : {}); + const transport = new TelegramTransport({ + botToken: "t", + allowFrom: ["42"], + stateFile, + onInbound: (_senderKey, _chatId, text) => { + inbound.push(text); + }, + fetch: fetchImpl, + }); + + await transport.pollOnce(); + + assert.deepEqual(inbound, ["hello", "status"]); + const persisted = JSON.parse(await readFile(stateFile, "utf8")) as { offset: number }; + assert.equal(persisted.offset, 103); + + await transport.pollOnce(); + const polls = calls.filter((c) => c.url.endsWith("/getUpdates")); + assert.equal((polls[1].body as { offset: number }).offset, 103); + } finally { + await rm(dir, { recursive: true, force: true }); + } +}); + +test("pollOnce resumes from a persisted offset after restart", async () => { + const dir = await mkdtemp(path.join(tmpdir(), "helm-channel-bridge-")); + try { + const stateFile = path.join(dir, "telegram-offset.json"); + // Prior run: process update 55, which persists offset 56. + const seeding = new TelegramTransport({ + botToken: "t", + allowFrom: ["42"], + stateFile, + onInbound: () => undefined, + fetch: fakeFetch((method) => (method === "getUpdates" ? [dmUpdate(42, "hi", 55)] : {})).fetchImpl, + }); + await seeding.pollOnce(); + + // Restarted transport: reloads the persisted offset before polling, so + // the batch confirmed by the previous run is never redelivered. + const { calls, fetchImpl } = fakeFetch(() => []); + const resumed = new TelegramTransport({ + botToken: "t", + allowFrom: ["42"], + stateFile, + onInbound: () => undefined, + fetch: fetchImpl, + }); + await resumed.restoreOffset(); + await resumed.pollOnce(); + + const poll = calls.find((c) => c.url.endsWith("/getUpdates")); + assert.ok(poll); + assert.equal((poll.body as { offset: number }).offset, 56); + } finally { + await rm(dir, { recursive: true, force: true }); + } +}); + +test("offset is persisted only after inbound handling completes", async () => { + const dir = await mkdtemp(path.join(tmpdir(), "helm-channel-bridge-")); + try { + const stateFile = path.join(dir, "telegram-offset.json"); + let release: () => void = () => undefined; + const handled = new Promise((resolve) => { + release = resolve; + }); + const { fetchImpl } = fakeFetch((method) => + method === "getUpdates" ? [dmUpdate(42, "hello", 101)] : {}); + const transport = new TelegramTransport({ + botToken: "t", + allowFrom: ["42"], + stateFile, + onInbound: () => handled, + fetch: fetchImpl, + }); + + const poll = transport.pollOnce(); + // Let the getUpdates round reach the (still pending) handler. + await new Promise((resolve) => setImmediate(resolve)); + await new Promise((resolve) => setImmediate(resolve)); + + // Handling has not completed: nothing may be persisted yet. + await assert.rejects(readFile(stateFile, "utf8"), /ENOENT/); + + release(); + await poll; + const persisted = JSON.parse(await readFile(stateFile, "utf8")) as { offset: number }; + assert.equal(persisted.offset, 102); + } finally { + await rm(dir, { recursive: true, force: true }); + } +}); + +test("handler failure keeps the update unconfirmed (no offset advance, nothing persisted)", async () => { + const dir = await mkdtemp(path.join(tmpdir(), "helm-channel-bridge-")); + try { + const stateFile = path.join(dir, "telegram-offset.json"); + const { calls, fetchImpl } = fakeFetch((method) => + method === "getUpdates" ? [dmUpdate(42, "boom", 201)] : {}); + const transport = new TelegramTransport({ + botToken: "t", + allowFrom: ["42"], + stateFile, + onInbound: () => Promise.reject(new Error("bridge exploded")), + fetch: fetchImpl, + }); + + await assert.rejects(transport.pollOnce(), /bridge exploded/); + await assert.rejects(readFile(stateFile, "utf8"), /ENOENT/); + + // The next poll must re-fetch from the un-advanced offset. + await assert.rejects(transport.pollOnce(), /bridge exploded/); + const polls = calls.filter((c) => c.url.endsWith("/getUpdates")); + assert.equal(polls.length, 2); + assert.equal((polls[0].body as { offset: number }).offset, 0); + assert.equal((polls[1].body as { offset: number }).offset, 0); + } finally { + await rm(dir, { recursive: true, force: true }); + } +}); + +test("offset persistence failure fails closed: offset not advanced, error surfaced", async () => { + const dir = await mkdtemp(path.join(tmpdir(), "helm-channel-bridge-")); + try { + // A directory as the state file makes writeFile fail (EISDIR). + const stateFile = path.join(dir, "offset-as-dir"); + await mkdir(stateFile); + const handled: string[] = []; + const { calls, fetchImpl } = fakeFetch((method) => + method === "getUpdates" ? [dmUpdate(42, "hello", 301)] : {}); + const transport = new TelegramTransport({ + botToken: "t", + allowFrom: ["42"], + stateFile, + onInbound: (_senderKey, _chatId, text) => { + handled.push(text); + }, + fetch: fetchImpl, + }); + + await assert.rejects( + transport.pollOnce(), + (error: unknown) => { + assert.ok(error instanceof TelegramApiError); + assert.ok(error.message.includes("Failed to persist the Telegram poll offset")); + return true; + }, + ); + // The update WAS handled (side effect ran), but it was NOT confirmed. + assert.deepEqual(handled, ["hello"]); + + // Next poll re-fetches from the un-advanced offset. Consumers must make + // side effects idempotent across that intentional redelivery. + await assert.rejects(transport.pollOnce(), /Failed to persist/); + const polls = calls.filter((c) => c.url.endsWith("/getUpdates")); + assert.equal(polls.length, 2); + assert.equal((polls[0].body as { offset: number }).offset, 0); + assert.equal((polls[1].body as { offset: number }).offset, 0); + } finally { + await rm(dir, { recursive: true, force: true }); + } +}); + +test("401/404 from the Bot API are terminal and surface an error status", async () => { + const statuses: TelegramTransportStatus[] = []; + const { fetchImpl } = fakeFetch(() => new TelegramApiError("Unauthorized", 401)); + const transport = new TelegramTransport({ + botToken: "revoked-token", + allowFrom: ["42"], + stateFile: "/tmp/x.json", + onInbound: () => undefined, + onStatus: (s) => statuses.push(s), + fetch: fetchImpl, + sleep: async () => undefined, + }); + + await transport.start(); + await new Promise((resolve) => setTimeout(resolve, 20)); + + const error = statuses.find((s) => s.state === "error"); + assert.ok(error); + if (error.state === "error") { + assert.ok(error.error.includes("Bot token rejected")); + } +}); + +test("send posts to the Bot API without leaking the token into message bodies", async () => { + const { calls, fetchImpl } = fakeFetch(() => ({})); + const transport = new TelegramTransport({ + botToken: "secret-token", + allowFrom: ["42"], + stateFile: "/tmp/x.json", + onInbound: () => undefined, + fetch: fetchImpl, + }); + + await transport.send("42", "⛔ HELM denied"); + + const send = calls.find((c) => c.url.endsWith("/sendMessage")); + assert.ok(send); + const body = send.body as { chat_id: string; text: string }; + assert.equal(body.chat_id, "42"); + assert.equal(body.text, "⛔ HELM denied"); + assert.ok(!body.text.includes("secret-token")); +}); diff --git a/packages/js/helm-channel-bridge/src/telegram.ts b/packages/js/helm-channel-bridge/src/telegram.ts new file mode 100644 index 0000000..6204867 --- /dev/null +++ b/packages/js/helm-channel-bridge/src/telegram.ts @@ -0,0 +1,307 @@ +// Telegram Bot API transport for the HELM channel bridge. +// +// Mechanism adapted from the Apache-2.0 Rowboat project (rowboatlabs/rowboat, +// apps/x/packages/core/src/channels/transports/telegram.ts); this is an +// original implementation. Deliberately dependency-free: the Bot API is plain +// HTTPS — getUpdates long polling (outbound only, works behind NAT) plus +// sendMessage. The operator supplies their own bot token (@BotFather) via +// environment variable ONLY; the token is never logged, persisted, or +// accepted from message/argument input. +// +// Fail-closed properties: +// - DMs only: group chats would let any member drive the bridge. +// - allowFrom is an explicit allowlist matched against BOTH the chat ID and +// the sender user ID; an empty allowlist denies everyone. Non-allowlisted +// chats are dropped silently — replying with a pairing hint would be a +// spam/discovery vector. +// - Update offsets are persisted only AFTER the update was fully handled, +// and an update is confirmed in-memory only after its offset was +// persisted. A persistence failure fails closed: the offset is not +// advanced, the error is surfaced, and Telegram redelivers the +// unconfirmed update. Delivery semantics are therefore at-least-once at +// the transport boundary with exactly-once confirmation per persisted +// offset — Telegram only confirms updates when a LATER getUpdates passes +// a higher offset, so without persistence every restart would redeliver +// the last batch; persisting before handling would instead lose commands +// on crash. Side-effecting handlers downstream must tolerate redelivery +// of the most recent unconfirmed update after a persistence failure. +// - 401/404 from the Bot API are terminal (token revoked / bot deleted); +// retrying forever would hammer the API and misreport status. + +import fs from "node:fs/promises"; +import path from "node:path"; +import type { FetchLike } from "./evaluator.js"; + +const POLL_TIMEOUT_S = 50; +const RETRY_DELAY_MS = 5000; +const MAX_RETRY_DELAY_MS = 60_000; + +export const TELEGRAM_BOT_TOKEN_ENV = "HELM_TELEGRAM_BOT_TOKEN"; + +export type TelegramTransportStatus = + | { state: "starting" } + | { state: "polling"; botUsername?: string } + | { state: "error"; error: string } + | { state: "disabled" }; + +export class TelegramApiError extends Error { + constructor( + message: string, + readonly code?: number, + ) { + super(message); + this.name = "TelegramApiError"; + } +} + +function isTerminal(error: unknown): boolean { + return error instanceof TelegramApiError && (error.code === 401 || error.code === 404); +} + +export interface TelegramUpdate { + update_id: number; + message?: { + message_id: number; + text?: string; + chat: { id: number; type: string }; + from?: { id: number; is_bot?: boolean }; + }; +} + +export interface TelegramTransportOptions { + /** Bot token, sourced from the HELM_TELEGRAM_BOT_TOKEN env var. Never logged. */ + botToken: string; + /** Explicit allowlist (as strings) matched against chat ID AND sender user ID. Empty allowlist denies everyone. */ + allowFrom: string[]; + /** JSON file holding { offset } across restarts. */ + stateFile: string; + /** + * chatId is the address to reply to; the caller owns reply routing. + * Awaited by the transport: the poll offset is persisted only after this + * handler completes, so it should return the handling promise (a rejected + * promise keeps the update unconfirmed and surfaces the error). + */ + onInbound: (senderKey: string, chatId: string, text: string) => void | Promise; + onStatus?: (status: TelegramTransportStatus) => void; + /** Injectable for tests. */ + fetch?: FetchLike; + /** Injectable for tests. */ + sleep?: (ms: number) => Promise; + pollTimeoutS?: number; +} + +export class TelegramTransport { + private abort: AbortController | null = null; + private stopped = false; + private offset = 0; + private botUsername: string | undefined; + + constructor(private readonly opts: TelegramTransportOptions) { + if (!opts.botToken.trim()) { + throw new Error( + `Telegram bot token is required — set the ${TELEGRAM_BOT_TOKEN_ENV} environment variable`, + ); + } + } + + async start(): Promise { + this.stopped = false; + this.opts.onStatus?.({ state: "starting" }); + void this.run(); + } + + stop(): void { + this.stopped = true; + this.abort?.abort(); + this.opts.onStatus?.({ state: "disabled" }); + } + + private fetchImpl(): FetchLike { + const impl = this.opts.fetch ?? globalThis.fetch as FetchLike | undefined; + if (!impl) { + throw new TelegramApiError("No fetch implementation is available"); + } + return impl; + } + + private async call(method: string, body?: unknown, signal?: AbortSignal): Promise { + const res = await this.fetchImpl()(`https://api.telegram.org/bot${this.opts.botToken}/${method}`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: body === undefined ? undefined : JSON.stringify(body), + ...(signal ? { signal } : {}), + }); + const payload = await res.json() as { + ok: boolean; + result?: unknown; + description?: string; + error_code?: number; + }; + if (!payload.ok) { + throw new TelegramApiError( + payload.description ?? `Telegram API error (${method})`, + payload.error_code, + ); + } + return payload.result; + } + + /** Load the persisted poll offset; called by start() and exposed for tests. */ + async restoreOffset(): Promise { + try { + const raw = await fs.readFile(this.opts.stateFile, "utf8"); + const parsed = JSON.parse(raw) as { offset?: unknown }; + if (typeof parsed.offset === "number" && Number.isFinite(parsed.offset)) { + this.offset = parsed.offset; + } + } catch { + // first run or unreadable state — start from 0 + } + } + + /** + * Persist the poll offset for an update that was fully handled. + * Fail closed: any persistence error throws, so the caller keeps the + * in-memory offset un-advanced and the update unconfirmed — a silent + * failure here would replay and re-execute side effects after restart. + */ + private async saveOffset(offset: number): Promise { + try { + await fs.mkdir(path.dirname(this.opts.stateFile), { recursive: true }); + await fs.writeFile(this.opts.stateFile, JSON.stringify({ offset })); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + throw new TelegramApiError( + `Failed to persist the Telegram poll offset — refusing to confirm processed updates: ${message}`, + ); + } + } + + private sleep(ms: number): Promise { + if (this.opts.sleep) return this.opts.sleep(ms); + return new Promise((resolve) => setTimeout(resolve, ms)); + } + + private async run(): Promise { + await this.restoreOffset(); + + // Validate the token, retrying transient failures with backoff. Only a + // definitive API rejection is terminal. + let delay = RETRY_DELAY_MS; + while (!this.stopped) { + try { + const me = await this.call("getMe") as { username?: string }; + if (this.stopped) return; + this.botUsername = me.username; + this.opts.onStatus?.({ state: "polling", botUsername: me.username }); + break; + } catch (error) { + if (this.stopped) return; + if (isTerminal(error)) { + this.opts.onStatus?.({ + state: "error", + error: "Bot token rejected — create a new token with @BotFather.", + }); + return; + } + const message = error instanceof Error ? error.message : String(error); + this.opts.onStatus?.({ state: "error", error: message }); + await this.sleep(delay); + delay = Math.min(delay * 2, MAX_RETRY_DELAY_MS); + } + } + + delay = RETRY_DELAY_MS; + let healthy = true; + while (!this.stopped) { + try { + await this.pollOnce(); + if (!healthy) { + healthy = true; + this.opts.onStatus?.({ state: "polling", botUsername: this.botUsername }); + } + delay = RETRY_DELAY_MS; + } catch (error) { + if (this.stopped) return; + if (isTerminal(error)) { + this.opts.onStatus?.({ + state: "error", + error: "Bot token rejected — create a new token with @BotFather.", + }); + return; + } + healthy = false; + const message = error instanceof Error ? error.message : String(error); + this.opts.onStatus?.({ state: "error", error: message }); + await this.sleep(delay); + delay = Math.min(delay * 2, MAX_RETRY_DELAY_MS); + } + } + } + + /** One getUpdates round; public so tests can drive the poll loop deterministically. */ + async pollOnce(): Promise { + this.abort = new AbortController(); + const updates = await this.call( + "getUpdates", + { + timeout: this.opts.pollTimeoutS ?? POLL_TIMEOUT_S, + offset: this.offset, + allowed_updates: ["message"], + }, + this.abort.signal, + ) as TelegramUpdate[]; + for (const update of updates) { + // Handle FIRST: the offset is advanced only after the update was fully + // handled, so a crash or handler failure can never lose a command. + await this.processUpdate(update); + const next = update.update_id + 1; + // Persist BEFORE confirming in-memory: if persistence fails, fail + // closed — the offset stays un-advanced, the error propagates to the + // poll loop's error status, and Telegram redelivers the update. + await this.saveOffset(next); + this.offset = next; + } + } + + /** Route one update; public so tests can exercise authorization directly. */ + async processUpdate(update: TelegramUpdate): Promise { + const message = update.message; + if (!message?.text || message.from?.is_bot) return; + // DMs only: group chats would let any member drive the bridge. + if (message.chat.type !== "private") return; + const chatId = String(message.chat.id); + const fromId = message.from ? String(message.from.id) : null; + // Both the chat and the sender user must be allowlisted: a allowlisted + // chat must not be drivable by an unknown sender identity. + if (!this.opts.allowFrom.includes(chatId) || fromId === null || !this.opts.allowFrom.includes(fromId)) { + // Silent drop — replying with a pairing hint would be a spam/discovery + // vector for anyone who finds the bot. + return; + } + await this.opts.onInbound(`telegram:${chatId}`, chatId, message.text); + } + + async send(chatId: string, text: string): Promise { + await this.call("sendMessage", { chat_id: chatId, text }); + } +} + +/** + * Build transport options with the bot token sourced from the environment. + * The token MUST come from HELM_TELEGRAM_BOT_TOKEN (or an equivalent + * operator-managed env var); it is never read from config files, command + * arguments, or inbound messages. + */ +export function telegramOptionsFromEnv( + env: NodeJS.ProcessEnv, + rest: Omit, +): TelegramTransportOptions { + const token = env[TELEGRAM_BOT_TOKEN_ENV]?.trim() ?? ""; + if (token === "") { + throw new Error( + `Telegram bot token missing — set the ${TELEGRAM_BOT_TOKEN_ENV} environment variable`, + ); + } + return { ...rest, botToken: token }; +} diff --git a/packages/js/helm-channel-bridge/tsconfig.json b/packages/js/helm-channel-bridge/tsconfig.json new file mode 100644 index 0000000..0765abb --- /dev/null +++ b/packages/js/helm-channel-bridge/tsconfig.json @@ -0,0 +1,15 @@ +{ + "compilerOptions": { + "target": "ES2022", + "module": "NodeNext", + "moduleResolution": "NodeNext", + "declaration": true, + "outDir": "dist", + "rootDir": "src", + "strict": true, + "lib": ["ES2022", "DOM"], + "types": ["node"], + "skipLibCheck": true + }, + "include": ["src/**/*.ts"] +}