fix(notifications): re-arm expired Telegram topic leases for owned trusted attachments (#4647)

packages/coding-agent/CHANGELOG.md

@@ -3,6 +3,7 @@
 ## [Unreleased]
 - Fixed resume listing scaling its read-syscall count with total transcript bytes. The trailing `header_patch` scan walks back to BOF whenever `cwd`/`title` stay unresolved (#3633), which is the common case because only `/rename` and workspace moves ever emit a patch; because the scan borrowed the caller's 4 KiB prefix buffer, that walk cost one `read` per 4 KiB of every candidate transcript on each `--resume`, `--continue`, and picker open. The scan now owns a 64 KiB buffer, so the same bytes are covered in ~16x fewer syscalls. Measured on a real 31-session workspace holding 105 MB of transcripts (largest 41 MB): 25,715 reads / 61.9 s before, 1,652 reads / 0.5 s after, with all 24 recovered titles unchanged. Buried-title recovery, the bytes examined, the `header_patch` marker prefilter, and listing results are unchanged.
 - `gjc team` worker auto-checkpoints no longer commit and merge root-level worker runtime state (`.gjc/state/**`, e.g. SDK broker endpoints like `.gjc/state/sdk/<session-id>.json` and settings migration markers) into the leader repo's default branch. The checkpoint classifier's protected prefixes now cover both GJC runtime roots — `.gjc/_session-*/` and `.gjc/state/` — while user-owned `.gjc/` content (config, agents, skills) stays eligible as reviewable worker work. The worker-runtime-state e2e guard now asserts absence at the actual leader merge-target path instead of an unrelated session-scoped path, and matcher boundary cases (`.gjc/state` bare entry vs `.gjc/state-*` siblings) are pinned (#4603).
+- Fixed Telegram forum topics freezing after the identity header: an attached, trusted session whose topic-host lease expired (20 s `HEARTBEAT_TTL_MS`) could never renew it, because `renewActiveTopicLeases` only renewed sessions that already passed the trusted-lease gate, so every later `turn_stream`/`context_update`/tool frame was rejected pre-send with "trusted attachment lease is stale" and the topic never updated again (#4647). A live attachment that still owns its exact logical session and holds an authorized recovery lease may now re-arm its own expired host lease — from the ownership heartbeat and once more before the publication gate — mirroring `acquireLease` admission (expired-but-owned active lease, or a same-owner resume inside the disconnect-grace window, which also covers the incident's persisted `disconnect_grace` record). Dropped sessions, closed endpoints, foreign lease owners, archive-fenced/inactive topics, malformed bindings, and cross-session ownership checks all still fail closed. Daemon generation bumped 169→170.
 - Discovered oMLX models now keep thinking metadata (`reasoning: true`, `supportsReasoningEffort`, `thinkingFormat: qwen-chat-template`) so `macos-omlx-*` role suffixes (`:low`/`:medium`/`:high`) survive clamp and reach oMLX as `chat_template_kwargs.reasoning_effort`.
 - Added built-in `MACOS LOCAL (OMLX)` model profiles (`macos-omlx-fast`, `macos-omlx-balanced`, `macos-omlx-quality`, `macos-omlx-abliterated-fast`, `macos-omlx-abliterated-balanced`) for oMLX local inference on Apple Silicon Macs with native full context support and single-LLM thinking effort role mappings to eliminate model swap latency.
 - Fixed an HTTP 400 that killed every deep-interview session on the `google-antigravity` provider before the first assistant turn. The Round-0 topology `ask` schema pinned `round` with `z.literal(0)`, which zod serializes as `const: 0` and the Cloud Code Assist normalizer rewrites to a numeric `enum: [0]` — a shape CCA rejects (`TYPE_STRING`). `round` is now pinned with an integer range `[0, 0]` instead, so the wire schema carries `type: integer` with the bounds spilled into the description (the same treatment `ambiguity` already gets) and no numeric enum remains. Runtime contract unchanged: only `0` validates (#4606).

packages/coding-agent/src/sdk/bus/telegram-daemon-contract.ts

@@ -274,7 +274,7 @@ export const SDK_LIFECYCLE_ROUTER_PROTOCOL_VERSION = 1;
  * Generation 169 delivers every ring-positioned session event live through the
  * bounded, capability-gated directed subscriber leg used by replay.
  */
-export const DAEMON_GENERATION = 169;
+export const DAEMON_GENERATION = 170;
 
 /**
  * Serving-compatibility boundary for daemon lifecycle requests. Epoch 7

packages/coding-agent/src/sdk/bus/telegram-daemon.ts

@@ -6841,7 +6841,11 @@ export class TelegramNotificationDaemon {
 		if (timer !== undefined) (this.opts.clearTimeoutImpl ?? clearTimeout)(timer);
 		this.#rejectedTopicCleanupTimers.delete(sessionId);
 	}
-	#leaseAllows(session: AttachmentSession, logicalSessionId = this.#logicalSessionId(session)): boolean {
+	#leaseAllows(
+		session: AttachmentSession,
+		logicalSessionId = this.#logicalSessionId(session),
+		options?: { allowLeaseRearm?: boolean },
+	): boolean {
 		if (this.#droppedSessions.has(session)) return false;
 		if (!this.#topicAdmissionAllows(session)) return false;
 		const closedBinding = this.closedEndpointKeys.get(logicalSessionId);
@@ -6853,13 +6857,22 @@ export class TelegramNotificationDaemon {
 		if (lease?.state !== "authorized" || lease.logicalSessionId !== logicalSessionId) return false;
 		if (this.#logicalSessionOwners.get(logicalSessionId) !== session) return false;
 		const record = this.topics.get(logicalSessionId);
+		// Lease re-arm mirrors `acquireLease` admission: an expired-but-still-owned
+		// active lease, or a same-owner grace-window resume. Every other authority
+		// state (archive/inactive/quarantine/malformed) stays fenced.
+		const authorityResumable =
+			options?.allowLeaseRearm !== true
+				? record?.authorityState === "active"
+				: record?.authorityState === "active" ||
+					(record?.authorityState === "disconnect_grace" &&
+						(record.disconnectGraceExpiresAt ?? 0) > this.runtime.now());
 		return (
 			lease.binding.logicalSessionId === logicalSessionId &&
 			(!record ||
-				(record.authorityState === "active" &&
+				(authorityResumable &&
 					!record.bindingMalformed &&
 					record.leaseOwner === this.installationHostId &&
-					(record.leaseExpiresAt ?? 0) > this.runtime.now() &&
+					(options?.allowLeaseRearm === true || (record.leaseExpiresAt ?? 0) > this.runtime.now()) &&
 					record.chatId === lease.binding.chatId))
 		);
 	}
@@ -9807,7 +9820,10 @@ export class TelegramNotificationDaemon {
 			)
 				continue;
 			const sessionId = this.#logicalSessionId(session);
-			if (renewed.has(sessionId) || !this.#leaseAllows(session, sessionId)) continue;
+			// A live trusted attachment may re-arm its own expired host lease: the
+			// authority above (owner, binding, state) is what makes renewal safe, and
+			// gating renewal on an unexpired lease made expiry permanent (#4647).
+			if (renewed.has(sessionId) || !this.#leaseAllows(session, sessionId, { allowLeaseRearm: true })) continue;
 			renewed.add(sessionId);
 			if (!(await this.#renewTopicLease(sessionId, session)))
 				logger.warn(`notifications: Telegram topic lease renewal was not admitted for session ${sessionId}`);
@@ -10280,8 +10296,15 @@ export class TelegramNotificationDaemon {
 		if (msg?.type === "event_replay_result") return;
 		if (!this.#topicAdmissionAllows(session)) return;
 		if (msg && typeof msg === "object") await this.#updateLogicalSessionForThreadedFrame(session, msg);
-		if (session.logicalSessionIdTrusted && !this.#leaseAllows(session))
-			await this.#failPublicationPreSend(publicationId, "trusted attachment lease is stale");
+		if (session.logicalSessionIdTrusted && !this.#leaseAllows(session)) {
+			// A still-owned live attachment re-arms its own expired host lease before
+			// the gate so slow turns do not permanently strand the topic (#4647);
+			// every other authority failure still fails closed below.
+			if (this.#leaseAllows(session, undefined, { allowLeaseRearm: true }))
+				await this.#renewTopicLease(this.#logicalSessionId(session), session);
+			if (!this.#leaseAllows(session))
+				await this.#failPublicationPreSend(publicationId, "trusted attachment lease is stale");
+		}
 		session.activePublicationId = publicationId;
 		try {
 			if (await this.#frameRouter.dispatch(session, msg as Record<string, unknown>)) return;

packages/coding-agent/test/notifications-telegram-topic-lease-renewal.test.ts

@@ -0,0 +1,299 @@
+import { afterEach, describe, expect, test } from "bun:test";
+import * as fs from "node:fs";
+import * as os from "node:os";
+import * as path from "node:path";
+import { Settings } from "../src/config/settings";
+import { daemonPaths } from "../src/sdk/bus/daemon-paths";
+import { type BotApi, HEARTBEAT_TTL_MS, TelegramNotificationDaemon } from "../src/sdk/bus/telegram-daemon";
+import type { NotificationSubscription } from "../src/sdk/router";
+
+const BOT_TOKEN = "1234567890:ABCDEFghijkLmnOpQrsTuvWxYz012345678";
+const HOST_ID = "lease-host";
+const SESSION_ID = "lease-session";
+
+/** The persisted fields the lease-renewal contract asserts over. */
+type PersistedTopic = {
+	identitySent?: boolean;
+	authorityState?: string;
+	leaseOwner?: string;
+	leaseExpiresAt?: number;
+	orphanedAt?: number;
+	disconnectGraceExpiresAt?: number;
+	archiveReason?: string;
+};
+
+type PersistedTopics = { topics: Record<string, PersistedTopic> };
+
+/** A recorded Bot API call: the method name plus the request body we sent. */
+type RecordedCall = { method: string; body: Record<string, unknown> };
+
+function tempAgentDir(): string {
+	return fs.mkdtempSync(path.join(os.tmpdir(), "gjc-telegram-lease-4647-"));
+}
+
+function settings(agentDir: string): Settings {
+	const isolated = Settings.isolated({
+		"notifications.enabled": true,
+		"notifications.telegram.enabled": true,
+		"notifications.telegram.botToken": BOT_TOKEN,
+		"notifications.telegram.chatId": "42",
+	}) as Settings;
+	return new Proxy(isolated, {
+		get(target, property) {
+			if (property === "getAgentDir") return () => agentDir;
+			const value = Reflect.get(target, property, target);
+			return typeof value === "function" ? value.bind(target) : value;
+		},
+	}) as Settings;
+}
+
+function notificationSubscription(sessionId: string, generation = 1): NotificationSubscription {
+	let active = true;
+	const cursor = { generation, seq: 0 };
+	return {
+		sessionId,
+		subscriptionId: `test:${sessionId}:${generation}`,
+		cursor,
+		isActive: () => active,
+		send: () => undefined,
+		advanceCursor: (nextGeneration, seq) => {
+			cursor.generation = nextGeneration;
+			cursor.seq = seq;
+		},
+		cancel: () => {
+			active = false;
+		},
+	};
+}
+
+/** Bot that creates topic 555 once and records every Bot API call. */
+function fakeBot(): { bot: BotApi; calls: RecordedCall[] } {
+	const calls: RecordedCall[] = [];
+	const bot: BotApi = {
+		call: async (method, body) => {
+			calls.push({ method, body: (body ?? {}) as Record<string, unknown> });
+			if (method === "createForumTopic") return { ok: true, result: { message_thread_id: 555 } };
+			if (method === "getChat") return { ok: true, result: { id: 42, type: "private" } };
+			if (method === "sendMessage") return { ok: true, result: { message_id: calls.length } };
+			return { ok: true, result: true };
+		},
+	};
+	return { bot, calls };
+}
+
+async function readPersistedTopics(agentDir: string): Promise<PersistedTopics> {
+	return (await Bun.file(path.join(daemonPaths(agentDir).dir, "telegram-topics.json")).json()) as PersistedTopics;
+}
+
+async function persistedTopic(agentDir: string, sessionId: string): Promise<PersistedTopic> {
+	const topics = await readPersistedTopics(agentDir);
+	const topic = topics.topics[sessionId];
+	if (!topic) throw new Error(`Expected a persisted topic record for ${sessionId}.`);
+	return topic;
+}
+
+/** Rewrite the durable topic registry through the Bun write API. */
+async function mutatePersistedTopics(agentDir: string, mutate: (topics: PersistedTopics) => void): Promise<void> {
+	const topics = await readPersistedTopics(agentDir);
+	mutate(topics);
+	await Bun.write(path.join(daemonPaths(agentDir).dir, "telegram-topics.json"), `${JSON.stringify(topics)}\n`);
+}
+
+type LeaseHarness = {
+	bot: ReturnType<typeof fakeBot>;
+	daemon: TelegramNotificationDaemon;
+	sessionId: string;
+	send: (frame: Record<string, unknown>, publicationId: string) => Promise<unknown>;
+};
+
+const agentDirs: string[] = [];
+afterEach(() => {
+	for (const dir of agentDirs.splice(0)) fs.rmSync(dir, { recursive: true, force: true });
+});
+
+/**
+ * Attach + authenticated replay so the session is `logicalSessionIdTrusted` with an
+ * authorized recovery lease bound to its durable topic — the incident state from
+ * issue #4647. `nowRef` is captured so a test can advance clock time afterwards.
+ */
+async function trustedAttachedSession(agentDir: string, nowRef: { now: number }): Promise<LeaseHarness> {
+	const bot = fakeBot();
+	const daemon = new TelegramNotificationDaemon({
+		settings: settings(agentDir),
+		ownerId: HOST_ID,
+		botToken: BOT_TOKEN,
+		chatId: "42",
+		botApi: bot.bot,
+		now: () => nowRef.now,
+		installationHostId: HOST_ID,
+	});
+	await daemon.loadTopics();
+	const routing = daemon.attachmentRoutingHarnessForTest();
+	const attachment = notificationSubscription(SESSION_ID);
+	routing.attach(attachment);
+	const session = daemon.sessions.get(attachment.sessionId);
+	if (!session) throw new Error("Expected a routed Telegram attachment session.");
+	await daemon.handleSessionMessage(session, {
+		type: "event_replay_result",
+		id: session.replayId,
+		ok: true,
+		generation: 1,
+		lastSeq: 0,
+		events: [{ payload: { type: "identity_header", sessionId: SESSION_ID, telegramTopicsEnabled: true } }],
+	});
+	return {
+		bot,
+		daemon,
+		sessionId: SESSION_ID,
+		send: (frame, publicationId) =>
+			daemon.handleSessionMessage(session, { sessionId: SESSION_ID, ...frame }, publicationId),
+	};
+}
+
+describe("Telegram trusted attachment topic-host lease renewal (#4647)", () => {
+	test("a live owned trusted attachment publishes after its host lease expired", async () => {
+		const agentDir = tempAgentDir();
+		agentDirs.push(agentDir);
+		const nowRef = { now: 1_000 };
+		const harness = await trustedAttachedSession(agentDir, nowRef);
+		try {
+			// Identity was replayed; the topic exists and the identity header was sent.
+			expect(harness.bot.calls.some(call => call.method === "createForumTopic")).toBe(true);
+			expect((await persistedTopic(agentDir, SESSION_ID)).identitySent).toBe(true);
+
+			// Advance past the 20s host-lease TTL while the attachment stays live and
+			// owned. Before the fix every later frame was rejected pre-send.
+			nowRef.now += HEARTBEAT_TTL_MS + 1_000;
+			await harness.send(
+				{ type: "turn_stream", phase: "finalized", text: "post-identity update" },
+				"lease-session:1:3",
+			);
+
+			const sent = harness.bot.calls.filter(call => call.method === "sendMessage");
+			expect(sent.length).toBeGreaterThan(0);
+			expect(sent.at(-1)?.body.message_thread_id).toBe(555);
+			const topic = await persistedTopic(agentDir, SESSION_ID);
+			expect(topic.authorityState).toBe("active");
+			expect((topic.leaseExpiresAt ?? 0) > nowRef.now).toBe(true);
+		} finally {
+			harness.daemon.requestStop();
+		}
+	});
+
+	test("ownership heartbeat re-arms an expired owned lease without a publication", async () => {
+		const agentDir = tempAgentDir();
+		agentDirs.push(agentDir);
+		const nowRef = { now: 1_000 };
+		const harness = await trustedAttachedSession(agentDir, nowRef);
+		try {
+			nowRef.now += HEARTBEAT_TTL_MS + 1_000;
+			expect(((await persistedTopic(agentDir, SESSION_ID)).leaseExpiresAt ?? 0) <= nowRef.now).toBe(true);
+
+			await (harness.daemon as unknown as { renewActiveTopicLeases(): Promise<void> }).renewActiveTopicLeases();
+
+			const topic = await persistedTopic(agentDir, SESSION_ID);
+			expect(topic.authorityState).toBe("active");
+			expect((topic.leaseExpiresAt ?? 0) > nowRef.now).toBe(true);
+		} finally {
+			harness.daemon.requestStop();
+		}
+	});
+
+	test("an expired lease owned by a foreign host is never re-armed", async () => {
+		const agentDir = tempAgentDir();
+		agentDirs.push(agentDir);
+		const nowRef = { now: 1_000 };
+		const harness = await trustedAttachedSession(agentDir, nowRef);
+		try {
+			await mutatePersistedTopics(agentDir, topics => {
+				topics.topics[SESSION_ID]!.leaseOwner = "another-installation";
+			});
+			// Reload so the registry (and the lease gate) observe the foreign owner.
+			await (harness.daemon as unknown as { loadTopics(): Promise<void> }).loadTopics();
+			nowRef.now += HEARTBEAT_TTL_MS + 1_000;
+
+			await expect(
+				harness.send({ type: "turn_stream", phase: "finalized", text: "no" }, "lease-session:1:9"),
+			).rejects.toThrow("trusted attachment lease is stale");
+			const topic = await persistedTopic(agentDir, SESSION_ID);
+			expect(topic.leaseOwner).toBe("another-installation");
+			expect((topic.leaseExpiresAt ?? 0) <= nowRef.now).toBe(true);
+		} finally {
+			harness.daemon.requestStop();
+		}
+	});
+
+	test("a non-owner trusted attachment still fails closed after TTL expiry", async () => {
+		const agentDir = tempAgentDir();
+		agentDirs.push(agentDir);
+		const nowRef = { now: 1_000 };
+		const harness = await trustedAttachedSession(agentDir, nowRef);
+		try {
+			// A successor attachment with the same session id replaces the transport;
+			// the predecessor handle must never re-arm a lease it no longer owns.
+			const successor = notificationSubscription(SESSION_ID, 2);
+			harness.daemon.attachmentRoutingHarnessForTest().attach(successor);
+			nowRef.now += HEARTBEAT_TTL_MS + 1_000;
+
+			await expect(
+				harness.send({ type: "turn_stream", phase: "finalized", text: "stale socket" }, "lease-session:1:7"),
+			).rejects.toThrow();
+
+			const topic = await persistedTopic(agentDir, SESSION_ID);
+			expect((topic.leaseExpiresAt ?? 0) <= nowRef.now).toBe(true);
+		} finally {
+			harness.daemon.requestStop();
+		}
+	});
+
+	test("a grace-window topic is resumed, not rejected, by its still-attached owner", async () => {
+		const agentDir = tempAgentDir();
+		agentDirs.push(agentDir);
+		const nowRef = { now: 1_000 };
+		const harness = await trustedAttachedSession(agentDir, nowRef);
+		try {
+			// The reporter's incident record: released to grace inside the window.
+			await mutatePersistedTopics(agentDir, topics => {
+				const topic = topics.topics[SESSION_ID]!;
+				topic.authorityState = "disconnect_grace";
+				topic.orphanedAt = nowRef.now;
+				topic.leaseExpiresAt = nowRef.now;
+				topic.disconnectGraceExpiresAt = nowRef.now + 60_000;
+			});
+			await (harness.daemon as unknown as { loadTopics(): Promise<void> }).loadTopics();
+			nowRef.now += HEARTBEAT_TTL_MS + 1_000;
+
+			await harness.send({ type: "turn_stream", phase: "finalized", text: "resume" }, "lease-session:1:5");
+
+			const topic = await persistedTopic(agentDir, SESSION_ID);
+			// The live owner resumed the exact topic: active again with a fresh lease.
+			expect(topic.authorityState).toBe("active");
+			expect(topic.orphanedAt).toBeUndefined();
+			expect((topic.leaseExpiresAt ?? 0) > nowRef.now).toBe(true);
+		} finally {
+			harness.daemon.requestStop();
+		}
+	});
+
+	test("an archive-fenced topic is never re-armed by a live attachment", async () => {
+		const agentDir = tempAgentDir();
+		agentDirs.push(agentDir);
+		const nowRef = { now: 1_000 };
+		const harness = await trustedAttachedSession(agentDir, nowRef);
+		try {
+			await mutatePersistedTopics(agentDir, topics => {
+				topics.topics[SESSION_ID]!.authorityState = "archive_pending";
+				topics.topics[SESSION_ID]!.archiveReason = "session_closed";
+			});
+			await (harness.daemon as unknown as { loadTopics(): Promise<void> }).loadTopics();
+			nowRef.now += HEARTBEAT_TTL_MS + 1_000;
+
+			await expect(
+				harness.send({ type: "turn_stream", phase: "finalized", text: "no" }, "lease-session:1:11"),
+			).rejects.toThrow("trusted attachment lease is stale");
+			expect((await persistedTopic(agentDir, SESSION_ID)).authorityState).toBe("archive_pending");
+		} finally {
+			harness.daemon.requestStop();
+		}
+	});
+});