fix(notifications): clear the disconnect-grace deadline when a topic is archived

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.

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<TopicRecord["authorityState"]>): 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;
 	}

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");
+});