From ab7b6bc90e037558b833aff8cf0d7c0cc4801ddd Mon Sep 17 00:00:00 2001 From: TerrifiedBug Date: Fri, 14 Aug 2026 14:03:15 +0100 Subject: [PATCH] fix: report telegram ask provenance --- docs/guide.md | 5 + src/index.ts | 192 +++++++++++++++++++++++++++++++-------- src/index.wiring.test.ts | 168 +++++++++++++++++++++++++++++++--- src/prompts.ts | 3 +- 4 files changed, 315 insertions(+), 53 deletions(-) diff --git a/docs/guide.md b/docs/guide.md index 1b0282d..d67a111 100644 --- a/docs/guide.md +++ b/docs/guide.md @@ -280,6 +280,11 @@ count as a mention. Requests are responder-, chat-, topic-, message-, and nonce-bound, stay answerable while the owning session runs, and use the shared state directory for cross-process answers. + Every result starts with an `Ask provenance` JSON record. `posted` lists only + surfaces that successfully received the question, `answeredBy` names the + accepted answer's origin, and `errors` preserves per-surface posting failures. A + terminal answer therefore cannot hide a failed Telegram post: the result + includes a `SURFACE ERROR [telegram]` line alongside the answer. `telegram_send` and `telegram_react` refuse any chat the inbound gate would not deliver from. `telegram_ask` responds only to the exact user who originated the turn. diff --git a/src/index.ts b/src/index.ts index 32c62a1..8783bed 100644 --- a/src/index.ts +++ b/src/index.ts @@ -319,6 +319,49 @@ export interface AskSurface { run(signal: AbortSignal): Promise; } +type AskSurfaceName = "terminal" | "telegram"; +type AskToolResult = { content: ContentBlock[]; isError?: true }; +type AskDecision = { result: AskToolResult; answeredBy?: AskSurfaceName }; +type AskSurfaceState = Record; +type AskSurfaceErrors = Partial>; + +/** Keep answer text compatible while making delivery and answer provenance + * machine-readable at the start of every telegram_ask result. */ +function withAskProvenance( + result: AskToolResult, + posted: AskSurfaceState, + answeredBy: AskSurfaceName | undefined, + surfaceErrors: AskSurfaceErrors, +): AskToolResult { + const postedSurfaces = (["terminal", "telegram"] as const).filter((surface) => posted[surface]); + const errors = { + ...(surfaceErrors.terminal === undefined ? {} : { terminal: surfaceErrors.terminal }), + ...(surfaceErrors.telegram === undefined ? {} : { telegram: surfaceErrors.telegram }), + }; + const lines = [ + `Ask provenance: ${JSON.stringify({ + posted: postedSurfaces, + ...(answeredBy === undefined ? {} : { answeredBy }), + errors, + })}`, + ...(surfaceErrors.terminal === undefined + ? [] + : [`SURFACE ERROR [terminal]: ${surfaceErrors.terminal}`]), + ...(surfaceErrors.telegram === undefined + ? [] + : [`SURFACE ERROR [telegram]: ${surfaceErrors.telegram}`]), + ]; + const first = result.content[0]; + if (first?.type === "text") lines.push(first.text); + return { + ...result, + content: [ + { type: "text", text: lines.join("\n") }, + ...(first?.type === "text" ? result.content.slice(1) : result.content), + ], + }; +} + /** * Present the same question on several surfaces at once (the terminal picker and * Telegram) and take whichever decides first, aborting the rest. A surface that @@ -1912,7 +1955,7 @@ export default function telegramExtension(pi: ExtensionAPI): void { name: "telegram_ask", label: "Telegram Ask", description: - "Ask the user one or more questions: single-select, multi-select, or free-text. Provide 2-8 options for a choice, or omit options for a free-text answer the user types as a reply. While active it shows the question on both the terminal and Telegram (each when available) and returns whichever the user answers first. It replaces the built-in ask on Telegram-originated turns and while away/always mode is on; when both are offered you are at the terminal, so prefer ask unless the question should also reach Telegram. Use it exactly as you would use ask.", + "Ask the user one or more questions: single-select, multi-select, or free-text. Provide 2-8 options for a choice, or omit options for a free-text answer the user types as a reply. While active it shows the question on both the terminal and Telegram (each when available) and returns whichever the user answers first. Every result names the surfaces successfully posted, the answer origin, and any per-surface posting error. It replaces the built-in ask on Telegram-originated turns and while away/always mode is on; when both are offered you are at the terminal, so prefer ask unless the question should also reach Telegram. Use it exactly as you would use ask.", approval: "read", // Agent-initiated custom turns (including conductor ticks) bypass // before_agent_start. A daemon profile therefore keeps its only interactive @@ -1957,59 +2000,132 @@ export default function telegramExtension(pi: ExtensionAPI): void { if (!target && !canTerminal) { return errorResult("telegram_ask has no surface available — no Telegram target and no interactive terminal."); } - const surfaces: AskSurface<{ content: ContentBlock[]; isError?: true }>[] = []; + const posted: AskSurfaceState = { terminal: false, telegram: false }; + const surfaceErrors: AskSurfaceErrors = {}; + const telegramPosting = target === undefined ? undefined : Promise.withResolvers(); + let telegramStarted = false; + const surfaces: AskSurface[] = []; if (target) { surfaces.push({ run: async (sig) => { - const outcome = await promptController.ask(target, questions, sig, { supersededText: "☑️ Closed at the terminal." }); - if (outcome.status === "answered") return { content: [{ type: "text", text: formatPromptResult(outcome) }] }; - if (outcome.status === "cancelled") return errorResult(formatPromptResult(outcome)); - return undefined; // expired, or superseded because the terminal was answered first + telegramStarted = true; + try { + const outcome = await promptController.ask(target, questions, sig, { + supersededText: "☑️ Closed at the terminal.", + onPosted: () => { + posted.telegram = true; + telegramPosting?.resolve(); + }, + }); + if (outcome.status === "answered") { + return { + result: { content: [{ type: "text", text: formatPromptResult(outcome) }] }, + answeredBy: "telegram", + }; + } + if (outcome.status === "cancelled") { + return { result: errorResult(formatPromptResult(outcome)) }; + } + return undefined; // expired, or superseded because the terminal was answered first + } catch (err) { + const detail = (err instanceof Error ? err.message : String(err)).replace(/\s+/g, " ").trim(); + surfaceErrors.telegram = detail || "unknown Telegram surface error"; + throw err; + } finally { + telegramPosting?.resolve(); + } }, }); } if (canTerminal) { surfaces.push({ run: async (sig) => { - const result = await ctx.ui.askDialog!( - questions.map((q) => ({ - id: q.id, - question: q.question, - options: q.options.map((o) => ({ label: o.label, ...(o.description ? { description: o.description } : {}) })), - ...(q.multi != null ? { multi: q.multi } : {}), - ...(q.recommended != null ? { recommended: q.recommended } : {}), - })), - { signal: sig }, - ); - // Aborted (a sibling surface won, or the turn stopped) → idle; only a real Esc cancels. - if (result === undefined) return sig.aborted ? undefined : errorResult("Question cancelled at the terminal."); - if (result.kind === "chat") { - return { content: [{ type: "text", text: "User chose to chat about this instead of answering." }] }; + try { + const pending = ctx.ui.askDialog!( + questions.map((q) => ({ + id: q.id, + question: q.question, + options: q.options.map((o) => ({ + label: o.label, + ...(o.description ? { description: o.description } : {}), + })), + ...(q.multi != null ? { multi: q.multi } : {}), + ...(q.recommended != null ? { recommended: q.recommended } : {}), + })), + { signal: sig }, + ); + posted.terminal = true; + const result = await pending; + // Aborted (a sibling surface won, or the turn stopped) → idle; only a real Esc cancels. + if (result === undefined) { + return sig.aborted + ? undefined + : { result: errorResult("Question cancelled at the terminal.") }; + } + if (result.kind === "chat") { + return { + result: { + content: [ + { type: "text", text: "User chose to chat about this instead of answering." }, + ], + }, + }; + } + if ( + result.results.length !== p.questions.length || + result.results.some((item, index) => item.id !== p.questions[index]?.id) + ) { + throw new Error("ask dialog returned results that do not match the requested questions"); + } + const answers = result.results.map((item) => ({ + id: item.id, + question: item.question, + selectedOptions: item.selectedOptions, + ...(item.customInput == null ? {} : { customInput: item.customInput }), + ...(item.note == null ? {} : { note: item.note }), + })); + return { + result: { + content: [ + { + type: "text", + text: formatPromptResult({ status: "answered", answers }), + }, + ], + }, + answeredBy: "terminal", + }; + } catch (err) { + if (sig.aborted) return undefined; + const detail = (err instanceof Error ? err.message : String(err)).replace(/\s+/g, " ").trim(); + surfaceErrors.terminal = detail || "unknown terminal surface error"; + throw err; } - if ( - result.results.length !== p.questions.length || - result.results.some((item, index) => item.id !== p.questions[index]?.id) - ) { - throw new Error("ask dialog returned results that do not match the requested questions"); - } - const answers = result.results.map((item) => ({ - id: item.id, - question: item.question, - selectedOptions: item.selectedOptions, - ...(item.customInput == null ? {} : { customInput: item.customInput }), - ...(item.note == null ? {} : { note: item.note }), - })); - return { content: [{ type: "text", text: formatPromptResult({ status: "answered", answers }) }] }; }, }); } + let decision: AskDecision; try { - const exhausted = (aborted: boolean): { content: ContentBlock[]; isError: true } => - errorResult(aborted ? "The question was cancelled because the task stopped." : "The question expired before it was answered."); - return await raceAskSurfaces(surfaces, exhausted, signal, (err) => log.debug(`[telegram] ask surface failed: ${String(err)}`)); + const exhausted = (aborted: boolean): AskDecision => ({ + result: errorResult( + aborted + ? "The question was cancelled because the task stopped." + : "The question expired before it was answered.", + ), + }); + decision = await raceAskSurfaces(surfaces, exhausted, signal, (err) => + log.debug(`[telegram] ask surface failed: ${String(err)}`), + ); } catch (err) { - return errorResult(err instanceof Error ? err.message : String(err)); + decision = { result: errorResult(err instanceof Error ? err.message : String(err)) }; } + if (telegramStarted) await telegramPosting?.promise; + return withAskProvenance( + decision.result, + posted, + decision.answeredBy, + surfaceErrors, + ); }, }); diff --git a/src/index.wiring.test.ts b/src/index.wiring.test.ts index a31e9ba..055858b 100644 --- a/src/index.wiring.test.ts +++ b/src/index.wiring.test.ts @@ -1,6 +1,6 @@ import type { ExtensionAPI } from "@oh-my-pi/pi-coding-agent"; import { afterEach, beforeEach, describe, expect, test } from "bun:test"; -import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { existsSync, mkdtempSync, readFileSync, readdirSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { basename, join } from "node:path"; import { type Access, defaultAccess, loadAccess } from "./access"; @@ -593,6 +593,20 @@ describe("telegram_ask execute (dual-surface)", () => { })), }); + const activateDualSurfaces = async (h: Harness): Promise => { + writeAccess({ enabled: false, allowFrom: ["42"] }); + writeFileSync(join(dir, ".env"), "TELEGRAM_BOT_TOKEN=111:wiring-test\n"); + await h.handlers.get("before_agent_start")?.[0]?.( + { + type: "before_agent_start", + prompt: + 'hi', + systemPrompt: [], + }, + { hasUI: true }, + ); + }; + test("maps a terminal submit to the answer", async () => { const h = harness(["ask"]); const res = await h.tools.get("telegram_ask")!.execute("t", { questions }, undefined, undefined, { @@ -600,7 +614,119 @@ describe("telegram_ask execute (dual-surface)", () => { ui: { askDialog: (qs: DialogQuestion[]) => submit(qs) }, }); expect(res.isError).toBeUndefined(); - expect(res.content[0].text).toContain("User selected: A"); + expect(res.content[0].text).toBe( + 'Ask provenance: {"posted":["terminal"],"answeredBy":"terminal","errors":{}}\nUser selected: A', + ); + }); + + test("dual-surface terminal answer reports both posts and terminal origin", async () => { + const h = harness(["ask", "read"]); + await activateDualSurfaces(h); + const telegramPosted = Promise.withResolvers(); + const telegramClosed = Promise.withResolvers(); + const realFetch = globalThis.fetch; + globalThis.fetch = (async (input: string | URL | Request) => { + const method = String(input).split("/").pop(); + if (method === "sendMessage") telegramPosted.resolve(); + if (method === "editMessageText") telegramClosed.resolve(); + return new Response(JSON.stringify({ ok: true, result: { message_id: 7 } }), { + headers: { "content-type": "application/json" }, + }); + }) as typeof fetch; + try { + const res = await h.tools.get("telegram_ask")!.execute( + "t", + { questions }, + undefined, + undefined, + { + hasUI: true, + ui: { + askDialog: async (qs: DialogQuestion[]) => { + await telegramPosted.promise; + return await submit(qs); + }, + }, + }, + ); + expect(res.content[0].text).toBe( + 'Ask provenance: {"posted":["terminal","telegram"],"answeredBy":"terminal","errors":{}}\nUser selected: A', + ); + await telegramClosed.promise; + } finally { + globalThis.fetch = realFetch; + } + }); + + test("dual-surface Telegram answer reports both posts and Telegram origin", async () => { + const h = harness(["ask", "read"]); + await activateDualSurfaces(h); + const realFetch = globalThis.fetch; + globalThis.fetch = (async () => + new Response(JSON.stringify({ ok: true, result: { message_id: 7 } }), { + headers: { "content-type": "application/json" }, + })) as typeof fetch; + try { + const result = h.tools.get("telegram_ask")!.execute( + "t", + { questions }, + undefined, + undefined, + { + hasUI: true, + ui: { + askDialog: ( + _qs: DialogQuestion[], + opts: { signal: AbortSignal }, + ) => + new Promise((resolve) => { + if (opts.signal.aborted) resolve(undefined); + else opts.signal.addEventListener("abort", () => resolve(undefined), { once: true }); + }), + }, + }, + ); + const answer = (async (): Promise => { + const prompts = join(dir, "prompts"); + for (let attempt = 0; attempt < 200; attempt += 1) { + const requestName = existsSync(prompts) + ? readdirSync(prompts).find( + (name) => name.endsWith(".json") && !name.endsWith(".answer.json"), + ) + : undefined; + if (requestName !== undefined) { + const request = JSON.parse( + readFileSync(join(prompts, requestName), "utf8"), + ) as { nonce: string }; + writeFileSync( + join(prompts, `${request.nonce}.answer.json`), + JSON.stringify({ + expiresAt: Date.now() + 60_000, + outcome: { + status: "answered", + answers: [ + { + id: "q", + question: "Pick one", + selectedOptions: ["B"], + }, + ], + }, + }), + ); + return; + } + await Bun.sleep(5); + } + throw new Error("Telegram prompt request was not persisted"); + })(); + const [res] = await Promise.all([result, answer]); + expect(res.content[0].text).toBe( + 'Ask provenance: {"posted":["terminal","telegram"],"answeredBy":"telegram","errors":{}}\nUser selected: B', + ); + } finally { + globalThis.fetch = realFetch; + } }); test("preserves a terminal note", async () => { @@ -630,19 +756,33 @@ describe("telegram_ask execute (dual-surface)", () => { expect(res.content[0].text.toLowerCase()).toContain("chat about this"); }); - test("terminal wins when the Telegram surface fails fast", async () => { - writeAccess({ allowFrom: [] }); // responder isn't authorized → Telegram surface rejects before any network call + test("Telegram post failure remains explicit beside a terminal answer", async () => { const h = harness(["ask", "read"]); - await h.handlers.get("before_agent_start")?.[0]?.( - { type: "before_agent_start", prompt: 'hi', systemPrompt: [] }, - {}, - ); - const res = await h.tools.get("telegram_ask")!.execute("t", { questions }, undefined, undefined, { - hasUI: true, - ui: { askDialog: (qs: DialogQuestion[]) => submit(qs, ["B"]) }, - }); - expect(res.isError).toBeUndefined(); - expect(res.content[0].text).toContain("User selected: B"); + await activateDualSurfaces(h); + const realFetch = globalThis.fetch; + globalThis.fetch = (async () => { + throw new Error("sendMessage unavailable"); + }) as typeof fetch; + try { + const res = await h.tools.get("telegram_ask")!.execute( + "t", + { questions }, + undefined, + undefined, + { + hasUI: true, + ui: { askDialog: (qs: DialogQuestion[]) => submit(qs, ["B"]) }, + }, + ); + expect(res.isError).toBeUndefined(); + expect(res.content[0].text).toBe( + 'Ask provenance: {"posted":["terminal"],"answeredBy":"terminal","errors":{"telegram":"sendMessage unavailable"}}\n' + + "SURFACE ERROR [telegram]: sendMessage unavailable\n" + + "User selected: B", + ); + } finally { + globalThis.fetch = realFetch; + } }); test("normalizes an omitted-options question into a free-text terminal dialog", async () => { diff --git a/src/prompts.ts b/src/prompts.ts index 8e51ef2..bfb5751 100644 --- a/src/prompts.ts +++ b/src/prompts.ts @@ -248,7 +248,7 @@ export class TelegramPromptController { target: PromptTarget, questions: PromptQuestion[], signal?: AbortSignal, - opts?: { supersededText?: string }, + opts?: { supersededText?: string; onPosted?: () => void }, ): Promise { if (!this.#authorize(target.responderId, target.chatId, target.chatType)) { throw new Error("The originating Telegram user is no longer authorized"); @@ -277,6 +277,7 @@ export class TelegramPromptController { text: first.text, reply_markup: first.reply_markup, }); + opts?.onPosted?.(); request.messageId = sent.message_id; await atomicJson(requestPath(nonce), request);