fix(sdk): run initial attachment replay off the reconcile tail (#4527)

packages/coding-agent/CHANGELOG.md

@@ -23,6 +23,7 @@
 ### Fixed
 - Default-model selection now reserves a causal fence before credential probing, so an already accepted prompt preflight cannot be overtaken and a selection accepted first blocks later prompt preflight through durable publication. The fence does not hold session admission across `waitForIdle`, allowing inherited auto-compaction continuations to obtain prompt admission; same-session reentrancy still fails fast, successor sessions remain protected, and disposal deterministically drains accepted selections while rejecting queued prompts without unhandled rejections (#4519).
 - Ordinary sessions no longer import or execute Claude Code and Codex directory hooks as competing runtime authorities. Runtime hook discovery is fail-closed to canonical native `.gjc/hooks/` providers while explicit configured paths, constrained plugin hooks, and foreign-provider import/diagnostic discovery remain available (#4516).
+- Telegram/Slack/Discord outbound publications no longer freeze after a session-host rehost. A rehosted fleet re-attaches every session in one reconcile pass, and each attachment's initial `event_replay` was awaited inside the serialized `#reconcileTail`, so one slow replay (up to its full retry budget) wedged all later reconciles and the sends funneling through them; leases and inbound polling stayed green while delivery died until daemon restart. Reconcile-driven attachments now publish immediately and run initial replay on the attachment's ready tail (matching the reconnect path); `start()` still drains those tails so bootstrap callers observe replay completion. Replay ordering, generation fences, cross-session isolation, and provider hooks are unchanged (#4527).
 - `gjc_coordinator_stop_session` no longer reports `close_failed` after a successful DR-1 terminal close. Reap now proves the same retained session is `terminal` and non-`live` for the exact `workspace/generation/incarnation` before completing local cleanup; rotated generation, different incarnation, ambiguous, still-live, and `terminal_uncertain` remain fail-closed (#4431).
 - The crash signature index can no longer quarantine itself. `parseCrashIndex` rejects an entry whose retained count exceeds its lifetime count, and compaction produced exactly that whenever the crash log still held more identity-bearing records for a signature than the journal had counted — which happens whenever a fatal's journal append fails or spends the per-process latch while its crash-log record is still written. The next read then discarded the whole index, taking every signature in it. The retained count is now held at the journaled lifetime.
 - Crash records whose event-journal append was lost are reportable again. Compaction recovers only never-indexed, structurally complete v1 records whose marker fingerprint recomputes from their own diagnostic text, trying only marker-proven reconstructions for the v1 header's ambiguous colon names, multiline messages, arbitrary multiline stacks, and separate serialized object payload. Record ids are deduplicated for recovery and retained counts; recovered ids remain durable until delayed journal events arrive; whole rotated-journal digests make publish-before-delete replay idempotent beyond the 256-id window; and timestamp ordering controls first/last-seen plus latest record metadata. Reported or acknowledged signatures evicted from the bounded index stay retired while their records survive: confirmed log pruning drops obsolete tombstones, uncertain crash-log or sidecar reads fail closed, retirement evidence publishes before the main index that depends on it, and replayed journal batches heal an interrupted stale tombstone without repeating counts. The separate ledger is bounded by the maximum distinct records in the bounded log rather than the live-index entry cap, so sequential retirements cannot exhaust admission (#4478, reported with measured field evidence by @yazzang-homelab).

packages/coding-agent/src/sdk/router/session-router.ts

@@ -394,7 +394,7 @@ export class SessionRouter {
 			await this.#serialReconcile(runEpoch);
 			if (!this.#running(runEpoch)) return;
 			const timer = (this.#deps.setInterval ?? setInterval)(
-				() => this.#schedule(this.#serialReconcile(runEpoch)),
+				() => this.#schedule(this.#serialReconcile(runEpoch, true)),
 				2_000,
 			);
 			this.#stopTimer = () => (this.#deps.clearInterval ?? clearInterval)(timer);
@@ -405,8 +405,13 @@ export class SessionRouter {
 	}
 
 	/** Exposed for deterministic callers and reconciliation tests. */
-	reconcile(): Promise<void> {
-		return this.#serialReconcile(this.#runEpoch);
+	async reconcile(): Promise<void> {
+		await this.#serialReconcile(this.#runEpoch, true);
+		// Periodic reconciliation may have published an attachment while its
+		// initial replay continues on that attachment's isolated ready tail.
+		// Explicit callers retain the historical synchronous contract without
+		// putting any ready tail back onto the fleet-wide reconcile tail.
+		await Promise.all([...this.#sessions.values()].map(attached => attached.readyTail));
 	}
 	/** Ingests a credential-bearing Broker lifecycle result directly into Router custody. */
 	async adoptLifecycleResult(
@@ -471,7 +476,8 @@ export class SessionRouter {
 		const listing = this.#index.listSessions();
 		const indexedCurrent =
 			listing.warnings.length === 0 ? listing.sessions.find(item => item.sessionId === sessionId) : undefined;
-		if (indexedCurrent && sameIndexedAuthority(indexed, indexedCurrent)) await this.#serialReconcile(this.#runEpoch);
+		if (indexedCurrent && sameIndexedAuthority(indexed, indexedCurrent))
+			await this.#serialReconcile(this.#runEpoch, true);
 		return capability;
 	}
 
@@ -547,7 +553,7 @@ export class SessionRouter {
 	): Promise<Record<string, unknown>> {
 		const publishing = this.#sessions.get(sessionId);
 		if (!expectedAttachment || publishing?.capability !== expectedAttachment || !publishing.initializingPublication)
-			await this.#serialReconcile(this.#runEpoch);
+			await this.#serialReconcile(this.#runEpoch, true);
 		const attached = this.#sessions.get(sessionId);
 		if (!attached || !this.#attachmentPublished(attached))
 			throw new SessionRouterError("pre_send", "SDK session attachment is unavailable: session not published.");
@@ -711,7 +717,7 @@ export class SessionRouter {
 		}
 	}
 
-	#serialReconcile(runEpoch: number): Promise<void> {
+	#serialReconcile(runEpoch: number, deferReplay = false): Promise<void> {
 		if (!this.#running(runEpoch)) return Promise.resolve();
 		const pending = this.#reconcilePending;
 		if (pending?.runEpoch === runEpoch) return this.#reconcileTail;
@@ -722,7 +728,7 @@ export class SessionRouter {
 			.then(async () => {
 				if (this.#reconcilePending === queued) this.#reconcilePending = undefined;
 				try {
-					await this.#reconcile(runEpoch);
+					await this.#reconcile(runEpoch, deferReplay);
 					if (!this.#running(runEpoch)) return;
 					this.#ready = true;
 					this.#deps.onReconciled?.();
@@ -735,7 +741,7 @@ export class SessionRouter {
 		return task;
 	}
 
-	async #reconcile(runEpoch: number): Promise<void> {
+	async #reconcile(runEpoch: number, deferReplay = false): Promise<void> {
 		if (!this.#running(runEpoch)) return;
 		await this.#index.open();
 		if (!this.#running(runEpoch)) return;
@@ -795,7 +801,8 @@ export class SessionRouter {
 				const session = live[nextAttachment++];
 				if (!session) return;
 				try {
-					if (await this.#attach(session, runEpoch)) attachedIds.add(session.sessionId);
+					if (await this.#attach(session, runEpoch, undefined, false, false, deferReplay))
+						attachedIds.add(session.sessionId);
 				} catch {
 					const failed = this.#sessions.get(session.sessionId);
 					if (failed?.runEpoch === runEpoch)
@@ -921,6 +928,7 @@ export class SessionRouter {
 		resolvedEndpoint?: SdkSessionEndpoint,
 		skipReplay = false,
 		deferPublication = false,
+		deferReplay = false,
 	): Promise<boolean> {
 		const retirementVersion = this.#retirementVersions.get(indexed.sessionId) ?? 0;
 		const retirement = this.#retirements.get(indexed.sessionId);
@@ -931,7 +939,7 @@ export class SessionRouter {
 		const retirementAfterValidation = this.#retirements.get(indexed.sessionId);
 		if (retirementAfterValidation) await retirementAfterValidation;
 		if ((this.#retirementVersions.get(indexed.sessionId) ?? 0) !== retirementVersion)
-			return await this.#attach(indexed, runEpoch, undefined, skipReplay, deferPublication);
+			return await this.#attach(indexed, runEpoch, undefined, skipReplay, deferPublication, deferReplay);
 		if (!this.#running(runEpoch)) return false;
 		if (!endpoint) return false;
 		const existing = this.#sessions.get(indexed.sessionId);
@@ -991,7 +999,7 @@ export class SessionRouter {
 			await client.close().catch(() => undefined);
 			const currentRetirement = this.#retirements.get(indexed.sessionId);
 			if (currentRetirement) await currentRetirement;
-			return await this.#attach(indexed, runEpoch, undefined, skipReplay, deferPublication);
+			return await this.#attach(indexed, runEpoch, undefined, skipReplay, deferPublication, deferReplay);
 		}
 		let attached: AttachedSession | undefined;
 		const barrier: ReplayBarrier = { held: undefined, detached: false, failed: false };
@@ -1026,7 +1034,7 @@ export class SessionRouter {
 						await this.#retireAttachment(attached, endpoint ? "replaced_same_generation" : undefined);
 						throw new SessionRouterError("pre_send", "SDK session attachment changed during publication.");
 					}
-				} else await this.#serialReconcile(runEpoch);
+				} else await this.#serialReconcile(runEpoch, true);
 				if (!attached || !this.#attachmentPublished(attached))
 					throw new SessionRouterError("pre_send", "SDK session attachment is stale.");
 				attached.client.send(this.#prepareFrame(attached, frame));
@@ -1138,10 +1146,10 @@ export class SessionRouter {
 			throw error;
 		}
 		if (deferPublication) return true;
-		return await this.#publishAttachment(attached, skipReplay);
+		return await this.#publishAttachment(attached, skipReplay, deferReplay);
 	}
 
-	async #publishAttachment(attached: AttachedSession, skipReplay: boolean): Promise<boolean> {
+	async #publishAttachment(attached: AttachedSession, skipReplay: boolean, deferReplay = false): Promise<boolean> {
 		if (attached.published) return this.#attachmentPublished(attached);
 		if (!this.#attachmentLive(attached)) return false;
 		const endpoint = await this.#readEndpoint(attached.indexed).catch(() => null);
@@ -1185,6 +1193,23 @@ export class SessionRouter {
 			attached.initializingPublication = false;
 		}
 		if (skipReplay) return true;
+		// When the caller drives the serialized reconcile tail (periodic
+		// re-attachment after a rehost), initial replay must not hold it: each
+		// replay owns its own retry budget, so awaiting it here wedges all later
+		// reconciles (and the sends that funnel through them) until the budget
+		// expires. The barrier still holds live frames, so ordering and
+		// generation fences are unchanged; replay just runs on the attachment's
+		// ready tail like the reconnect path (#4527).
+		if (deferReplay) {
+			attached.readyTail = attached.readyTail
+				.catch(() => undefined)
+				.then(async () => {
+					if (!this.#attachmentLive(attached)) return;
+					if (!(await this.#deliverRecoveredFrames(attached))) return;
+					await this.#replayAttachment(attached, attached.cursor.seq);
+				});
+			return true;
+		}
 		if (!(await this.#deliverRecoveredFrames(attached))) return false;
 		await this.#replayAttachment(attached, attached.cursor.seq);
 		return true;

packages/coding-agent/test/sdk-session-router-authority.test.ts

@@ -1546,4 +1546,125 @@ describe("SessionRouter dispatch authority", () => {
 			await fixture.router.stop();
 		}
 	});
+
+	test("periodic reconcile converges while a rehosted attachment's replay is wedged (#4527)", async () => {
+		// Reproduces the production wedge: a session-host rehost bumps
+		// endpointGeneration, and the periodic reconcile replaces the attachment
+		// with one whose event_replay never settles. Before the fix, that replay
+		// was awaited inside the serialized reconcile tail, so every later tick
+		// froze until the replay budget expired; publications died while leases
+		// and inbound stayed green (#4527).
+		const repo = await fsPromises.mkdtemp(path.join(os.tmpdir(), "gjc-router-4527-"));
+		tempDirs.push(repo);
+		const agentDir = path.join(repo, ".gjc", "agent");
+		const stateRoot = path.join(repo, ".gjc", "state");
+		const endpointDir = path.join(stateRoot, "sdk");
+		await fsPromises.mkdir(endpointDir, { recursive: true });
+		const sessionId = "wedge";
+		const endpointFile = path.join(endpointDir, `${sessionId}.json`);
+		await Bun.write(endpointFile, JSON.stringify({ sessionId, url: "ws://wedge.test", token: "v1", pid: 42 }));
+		let generation = 1;
+		let wedgeReplay = false;
+		const wedgedGate = Promise.withResolvers<void>();
+		let reconcileCount = 0;
+		let tick: (() => void) | undefined;
+
+		const index = {
+			open: async () => {},
+			refresh: async () => {},
+			listSessions: () => ({
+				indexSeq: generation,
+				sessions: [
+					{
+						sessionId,
+						locator: { repo, stateRoot },
+						endpointGeneration: generation,
+						pid: 42,
+						endpointMtimeMs: fs.statSync(endpointFile).mtimeMs,
+						live: true,
+						indexSeq: generation,
+						ambiguous: false,
+						terminal: false,
+					},
+				],
+				warnings: [],
+			}),
+		} as unknown as SessionIndex;
+
+		const router = new SessionRouter({
+			agentDir,
+			deps: {
+				createIndex: () => index,
+				createClient: async () => ({
+					onFrame: () => () => {},
+					request: async (frame: Record<string, unknown>) => {
+						if (wedgeReplay && frame.type === "event_replay") await wedgedGate.promise;
+						return { events: [] };
+					},
+					close: async () => {},
+					send: () => {},
+				}),
+				onReconciled: () => {
+					reconcileCount++;
+				},
+				setInterval: ((callback: () => void) => {
+					tick = callback;
+					return 0;
+				}) as unknown as typeof setInterval,
+				clearInterval: (() => {}) as unknown as typeof clearInterval,
+			},
+		});
+
+		try {
+			await router.start();
+			expect(router.attachment(sessionId)?.isCurrent()).toBe(true);
+			const baseline = reconcileCount;
+
+			// Bump generation and rewrite the endpoint file: the periodic
+			// reconcile must replace the attachment. After the replacement,
+			// the new host's event_replay never settles.
+			generation = 2;
+			await Bun.write(endpointFile, JSON.stringify({ sessionId, url: "ws://wedge.test", token: "v2", pid: 42 }));
+			wedgeReplay = true;
+
+			// A publication-driven reconcile can observe the rehost before the
+			// periodic timer. It must publish and dispatch without awaiting the
+			// replacement attachment's wedged replay on the shared tail.
+			const requestSettled = await Promise.race([
+				Bun.sleep(500).then(() => false),
+				router.request(sessionId, { type: "test" }).then(() => true),
+			]);
+			expect(requestSettled).toBe(true);
+			expect(reconcileCount).toBeGreaterThan(baseline);
+
+			// A later periodic tick must also converge: the reconcile tail is not
+			// held by the wedged replay living on the attachment's ready tail.
+			const beforeSecond = reconcileCount;
+			tick!();
+			const secondSettled = await Promise.race([
+				Bun.sleep(500).then(() => false),
+				(async () => {
+					for (let i = 0; i < 500 && reconcileCount <= beforeSecond; i++) await Bun.sleep(1);
+					return reconcileCount > beforeSecond;
+				})(),
+			]);
+			expect(secondSettled).toBe(true);
+
+			// Explicit reconciliation preserves its synchronous replay contract,
+			// but joins the per-attachment tail outside the serialized reconcile
+			// tail so periodic fleet convergence remains independent.
+			let explicitSettled = false;
+			const explicitReconcile = router.reconcile().then(() => {
+				explicitSettled = true;
+			});
+			await Bun.sleep(10);
+			expect(explicitSettled).toBe(false);
+			wedgedGate.resolve();
+			await explicitReconcile;
+			expect(explicitSettled).toBe(true);
+		} finally {
+			wedgedGate.resolve();
+			await router.stop();
+		}
+	});
 });