From a62cd9a210633f62d381d14aad3006459a6916dc Mon Sep 17 00:00:00 2001 From: yazzang-homelab Date: Thu, 6 Aug 2026 10:57:21 +0900 Subject: [PATCH] fix(notifications): clear the disconnect-grace deadline when a topic is archived MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `parseTopicRegistryState` rejects a record that carries `disconnectGraceExpiresAt` in any state other than `disconnect_grace`, but only the grace-to-active transitions deleted it. Every archive transition (`beginArchive`, `restoreArchiveFence`, `scheduleArchiveRetry`, the exhausted-to-pending retry, `settleArchive`, and the restore path that retires a rebound topic) left it in place, so the daemon published a snapshot its own parser refuses. The failure is permanent, not transient: after the poisoned publish, every `loadTopics` and every `compareAndSet` throws `shared topic authority unavailable`, which on a shared-authority daemon exits the process. On a live installation three consecutive daemons died that way within three hours, and the orphaned topic stayed in `archive_pending` for hours with no owner to archive it or answer in it. Both halves are needed: transitions out of grace now go through one setter that drops the deadline, and a registry that already carries one loads with it normalized away instead of failing every publish forever. The reverse inconsistency — a `disconnect_grace` record missing its own deadline or orphan observation — is genuinely ambiguous and stays fatal. Lore-id: 7e41c0b8 Constraint: a settled authority state must round-trip through its own parser Rejected: relax the parser for both directions | a grace record without a deadline is ambiguous and must stay fatal Rejected: repair on write only | already-poisoned installations would never start a daemon again Confidence: high Scope-risk: narrow Reversibility: easy Tested: grace to archive_pending to inactive round-trips through parseTopicRegistryState; the real poisoned on-disk record loads normalized; a grace record without its deadline still throws --- packages/coding-agent/CHANGELOG.md | 1 + .../src/sdk/bus/topic-registry.ts | 52 ++++++++---- .../test/notifications-topic-registry.test.ts | 80 +++++++++++++++++++ 3 files changed, 117 insertions(+), 16 deletions(-) diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 941674524f..78359d93e0 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -18,6 +18,7 @@ - Made Telegram reference-client capability diagnostics safe for TUI embedding. - Custom `anthropic-messages` providers can now configure `compat.promptCacheMode` (`none`, `explicit`, or `automatic`) and `compat.supportsLongCacheRetention` at provider, model, and model-override levels. Canonical Anthropic defaults to automatic caching, while non-canonical Claude-family endpoints default to gateway-safe explicit block markers and can opt into top-level automatic caching when supported. - A Telegram notification daemon whose reconciliation pass fails no longer exits. The pass persists through the shared topic authority, and a momentarily unavailable authority (lock contention or a rejected compare-and-set) rejected out of both the scan timer and the run loop into the process-level fatal handler, killing the owner. Every session topic was then left behind as an unarchived shell that answers nothing — including for sessions that were still live and lost their notifications. The pass now reports the failure and the next scan interval retries it; the queue-flush timer is guarded the same way. +- Archiving a Telegram topic no longer poisons the shared topic registry. Every transition out of `disconnect_grace` kept the record's `disconnectGraceExpiresAt`, which `parseTopicRegistryState` rejects in any settled authority state, so the daemon published a snapshot it could not read back: the next load and every compare-and-set failed with `shared topic authority unavailable`, and the orphaned topic was never archived. Transitions out of grace now clear that deadline, and a registry already carrying one loads with it dropped instead of failing every publish. - MiniMax M3 preset and profile ids canonicalized to `MiniMax-M3` (issue #3896): the `minimax` / `minimax-cn` onboarding presets and the `minimax-eco` / `minimax-medium` / `minimax-pro` builtin model profiles no longer reference the removed lowercase `minimax-m3` / `minimax-v3` first-class catalog ids. - Deep-interview round identity and input caps are now Unicode-canonical. Question text, selected options, and custom input are canonicalized to NFC before hashing and persisting, so the same Korean answer arriving in decomposed form (macOS-sourced pastes and some IME/clipboard paths emit NFD) no longer produces a second `answer_hash` — the documented append-or-merge no-op holds, and intent-review approval evidence still matches the user's recorded answer. Free-text caps are measured on the NFC form, so decomposed Hangul is charged the same character budget as the identical composed text instead of 2–3 code points per syllable (#3871). - Telegram notifications no longer disappear in a paired private chat whose bot has no Threaded Mode. Telegram answers `createForumTopic` there with `Bad Request: the chat is not a forum`, which was not recognized as a capability refusal, and the refusal verdict lived in caller-local flags — so every frame that joined the shared in-flight topic creation rethrew and its message (identity headers after `/resume`, asks, context updates) was dropped instead of being delivered flat. The rejection is now carried by typed errors that every awaiter of the same creation classifies identically, `the chat is not a forum` counts as a capability refusal, and a confirmed refusal is latched so later frames stop re-issuing a rejected `createForumTopic` per message. diff --git a/packages/coding-agent/src/sdk/bus/topic-registry.ts b/packages/coding-agent/src/sdk/bus/topic-registry.ts index a858203897..a670dffbbc 100644 --- a/packages/coding-agent/src/sdk/bus/topic-registry.ts +++ b/packages/coding-agent/src/sdk/bus/topic-registry.ts @@ -303,10 +303,16 @@ export function parseTopicRegistryState(value: unknown): TopicRegistryState | un value => value !== undefined, ).length; if (leaseFieldCount !== 0 && leaseFieldCount !== 3) malformed(); + // A `disconnect_grace` record without its deadline or orphan observation is + // genuinely ambiguous and stays fatal. The reverse — a grace deadline left + // behind in a settled authority state — is not ambiguous: the authority + // state is explicit and the stale deadline is inert. Releases before this + // fix persisted exactly that on every archive transition, so rejecting it + // permanently bricks the daemon on its own snapshot. Normalize it away in + // `load` instead. if ( - raw.authorityState === "disconnect_grace" - ? raw.disconnectGraceExpiresAt === undefined || raw.orphanedAt === undefined - : raw.disconnectGraceExpiresAt !== undefined + raw.authorityState === "disconnect_grace" && + (raw.disconnectGraceExpiresAt === undefined || raw.orphanedAt === undefined) ) malformed(); const hasBinding = hasAnyBinding(raw); @@ -583,6 +589,10 @@ export class TopicRegistry { : {}), ...(bindingMalformed ? { bindingMalformed: true as const } : {}), }; + // Drop a grace deadline that a pre-fix release left behind on a settled + // authority state, so the normalized snapshot round-trips through + // `parseTopicRegistryState` instead of failing every later publish. + if (record.authorityState !== "disconnect_grace") delete record.disconnectGraceExpiresAt; this.epochs.set(sessionId, Math.max(fenceEpoch, record.authorityEpoch ?? 0)); // Pre-generation-17 records have no endpoint authority. Retire them locally: // their unknown remote topic must neither be rebound nor deleted cross-chat. @@ -829,9 +839,8 @@ export class TopicRegistry { record.endpointGeneration = binding.endpointGeneration; if (!sameEndpoint) record.endpointIncarnation = (record.endpointIncarnation ?? 0) + 1; if (record.authorityState === "disconnect_grace") { - record.authorityState = "active"; + this.#setAuthorityState(record, "active"); delete record.orphanedAt; - delete record.disconnectGraceExpiresAt; this.rebuildInboundRoutes(); } return "bound"; @@ -945,7 +954,7 @@ export class TopicRegistry { this.staged.delete(sessionId); if ((this.epochs.get(sessionId) ?? 0) !== epoch) { record.authorityEpoch = this.epochs.get(sessionId) ?? 0; - record.authorityState = "archive_pending"; + this.#setAuthorityState(record, "archive_pending"); this.topics.set(sessionId, record); throw new Error("topic authority was revoked during creation"); } @@ -1017,9 +1026,8 @@ export class TopicRegistry { record.leaseHeartbeatAt = now; record.leaseExpiresAt = now + ttlMs; if (record.authorityState === "disconnect_grace") { - record.authorityState = "active"; + this.#setAuthorityState(record, "active"); delete record.orphanedAt; - delete record.disconnectGraceExpiresAt; this.rebuildInboundRoutes(); } return true; @@ -1066,12 +1074,24 @@ export class TopicRegistry { clearOrphaned(sessionId: string): boolean { const record = this.topics.get(sessionId); if (record?.authorityState !== "disconnect_grace" || record.orphanedAt === undefined) return false; - record.authorityState = "active"; - delete record.disconnectGraceExpiresAt; + this.#setAuthorityState(record, "active"); delete record.orphanedAt; return true; } + /** + * `disconnectGraceExpiresAt` is grace-state metadata, and + * `parseTopicRegistryState` rejects a record that still carries it in any + * other authority state. Every transition out of `disconnect_grace` must + * therefore drop it, or the daemon persists a snapshot it can no longer read + * back: the next load and every compare-and-set fail, and a shared-authority + * daemon exits instead of archiving the topic. + */ + #setAuthorityState(record: TopicRecord, state: NonNullable): void { + record.authorityState = state; + if (state !== "disconnect_grace") delete record.disconnectGraceExpiresAt; + } + /** Last durably consumed SDK event cursor for reconnect replay. */ replayCursor(sessionId: string): { generation: number; seq: number } | undefined { const record = this.topics.get(sessionId); @@ -1206,7 +1226,7 @@ export class TopicRegistry { return false; } else { record.authorityEpoch = deleteEpoch; - record.authorityState = "archive_pending"; + this.#setAuthorityState(record, "archive_pending"); if (this.byTopic.get(record.topicId) === snapshot.sessionId) this.byTopic.delete(record.topicId); } this.epochs.set(snapshot.sessionId, deleteEpoch); @@ -1229,7 +1249,7 @@ export class TopicRegistry { this.epochs.set(sessionId, Number.MAX_SAFE_INTEGER); if (record) { record.authorityEpoch = Number.MAX_SAFE_INTEGER; - record.authorityState = "archive_exhausted"; + this.#setAuthorityState(record, "archive_exhausted"); if (this.byTopic.get(record.topicId) === sessionId) this.byTopic.delete(record.topicId); } return undefined; @@ -1238,7 +1258,7 @@ export class TopicRegistry { this.epochs.set(sessionId, epoch); if (!record) return undefined; record.authorityEpoch = epoch; - record.authorityState = "archive_pending"; + this.#setAuthorityState(record, "archive_pending"); if (hostId) record.archiveHostId = hostId; record.archiveLeaseEpoch = epoch; if (this.byTopic.get(record.topicId) === sessionId) this.byTopic.delete(record.topicId); @@ -1367,7 +1387,7 @@ export class TopicRegistry { this.authorityEpoch(sessionId) !== dispatchedAuthorityEpoch ) return false; - record.authorityState = "inactive"; + this.#setAuthorityState(record, "inactive"); if (this.byTopic.get(record.topicId) === sessionId) this.byTopic.delete(record.topicId); this.archiveJobs.delete(sessionId); return true; @@ -1382,7 +1402,7 @@ export class TopicRegistry { const epoch = Math.min(currentEpoch + 1, Number.MAX_SAFE_INTEGER - 1); record.authorityEpoch = epoch; record.archiveLeaseEpoch = epoch; - record.authorityState = "archive_pending"; + this.#setAuthorityState(record, "archive_pending"); } } return (record.authorityState === "archive_pending" || record.authorityState === "archive_exhausted") && @@ -1419,7 +1439,7 @@ export class TopicRegistry { ...(diagnostic ? { safeDiagnostic: diagnostic.slice(0, 256) } : {}), ...(exhausted ? { safeDiagnostic: "archive retry remains discoverable after retry budget" } : {}), }; - record.authorityState = "archive_pending"; + this.#setAuthorityState(record, "archive_pending"); this.archiveJobs.set(sessionId, job); return job; } diff --git a/packages/coding-agent/test/notifications-topic-registry.test.ts b/packages/coding-agent/test/notifications-topic-registry.test.ts index 40513461aa..81f35825f6 100644 --- a/packages/coding-agent/test/notifications-topic-registry.test.ts +++ b/packages/coding-agent/test/notifications-topic-registry.test.ts @@ -818,3 +818,83 @@ test("retains inactive predecessor evidence when an authenticated successor rota ]); expect(new TopicRegistry(serialized).serialize().retiredTopics).toEqual(serialized.retiredTopics); }); + +test("archiving a disconnect-grace topic publishes a snapshot the daemon can read back", async () => { + // A grace deadline left on a settled authority state is rejected by + // parseTopicRegistryState, so persisting it makes every later load and + // compare-and-set fail and takes the shared-authority daemon down with it. + const registry = new TopicRegistry(); + await registry.getOrCreateTopic("session", async () => "700"); + expect(registry.acquireLease("session", "host-a", 100, 1_000, 60_000)).toBe(true); + expect(registry.releaseLeaseToGrace("session", "host-a", 200, 60_000)).toBe(true); + const grace = registry.serialize(); + expect(grace.topics.session).toMatchObject({ authorityState: "disconnect_grace", disconnectGraceExpiresAt: 60_200 }); + expect(parseTopicRegistryState(grace)).toBeDefined(); + + expect(registry.beginArchive("session", "host-a", 300_000)?.authorityState).toBe("archive_pending"); + const archiving = registry.serialize(); + expect(archiving.topics.session?.authorityState).toBe("archive_pending"); + expect(archiving.topics.session?.disconnectGraceExpiresAt).toBeUndefined(); + expect(parseTopicRegistryState(archiving)).toBeDefined(); + + expect(registry.settleArchive("session", "700", archiving.topics.session?.authorityEpoch ?? -1)).toBe(true); + const settled = registry.serialize(); + expect(settled.topics.session?.disconnectGraceExpiresAt).toBeUndefined(); + expect(parseTopicRegistryState(settled)).toBeDefined(); +}); + +test("a grace deadline persisted by an older release loads instead of bricking the registry", () => { + // Reproduces a real on-disk snapshot: the archive transition kept the grace + // deadline, so the owning daemon could no longer parse its own registry. + const poisoned = { + version: 2 as const, + registryGeneration: 451, + topics: { + session: { + topicId: "14258", + topicOrigin: "daemon_created" as const, + sessionUuid: "1f5b3af9-5b97-4a78-b60e-1ef69700b36f", + identitySent: true, + createdAt: 1_785_971_484_922, + orphanedAt: 1_785_971_777_162, + authorityEpoch: 1, + authorityState: "archive_pending" as const, + chatId: "7731731210", + endpointKey: "ae6b6ed0e9f8e3d857f3d4d2c33f2b58f", + endpointDigest: "ae6b6ed0e9f8e3d857f3d4d2c33f2b58f", + endpointGeneration: 1, + endpointIncarnation: 0, + disconnectGraceExpiresAt: 1_785_971_807_162, + archiveHostId: "6080b838b93d01cd76fa4e12a9c2e67f7", + archiveLeaseEpoch: 1, + }, + }, + }; + const parsed = parseTopicRegistryState(poisoned); + expect(parsed).toBeDefined(); + const normalized = new TopicRegistry(parsed!).serialize(); + expect(normalized.topics.session?.authorityState).toBe("archive_pending"); + expect(normalized.topics.session?.disconnectGraceExpiresAt).toBeUndefined(); + // The normalized snapshot must republish, which is what a compare-and-set does. + expect(parseTopicRegistryState(normalized)).toBeDefined(); +}); + +test("a disconnect-grace record without its own deadline stays fatal", () => { + expect(() => + parseTopicRegistryState({ + version: 2 as const, + registryGeneration: 1, + topics: { + session: { + topicId: "700", + topicOrigin: "daemon_created" as const, + sessionUuid: "s", + identitySent: true, + createdAt: 1, + orphanedAt: 2, + authorityState: "disconnect_grace" as const, + }, + }, + }), + ).toThrow("malformed Telegram topic state"); +});