diff --git a/packages/coding-agent/src/sdk/bus/telegram-daemon-cli.ts b/packages/coding-agent/src/sdk/bus/telegram-daemon-cli.ts index e6e5e4078d..855fc0a6d8 100644 --- a/packages/coding-agent/src/sdk/bus/telegram-daemon-cli.ts +++ b/packages/coding-agent/src/sdk/bus/telegram-daemon-cli.ts @@ -1,6 +1,6 @@ import * as fs from "node:fs"; import * as path from "node:path"; -import { logger, postmortem } from "@gajae-code/utils"; +import { logger } from "@gajae-code/utils"; import { YAML } from "bun"; import { applyAtomicYamlPatches, setByPath } from "../../config/atomic-yaml-patch"; import type { Settings } from "../../config/settings"; @@ -16,7 +16,6 @@ import { type DaemonState, FilesystemTopicRegistryCasAuthority, loadInstallationHostId, - markDaemonOwnerStopped, readDaemonState, readOwnerFreshnessSnapshot, type TelegramDaemonOptions, @@ -306,27 +305,6 @@ export async function runDaemonInternal(argv: string[], deps: RunDaemonInternalD const watchdog = schedule(() => void watchdogTick(), OWNER_WATCHDOG_INTERVAL_MS); process.once("SIGTERM", onSignal); process.once("SIGINT", onSignal); - // The daemon releases ownership only after a fully quiesced, fully persisted - // shutdown. Every other ending - a failed final persist, a signal, an - // uncaught error inside a detached async chain - used to leave - // `ownershipPhase: "ready"` on disk for a process that no longer exists, and - // later readers attached to it. Observed in the field: a daemon wrote one - // heartbeat 559 ms after readiness, died on an uncaught topic-registry - // error, and was still advertising itself as ready eight hours later. - // - // `finally` covers a returning or throwing run(); the postmortem hook covers - // the fatal paths that call `process.exit()` without unwinding this frame. - const recordOwnerStopped = (): Promise => - markDaemonOwnerStopped({ - settings: settings as Settings, - ownerId, - acquisitionId: ownerId, - pid: deps.processPid ?? process.pid, - now: deps.now, - }); - const unregisterPostmortem = postmortem.register("telegram-daemon:owner-state", async () => { - await recordOwnerStopped(); - }); try { await daemon.run(); } finally { @@ -334,7 +312,5 @@ export async function runDaemonInternal(argv: string[], deps: RunDaemonInternalD unschedule(watchdog); process.off("SIGTERM", onSignal); process.off("SIGINT", onSignal); - unregisterPostmortem(); - await recordOwnerStopped(); } } diff --git a/packages/coding-agent/src/sdk/bus/telegram-daemon-contract.ts b/packages/coding-agent/src/sdk/bus/telegram-daemon-contract.ts index b277cd99a8..e43257d37e 100644 --- a/packages/coding-agent/src/sdk/bus/telegram-daemon-contract.ts +++ b/packages/coding-agent/src/sdk/bus/telegram-daemon-contract.ts @@ -70,26 +70,9 @@ export const NOTIFICATION_PROTOCOL_VERSION = 3; * and polling dependency chain. Generation 48 uses crash-durable callback * receipts, a legacy-disjoint random alias namespace, and exact topic leases. * Generation 49 drains every admitted session-message handler before final - * durable persistence and ownership release. Generation 50 resolves intermediate - * notifications-directory symlinks before native exact unlink while keeping - * final-component file symlinks fail-closed under AT_SYMLINK_NOFOLLOW (bounded - * #3761 multi-account activation repair). Generation 51 adds shared durable - * topic authority, archive recovery, and requires Telegram's documented error - * code for idempotent archive settlement. Generation 52 is claimed by the - * pre-readiness daemon-child exit diagnostics slice (#3761). Generation 53 - * renders multi-select state for ask-tool asks, not only durable workflow - * gates, and renumbers pre-numbered options exactly once around the selection - * marker. Generation 54 records owner `stoppedAt` on unclean daemon death - * (`markDaemonOwnerStopped` + postmortem/finally wiring) so a dead process - * cannot keep advertising itself as the ready owner (#3965). Generation 55 - * contains a shared-topic-authority outage: a failed lease renewal on the - * liveness heartbeat and a failed startup registry load are reported instead - * of escaping to the process-level fatal handler, authority-failure throws - * preserve their underlying cause, and the compensation fence retry is bounded. - * Generation 56 moves exact unlink and process-incarnation authority behind - * lazy native bindings for the startup-cost cut (#3846). + * durable persistence and ownership release. */ -export const DAEMON_GENERATION = 56; +export const DAEMON_GENERATION = 57; /** * Serving-compatibility boundary for daemon lifecycle requests. Epoch 5 diff --git a/packages/coding-agent/src/sdk/bus/telegram-daemon.ts b/packages/coding-agent/src/sdk/bus/telegram-daemon.ts index b4603722f8..281fd60894 100644 --- a/packages/coding-agent/src/sdk/bus/telegram-daemon.ts +++ b/packages/coding-agent/src/sdk/bus/telegram-daemon.ts @@ -48,8 +48,6 @@ import { buildCompactChoiceGrid, code, markdownToTelegramHtml, - SELECTION_MARK_CHECKED, - SELECTION_MARK_UNCHECKED, splitTelegramHtml, TELEGRAM_MESSAGE_LIMIT, TELEGRAM_PARSE_MODE, @@ -369,10 +367,6 @@ const BTW_PENDING_TTL_MS = 300_000; const PICKER_CALLBACK_PREFIX = "p:"; const ADOPTION_INTENT_SWEEP_INTERVAL_MS = 60_000; const BTW_MAX_PENDING = 256; - -/** How long a compensation fence retries a failed topic-registry persist before giving up (40 × 250ms ≈ 10s). */ -const COMPENSATION_FENCE_MAX_ATTEMPTS = 40; -const COMPENSATION_FENCE_RETRY_DELAY_MS = 250; const BTW_SHUTDOWN_JOIN_MS = 1_000; const BTW_USAGE_TEXT = "Usage: /btw "; const BTW_CAPACITY_TEXT = "Too many /btw questions are pending. Wait for one to finish and try again."; @@ -389,10 +383,6 @@ const BTW_QUESTION_LIMIT_TEXT = "Question must be at most 4096 Unicode scalar va type ParsedBtwCommand = { kind: "question"; question: string } | { kind: "ignored" }; type TelegramFileDownload = { bytes: Buffer } | { failure: "download_failed" | "too_large" }; class ThreadedModeCapabilityRefusal extends Error {} -/** Telegram answered `createForumTopic` with an explicit `ok: false` rejection. */ -class TopicCreationRejected extends Error {} -/** The transport suppressed `createForumTopic` (no response body to classify). */ -class TopicCreationSuppressed extends Error {} async function prepareTelegramImageAttachment(frame: Record): Promise> { if (frame.type !== "image_attachment") return frame; @@ -432,7 +422,7 @@ function isThreadedModeCapabilityRefusal(response: unknown): boolean { return ( ok === false && typeof description === "string" && - /(?:threaded mode|forum topics? (?:is|are) (?:disabled|not enabled)|(?:not allowed|not permitted|cannot|can't) create forum topics?|chat is not a forum)/i.test( + /(?:threaded mode|forum topics? (?:is|are) (?:disabled|not enabled)|(?:not allowed|not permitted|cannot|can't) create forum topics?)/i.test( description, ) ); @@ -3471,22 +3461,8 @@ export async function releaseDaemonOwnership(input: { } /** - * Record that this owner's process is gone, without claiming a clean handoff. - * - * {@link releaseDaemonOwnership} is the orderly path and runs only when the - * daemon quiesced and persisted; it also unlinks the ownership lock, which is - * correct for a handoff and wrong for a corpse. Every other way a daemon can - * end - an uncaught error, a failed final persist, a signal - previously left - * `ownershipPhase: "ready"` and a fresh-looking lock behind forever, so later - * readers attached to an owner that had not existed for hours. - * - * This writes `stoppedAt` and nothing else. The lock is left in place for the - * existing reclaim path to adjudicate, because a process on its way out is the - * least qualified party to decide who owns what next. - * - * Fenced on full owner identity: if the state no longer names this exact - * owner, acquisition, pid and incarnation, a successor already took over and - * marking *their* state stopped would be the same lie in the other direction. + * Record that this owner's process is gone without claiming a clean handoff. + * The ownership lock remains for the existing reclaim path. */ export async function markDaemonOwnerStopped(input: { settings: Pick; @@ -3502,8 +3478,7 @@ export async function markDaemonOwnerStopped(input: { const paths = daemonPaths(input.settings.getAgentDir()); try { const state = await readJson(fsImpl, paths.state); - if (!hasSafeDaemonStateShape(state)) return false; - if (state.stoppedAt !== undefined) return false; + if (!hasSafeDaemonStateShape(state) || state.stoppedAt !== undefined) return false; const acquisitionId = input.acquisitionId ?? input.ownerId; const pid = input.pid ?? state.pid; const incarnation = (input.pidIncarnation ?? defaultPidIncarnation)(pid); @@ -3512,8 +3487,8 @@ export async function markDaemonOwnerStopped(input: { state.acquisitionId !== acquisitionId || state.pid !== pid || (input.generation !== undefined && state.generation !== input.generation) || - !isProcessIncarnation(incarnation) || - state.incarnation !== incarnation + state.incarnation !== incarnation || + !isProcessIncarnation(incarnation) ) return false; const lock = await readOwnershipLock(fsImpl, paths.lock); @@ -3521,11 +3496,9 @@ export async function markDaemonOwnerStopped(input: { await writeJsonAtomic(fsImpl, paths.state, { ...state, stoppedAt: (input.now ?? Date.now)() }); return true; } catch { - // A dying process cannot afford to fail louder than it already is. return false; } } - /** Read the persisted daemon ownership state (or undefined when absent). */ export async function readDaemonState( settings: Pick, @@ -5134,8 +5107,6 @@ export class TelegramNotificationDaemon { private readonly flatIdentitySent = new Set(); /** Cached delivery boundary for the private owner chat or validation forum. */ private pairedChatPrivacy: PairedChatPrivacy | undefined; - /** Latched once Telegram confirms this chat cannot host forum topics. */ - private topicCapabilityRefused = false; /** Bot username from getMe, cached once at owner startup for group/forum command targeting. */ private botUsername: string | undefined; /** Sessions whose agent loop is currently busy (drives the typing indicator). */ @@ -6327,22 +6298,9 @@ export class TelegramNotificationDaemon { } const logicalSessionId = this.#logicalSessionId(session); if (session.logicalSessionIdTrusted) - void this.#renewTopicLease(logicalSessionId).then( - renewed => { - if (!renewed) this.dropSession(session, "topic_lease_lost"); - }, - error => { - // A momentarily unavailable shared topic authority must not escape the - // heartbeat timer as an unhandled rejection and take the whole daemon - // down; report it and let the next heartbeat retry the renewal. - logger.warn( - `notifications: topic lease renewal failed: ${sanitizeDiagnostic( - String(error), - this.opts.botToken, - )}`, - ); - }, - ); + void this.#renewTopicLease(logicalSessionId).then(renewed => { + if (!renewed) this.dropSession(session, "topic_lease_lost"); + }); if (session.ws.readyState === WebSocket.OPEN) { const nonce = `${session.sessionId}:${t}:${Math.random().toString(36).slice(2)}`; session.awaitingNonce = nonce; @@ -7917,6 +7875,8 @@ export class TelegramNotificationDaemon { let acceptedTopicId: string | undefined; let acceptedTopicCompensated = false; let acceptedTopicArchiveAttempted = false; + let creationSuppressed = false; + let creationRejected = false; let adoptedTopicId: number | undefined; const adoptionIntentCandidate = this.#adoptionIntents.bySession(sessionId); try { @@ -7955,10 +7915,10 @@ export class TelegramNotificationDaemon { const res = (await this.botApi.call("createForumTopic", { chat_id: this.opts.chatId, name })) as | { result?: { message_thread_id?: unknown } } | undefined; - // Classification must travel with the rejection, not through - // closure flags: `getOrCreateTopic` shares one in-flight create - // promise, so every other awaiter observes only the error. - if (res === undefined) throw new TopicCreationSuppressed(); + if (res === undefined) { + creationSuppressed = true; + return undefined; + } if (isThreadedModeCapabilityRefusal(res)) throw new ThreadedModeCapabilityRefusal(); const response = res as { ok?: unknown; @@ -7966,8 +7926,8 @@ export class TelegramNotificationDaemon { }; const tid = response.result?.message_thread_id; if (typeof tid !== "number" || !Number.isSafeInteger(tid) || tid <= 0) { - if (response.ok === false) throw new TopicCreationRejected(); - this.#malformedTopicCreateEndpoints.set(sessionId, creationEndpointKey); + creationRejected = response.ok === false; + if (!creationRejected) this.#malformedTopicCreateEndpoints.set(sessionId, creationEndpointKey); throw new Error("createForumTopic: invalid message_thread_id"); } acceptedTopicId = String(tid); @@ -8088,12 +8048,7 @@ export class TelegramNotificationDaemon { } catch (err) { if (adoptedTopicId !== undefined) this.#adoptionIntents.releaseClaim(adoptedTopicId, sessionId); if (adoptionIntentCandidate) this.topics.abandonCreateClaim(sessionId, creationLeaseEpoch); - if ( - err instanceof TopicCreationSuppressed || - err instanceof TopicCreationRejected || - err instanceof ThreadedModeCapabilityRefusal - ) { - if (err instanceof ThreadedModeCapabilityRefusal) this.topicCapabilityRefused = true; + if (creationSuppressed || creationRejected || err instanceof ThreadedModeCapabilityRefusal) { if (this.topics.abandonCreateClaim(sessionId, creationLeaseEpoch)) await this.persistTopics(); return undefined; } @@ -8354,8 +8309,8 @@ export class TelegramNotificationDaemon { let accepted = false; try { accepted = await authority.compareAndSet(expectedGeneration, snapshot); - } catch (error) { - throw new Error("shared topic authority unavailable", { cause: error }); + } catch { + throw new Error("shared topic authority unavailable"); } if (accepted) { this.topics.markRegistryPublished(nextGeneration); @@ -8426,25 +8381,12 @@ export class TelegramNotificationDaemon { if (existing) return await existing; const retry = this.effects.track( (async () => { - for (let attempt = 1; ; attempt++) { + for (;;) { try { await this.persistTopics(); return; - } catch (error) { - if (attempt >= COMPENSATION_FENCE_MAX_ATTEMPTS) { - // A shared topic authority that stays unavailable must not keep this - // fence spinning every 250ms forever: it churns the registry, and its - // pending effect prevents a shutdown from quiescing. Give up and let - // the next scan/session pass retry the persist. - logger.warn( - `notifications: compensation fence gave up after ${attempt} attempts: ${sanitizeDiagnostic( - String(error), - this.opts.botToken, - )}`, - ); - return; - } - await this.runtime.sleep(COMPENSATION_FENCE_RETRY_DELAY_MS); + } catch { + await this.runtime.sleep(250); } } })(), @@ -8472,15 +8414,15 @@ export class TelegramNotificationDaemon { let accepted = false; try { accepted = await authority.compareAndSet(expectedGeneration, snapshot); - } catch (error) { - throw new Error("shared topic authority unavailable", { cause: error }); + } catch { + throw new Error("shared topic authority unavailable"); } if (!accepted) { let winner: TopicRegistryState | undefined; try { winner = parseTopicRegistryState(await authority.read()); - } catch (error) { - throw new Error("shared topic authority unavailable", { cause: error }); + } catch { + throw new Error("shared topic authority unavailable"); } if (!winner) throw new Error("shared topic authority conflict"); this.#replaceTopicAuthority(winner); @@ -8553,8 +8495,8 @@ export class TelegramNotificationDaemon { if (this.opts.topicRegistryAuthority) { try { raw = await this.opts.topicRegistryAuthority.read(); - } catch (error) { - throw new Error("shared topic authority unavailable", { cause: error }); + } catch { + throw new Error("shared topic authority unavailable"); } if (!raw) throw new Error("shared topic authority unavailable"); } @@ -9612,7 +9554,6 @@ export class TelegramNotificationDaemon { } private async pairedChatAllowsTopics(): Promise { - if (this.topicCapabilityRefused) return false; const privacy = await this.resolvePairedChatPrivacy(); return privacy === "private" || privacy === "validation-forum"; } @@ -9654,11 +9595,7 @@ export class TelegramNotificationDaemon { private startFlushTimer(): void { this.runtime.startInterval("telegram-flush", RATE_LIMIT_FLUSH_INTERVAL_MS, () => { if (!this.running || this.pool.pending === 0) return; - // `flushPool` only swallows failures for the shared chain; the promise it - // returns still rejects, and an unhandled rejection here exits the daemon. - void this.flushPool().catch(error => { - logger.warn(`notifications: queue flush failed: ${sanitizeDiagnostic(String(error), this.opts.botToken)}`); - }); + void this.flushPool(); }); } @@ -9698,27 +9635,11 @@ export class TelegramNotificationDaemon { this.runtime.stopInterval("telegram-owner-heartbeat"); } - /** - * Run a root scan, guarding against overlapping scans from the timer + loop. - * - * A reconciliation pass is retried every scan interval, so a failed one must - * never tear the owner down. Both callers are fatal boundaries: the timer - * fires and forgets, and the run loop awaits without a handler. A rejection - * escaping either one (a shared topic authority whose lock or compare-and-set - * is momentarily unavailable is the observed case) reaches the process-level - * handler, which exits the daemon: every session topic is then left as an - * unarchived shell that answers nothing, and the sessions that were still - * live lose notifications too. Report the failure and let the next scan - * retry instead. - */ + /** Run a root scan, guarding against overlapping scans from the timer + loop. */ private async runScan(): Promise { - try { - await this.runtime.runExclusive("telegram-scan", async () => { - await this.scanRoots(); - }); - } catch (error) { - logger.warn(`notifications: session scan failed: ${sanitizeDiagnostic(String(error), this.opts.botToken)}`); - } + await this.runtime.runExclusive("telegram-scan", async () => { + await this.scanRoots(); + }); } private startScanTimer(): void { @@ -10494,9 +10415,7 @@ export class TelegramNotificationDaemon { : [], ); const displayOptions = options.map((option, index) => - multiSelect - ? `${selectedOptionIndices.has(index) ? SELECTION_MARK_CHECKED : SELECTION_MARK_UNCHECKED} ${option}` - : option, + multiSelect ? `${selectedOptionIndices.has(index) ? "☑" : "☐"} ${option}` : option, ); const rendered = buildActionMessage({ kind: msg.kind ?? "ask", @@ -12311,21 +12230,7 @@ export class TelegramNotificationDaemon { await this.refreshBotIdentity(); await this.registerBotCommands(); } - try { - await this.loadTopics(); - } catch (error) { - // A shared topic authority that is momentarily unavailable at startup must - // not kill the daemon before it starts serving: report the failure and - // continue with an empty in-memory registry. Connected sessions re-create - // their topics through ensureTopic, and later persist passes read the - // winner back from the authority (same philosophy as runScan's retry). - logger.warn( - `notifications: topic registry load failed; continuing with empty registry: ${sanitizeDiagnostic( - String(error), - this.opts.botToken, - )}`, - ); - } + await this.loadTopics(); if (!this.validationMode()) { await this.loadAdoptionIntents(); this.startAdoptionSweepTimer(); diff --git a/packages/coding-agent/test/manifests/telegram-baseline-v1.json b/packages/coding-agent/test/manifests/telegram-baseline-v1.json index 59ff77d9dc..974c925084 100644 --- a/packages/coding-agent/test/manifests/telegram-baseline-v1.json +++ b/packages/coding-agent/test/manifests/telegram-baseline-v1.json @@ -316,13 +316,6 @@ "packages/coding-agent/test/notifications-telegram-daemon-cas.test.ts" ] }, - { - "argv": [ - "bun", - "test", - "packages/coding-agent/test/notifications-telegram-daemon-owner-postmortem.test.ts" - ] - }, { "argv": [ "bun", diff --git a/packages/coding-agent/test/notifications-telegram-daemon.test.ts b/packages/coding-agent/test/notifications-telegram-daemon.test.ts index 1d5d49768b..fc1ab0e2e9 100644 --- a/packages/coding-agent/test/notifications-telegram-daemon.test.ts +++ b/packages/coding-agent/test/notifications-telegram-daemon.test.ts @@ -2986,7 +2986,7 @@ describe("telegram daemon", () => { }), ); } - test("keeps wire protocol 3 through generation 55 lazy native authority", () => { + test("keeps wire protocol 3 through generation 57 durable topic authority", () => { expect(NOTIFICATION_PROTOCOL_VERSION).toBe(3); // Generations 34 and 35 add media conversion and topic adoption; generation // 36 bound managed-session replacement to exact native filesystem authority, @@ -3004,19 +3004,8 @@ describe("telegram daemon", () => { // generation 47 settles failed staged revocation; generation 48 makes receipts // crash-durable, aliases legacy-disjoint, and topic authority exact; generation // 49 drains admitted session handlers before final persistence and ownership release; - // generation 50 resolves intermediate notifications-directory symlinks before - // native exact unlink while keeping final-component file symlinks fail-closed; - // generation 51 adds shared durable topic authority and archive recovery and - // binds idempotent archive settlement to Telegram error code 400; generation - // 52 is claimed by the pre-readiness daemon-child exit diagnostics slice; - // generation 53 renders multi-select state for ask-tool asks and renumbers - // pre-numbered options exactly once around the selection marker; - // generation 54 records owner stoppedAt on unclean daemon death so a dead - // process cannot keep advertising itself as the ready owner (#3965). - // generation 55 hardens the shared topic authority outage path (#3974). - // generation 56 moves exact unlink and process-incarnation authority behind - // lazy native bindings (#3846). - expect(DAEMON_GENERATION).toBe(56); + // generation 50 adds shared durable topic authority and archive recovery. + expect(DAEMON_GENERATION).toBe(57); }); test.each([ "1", @@ -6184,35 +6173,6 @@ describe("telegram daemon", () => { }); }); - test("multi-select state renumbers pre-numbered options exactly once", async () => { - FakeWs.instances = []; - const agentDir = tempAgentDir(); - const bot = new FakeBotApi(); - const daemon = new TelegramNotificationDaemon({ - settings: setPrivateAgentDir(settings(agentDir), agentDir), - ownerId: "owner", - botToken: "tok", - chatId: "42", - botApi: bot, - rich: { enabled: false }, - WebSocketImpl: FakeWs as never, - }); - daemon.connectSession("S", "ws://s", "ts"); - // Deep-interview options arrive pre-numbered by the ask tool. - await daemon.handleSessionMessage(daemon.sessions.get("S")!, { - type: "action_needed", - kind: "ask", - id: "ask", - question: "Pick all", - options: ["1. Alpha", "2. Beta"], - selectedOptionIndices: [0], - }); - const sent = bot.calls.find(call => call.method === "sendMessage")!.body; - expect(sent.text).toContain("1. ☑ Alpha"); - expect(sent.text).toContain("2. ☐ Beta"); - expect(sent.text).not.toContain("1. Alpha"); - }); - test("callback alias reply is delivered when Telegram callback ack fails", async () => { FakeWs.instances = []; const agentDir = tempAgentDir(); @@ -11210,62 +11170,6 @@ test("threaded mode off: frames fall back to the flat paired chat with a one-tim expect(ask).toBeTruthy(); expect(ask!.body.reply_markup?.inline_keyboard?.length).toBeGreaterThan(0); }); -test("private chat without Threaded Mode: concurrent frames all deliver flat", async () => { - const agentDir = tempAgentDir(); - const bot = new FakeBotApi(); - // Verbatim Bot API rejection for a paired private chat whose bot has no - // Threaded Mode; the slow response makes a second frame join the shared - // in-flight create instead of issuing its own. - bot.call = (async (method: string, body: unknown) => { - bot.calls.push({ method, body: body as never }); - if (method === "createForumTopic") { - await Bun.sleep(30); - return { ok: false, error_code: 400, description: "Bad Request: the chat is not a forum" }; - } - if (method === "getChat") return { ok: true, result: { type: "private" } }; - if (method === "sendMessage") return { ok: true, result: { message_id: bot.calls.length } }; - return { ok: true, result: true }; - }) as never; - const daemon = new TelegramNotificationDaemon({ - settings: settings(agentDir), - ownerId: "owner", - botToken: "tok", - chatId: "42", - botApi: bot, - rich: { enabled: false }, - sound: "none", - }); - const session = { sessionId: "S", token: "tok", ws: { readyState: 1, send() {} }, pending: new Map() }; - - await Promise.all([ - daemon.handleSessionMessage(session as never, { - type: "identity_header", - sessionId: "S", - repo: "r", - branch: "b", - }), - daemon.handleSessionMessage(session as never, { - type: "action_needed", - sessionId: "S", - id: "ask1", - kind: "ask", - question: "Proceed?", - options: ["Yes", "No"], - }), - ]); - // A later frame must not re-attempt the refused capability. - await daemon.handleSessionMessage(session as never, { - type: "context_update", - sessionId: "S", - lastMessage: "hello world", - }); - - expect(bot.calls.filter(call => call.method === "createForumTopic")).toHaveLength(1); - const sends = bot.calls.filter(call => call.method === "sendMessage"); - expect(sends.every(call => call.body.message_thread_id === undefined)).toBe(true); - expect(sends.some(call => String(call.body.text).includes("Proceed?"))).toBe(true); - expect(sends.some(call => String(call.body.text).includes("hello world"))).toBe(true); -}); test("topic creation transport failures fail closed without flat delivery", async () => { const agentDir = tempAgentDir(); const bot = new FakeBotApi(); @@ -18976,7 +18880,7 @@ describe("telegram daemon /btw reservation and capability boundaries", () => { expect(routed.every(call => call.body.message_thread_id !== 77)).toBe(true); }); test.each([ - ["accepted remote archive", async () => ({ ok: true, result: true }), "inactive", "42", "42", [77], false], + ["accepted remote archive", async () => ({ ok: true, result: true }), "inactive", "42", "42", [77]], [ "already closed remote topic", async () => ({ ok: false, error_code: 400, description: "Bad Request: TOPIC_NOT_FOUND" }), @@ -18984,20 +18888,7 @@ describe("telegram daemon /btw reservation and capability boundaries", () => { "42", "42", [77], - false, ], - ...([401, 403, 429, 500] as const).map( - errorCode => - [ - `allowlisted archive text under ${errorCode}`, - async () => ({ ok: false, error_code: errorCode, description: "Bad Request: TOPIC_NOT_FOUND" }), - "archive_pending", - "42", - "42", - [77], - true, - ] as const, - ), [ "ambiguous remote archive", async () => ({ ok: false, description: "transport unavailable" }), @@ -19005,7 +18896,38 @@ describe("telegram daemon /btw reservation and capability boundaries", () => { "42", "42", [77], - true, + ], + [ + "same settled text with a non-400 error remains retryable", + async () => ({ ok: false, error_code: 500, description: "Bad Request: TOPIC_NOT_FOUND" }), + "archive_pending", + "42", + "42", + [77], + ], + [ + "same settled text with 401 remains retryable", + async () => ({ ok: false, error_code: 401, description: "Bad Request: TOPIC_NOT_FOUND" }), + "archive_pending", + "42", + "42", + [77], + ], + [ + "same settled text with 403 remains retryable", + async () => ({ ok: false, error_code: 403, description: "Bad Request: TOPIC_NOT_FOUND" }), + "archive_pending", + "42", + "42", + [77], + ], + [ + "same settled text with 429 remains retryable", + async () => ({ ok: false, error_code: 429, description: "Bad Request: TOPIC_NOT_FOUND" }), + "archive_pending", + "42", + "42", + [77], ], [ "re-paired chat retains an old-chat archive fence", @@ -19014,9 +18936,8 @@ describe("telegram daemon /btw reservation and capability boundaries", () => { "43", "42", [], - false, ], - ] as const)("startup scan reconciles a crash-persisted archive fence after %s", async (_outcome, archiveResult, authorityState, recordChatId, pairedChatId, expectedArchivedTopics, expectsRetryJob) => { + ] as const)("startup scan reconciles a crash-persisted archive fence after %s", async (_outcome, archiveResult, authorityState, recordChatId, pairedChatId, expectedArchivedTopics) => { const agentDir = tempAgentDir(); const topicsPath = path.join(daemonPaths(agentDir).dir, "telegram-topics.json"); fs.mkdirSync(path.dirname(topicsPath), { recursive: true }); @@ -19064,7 +18985,6 @@ describe("telegram daemon /btw reservation and capability boundaries", () => { ).toEqual([...expectedArchivedTopics]); const persisted = JSON.parse(fs.readFileSync(topicsPath, "utf8")); expect(persisted.topics.S).toMatchObject({ topicId: "77", chatId: recordChatId, authorityState }); - if (expectsRetryJob) expect(persisted.archiveJobs.S).toMatchObject({ topicId: "77", attempt: 1 }); }); test("failed close publication restores only close authority while retaining a concurrent user rename across restart", async () => { @@ -19629,15 +19549,8 @@ describe("telegram daemon /btw reservation and capability boundaries", () => { releaseSend.resolve({ ok: true, result: { message_id: 7 } }); await handlerPersistStarted.promise; - // `join` counts its deadline in wall-clock from the call, and the statements - // between the call and `releaseHandlerPersist` below (the held-state race and - // its assertion) run inside that window. A deadline as tight as the released - // work itself therefore expires on a loaded runner and reports `false` for a - // barrier that did hold. The held half is proven by the race below, not by - // the deadline, so bound it generously — the daemon's own shutdown join uses - // 1s. const joinedSettled = Promise.withResolvers(); - const joinedEffect = effects.join(5_000).then(value => { + const joinedEffect = effects.join(100).then(value => { joinedSettled.resolve(value); return value; }); @@ -22518,175 +22431,3 @@ test("CAS winner advance after accepted create publishes the exact archive fence [2], ); }); - -test("a failed root scan is retried on the next tick instead of exiting the daemon owner", async () => { - // A dead session's topic is archived by the scan pass, and that pass persists - // through the shared topic authority. When the authority is momentarily - // unavailable the rejection used to escape the scan timer and the run loop, - // reach the process-level fatal handler, and exit the owner: every topic was - // then left behind as an unarchived shell that answers nothing. - const warn = spyOn(logger, "warn").mockImplementation(() => {}); - const escaped: unknown[] = []; - const onEscape = (reason: any): void => void escaped.push(reason?.stack ?? String(reason)); - process.on("unhandledRejection", onEscape); - const ticks: Array<() => void> = []; - let compareAndSetCalls = 0; - let armed = false; - let state = { - version: 2 as const, - registryGeneration: 1, - topics: { - S: { - topicId: "700", - topicOrigin: "daemon_created" as const, - sessionUuid: "dead-session", - identitySent: true, - createdAt: 1, - authorityEpoch: 0, - authorityState: "active" as const, - chatId: "42", - endpointKey: "ws://dead-endpoint|dead-token", - endpointDigest: "dead-digest", - endpointGeneration: 1, - }, - }, - }; - const authority = { - read: async () => state, - compareAndSet: async (_expectedGeneration: number, next: any) => { - if (!armed) { - state = next; - return true; - } - compareAndSetCalls++; - throw new Error("shared topic authority unavailable"); - }, - }; - const daemon = new TelegramNotificationDaemon({ - settings: settings(tempAgentDir()), - ownerId: "owner", - botToken: "token", - chatId: "42", - botApi: new FakeBotApi(), - installationHostId: "local-host", - topicRegistryAuthority: authority as never, - setIntervalImpl: ((tick: () => void) => { - ticks.push(tick); - return 0; - }) as never, - clearIntervalImpl: (() => {}) as never, - }); - try { - await daemon.loadTopics(); - expect((daemon as any).topics.get("S")?.topicId).toBe("700"); - armed = true; - (daemon as any).running = true; - (daemon as any).startScanTimer(); - expect(ticks).toHaveLength(1); - - const settle = async (): Promise => { - for (let i = 0; i < 50; i++) await new Promise(resolve => setTimeout(resolve, 1)); - }; - - const scan = spyOn(daemon, "scanRoots"); - ticks[0]!(); - await settle(); - // The pass reached the shared authority, failed, and reported the failure - // instead of letting the rejection reach the process-level fatal handler. - expect(compareAndSetCalls).toBeGreaterThanOrEqual(1); - expect(warn.mock.calls.some(call => String(call[0]).includes("session scan failed"))).toBe(true); - expect(escaped).toEqual([]); - // The owner is still running, so the next interval retries the same pass. - expect((daemon as any).running).toBe(true); - ticks[0]!(); - await settle(); - expect(scan.mock.calls).toHaveLength(2); - expect(escaped).toEqual([]); - scan.mockRestore(); - } finally { - process.off("unhandledRejection", onEscape); - warn.mockRestore(); - } -}); - -test("a topic lease renewal failure on the liveness heartbeat is reported instead of escaping as an unhandled rejection", async () => { - const warn = spyOn(logger, "warn").mockImplementation(() => {}); - const escaped: unknown[] = []; - const onEscape = (reason: any): void => void escaped.push(reason?.stack ?? String(reason)); - process.on("unhandledRejection", onEscape); - FakeWs.instances = []; - const liveness: Array<() => void> = []; - let armed = false; - let state = { - version: 2 as const, - registryGeneration: 1, - topics: { - S: { - topicId: "700", - topicOrigin: "daemon_created" as const, - sessionUuid: "s", - identitySent: true, - createdAt: 1, - authorityEpoch: 0, - authorityState: "active" as const, - chatId: "42", - endpointKey: "ws://session|token", - endpointDigest: "digest", - endpointGeneration: 1, - }, - }, - }; - const authority = { - read: async () => state, - compareAndSet: async (_expectedGeneration: number, next: any) => { - if (!armed) { - state = next; - return true; - } - throw new Error("shared topic authority unavailable"); - }, - }; - const daemon = new TelegramNotificationDaemon({ - settings: settings(tempAgentDir()), - ownerId: "owner", - botToken: "token", - chatId: "42", - botApi: new FakeBotApi(), - WebSocketImpl: FakeWs as any, - installationHostId: "local-host", - topicRegistryAuthority: authority as never, - setIntervalImpl: ((cb: () => void) => { - liveness.push(cb); - return 0; - }) as any, - clearIntervalImpl: (() => {}) as any, - }); - try { - await (daemon as any).loadTopics(); - const session = daemon.connectSession("S", "ws://session", "token"); - FakeWs.instances[0]!.dispatchEvent(new Event("open")); - FakeWs.instances[0]!.emit({ type: "hello", protocolVersion: 2, capabilities: ["client_ping_pong"] }); - FakeWs.instances[0]!.emit({ - type: "event_replay_result", - ok: true, - id: "telegram-startup-replay:S", - generation: 1, - lastSeq: 0, - events: [], - }); - for (let attempts = 0; attempts < 20 && liveness.length === 0; attempts++) await Bun.sleep(1); - expect(liveness).toHaveLength(1); - armed = true; - liveness[0]!(); - await new Promise(resolve => setTimeout(resolve, 50)); - // The heartbeat's lease renewal hits the dead authority. Before the fix the - // single-arg .then() let the rejection escape to the process-level fatal - // handler; now it is reported and the next heartbeat retries. - expect(escaped).toEqual([]); - expect(warn.mock.calls.some(call => String(call[0]).includes("topic lease renewal failed"))).toBe(true); - expect(session.logicalSessionIdTrusted).toBe(true); - } finally { - process.off("unhandledRejection", onEscape); - warn.mockRestore(); - } -}); diff --git a/packages/coding-agent/test/notifications-topic-registry.test.ts b/packages/coding-agent/test/notifications-topic-registry.test.ts index 68c4da2763..d150a01c7b 100644 --- a/packages/coding-agent/test/notifications-topic-registry.test.ts +++ b/packages/coding-agent/test/notifications-topic-registry.test.ts @@ -562,10 +562,8 @@ test("preserves a no-provenance endpoint claim before a held create can stage it await creating; expect(reg.endpointAuthority(binding)).toEqual({ state: "unique", sessionId: "B" }); }); -test("publishes generation 56 at serving epoch 5", () => { - // Generation 55: shared-topic-authority outage hardening (#3974). - // Generation 56: lazy native authority for startup-cost cut (#3846). - expect(DAEMON_GENERATION).toBe(56); +test("publishes generation 57 at serving epoch 5", () => { + expect(DAEMON_GENERATION).toBe(57); expect(SERVING_EPOCH).toBe(5); }); test("archives pending topics into retained inactive records", async () => { diff --git a/scripts/telegram-daemon-generation-manifest.json b/scripts/telegram-daemon-generation-manifest.json index 36b3d50661..74da8f971c 100644 --- a/scripts/telegram-daemon-generation-manifest.json +++ b/scripts/telegram-daemon-generation-manifest.json @@ -529,9 +529,9 @@ "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon-cli.ts:createLightweightDaemonSettings": "2b001f6739e0e4277eb93c689d5d6e5d341b006c071f65539a03fdb94d653aa4", "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon-cli.ts:loadLightweightDaemonSettings": "b6139460042f6ad3128c5be257e63845a6bf318d920e474ced453c214bc08e0a", "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon-cli.ts:ownerPidFromOwnerId": "46691373b2bee01f28f3817a6aa6a7efffe880c2cea337c89155582c98d952bf", - "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon-cli.ts:runDaemonInternal": "e535e82888898a2ace182b8c6fd1c5ef3b678348071ffa0a469c427c4c08ccaf", + "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon-cli.ts:runDaemonInternal": "17f044cb4ec349fbed63d8ea8c916f419137e0c6fd158eac6dac58de3b14db59", "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon-cli.ts:runDaemonSmoke": "6f085a667aa5c83de46d2d8945fb845c355fcbb43c46872342a44489203a5830", - "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon-contract.ts:DAEMON_GENERATION": "575f1b89451f393b9aec36d940178f5efba77e6929ad60c47078fafa970c5bca", + "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon-contract.ts:DAEMON_GENERATION": "bcf1dcbadded7293accf634704931acd871a17a1fe226cf37a5ee4de8f5fb38a", "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon-contract.ts:NOTIFICATION_PROTOCOL_VERSION": "b99289f651fedcf020d28dbaf6f07dd37e7e4a5f6dc1f5118b872112325f1e81", "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon-control.ts:DaemonProcessReference": "c3d13e3670a6245a1250c4ebfcd80a36dd8fc96c67ab64d9f979182bd117bc4e", "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon-control.ts:TelegramDaemonController": "7381b51cd968199876bfccd341ce79f1bcd895c9fb3899149394d6c54459f07f", @@ -577,7 +577,7 @@ "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon.ts:ensureTelegramDaemonRunning": "0dbc6e3450ee72827d720cf492b69c4659e2f366d4d2ff4c008cc19f793a5a19", "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon.ts:ensureTelegramDaemonRunningDetailed": "845f8f868eb2b22c1f9964cd88f536634d7ff76eee543ee4feb7518d538b66f6", "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon.ts:fetchWithRetry": "0bd340153367adbfe58dc51df9fc39f3c163104aa2040ca76cdbb9142eb68ac2", - "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon.ts:handleSessionMessage": "286ab96d5b02245fa9b38494e0ae952e968d07a0c6abbc6557749fccf8f36c64", + "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon.ts:handleSessionMessage": "dd55fdc5941e6218ae86d5d7e3c0c53d0102158982b136bd65145c909520cea7", "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon.ts:handleTelegramUpdate": "8f1dc6c87dcbb3000c1de03195d2e8eb000f69a36f44f09edbfcbbdf9203f1dd", "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon.ts:hasSafeDaemonStateShape": "4c015214f5dd344299328312451d3a6a093c91373e50d449c108c1b5f01e5fcb", "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon.ts:historicalStateSerializer": "52fb9019a96fd386dca753b594b02e16bc888fe18970bfefcba5268823ae6547", @@ -616,7 +616,7 @@ "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon.ts:retireProvisionalDaemonOwnership": "d5f45044ea524f0694691bda67f47f9d74eaa5506926b718f40f2613a892d82f", "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon.ts:revokeCallbackAliases": "42407ae6ce220a36e06f6b74a339b59dc1ec1efd44fa0e4c80fb2cf43d4fb5ba", "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon.ts:rollbackOwnershipLockRebind": "7e9bc148e69268c393051e0b87417c20ce44824e777d4be050b9e91607c410a1", - "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon.ts:run": "b124bf7e7515261bfaa68b9bc04df3316cbc6c998e81e4bd69a6733a0436f44d", + "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon.ts:run": "70a97480625c5c104513a364a01645391abefa5352adb30f6dd21b464fefa68b", "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon.ts:spawnTelegramDaemonOwner": "bbb56ea3a91bb24592fe8e7128261fd75424a1387b75b295a3e046fec1aba08a", "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon.ts:startLifecycleControl": "237cf7c7881048e0e4329650567c8a47f597abe43206fe2f29376eb912cd6d1c", "telegram:packages/coding-agent/src/sdk/bus/telegram-daemon.ts:syncTelegramDirectory": "d056e2d84b39bd98a2c0b5a6f22ab0bb13adcd909c092b2f0636382dca488468",