From 0c81dae753df49bb85228883db44db6855aaca25 Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Sat, 12 Sep 2026 06:44:02 +0300 Subject: [PATCH] src,index.ts,test: skip reconciliation on a replaced session --- index.ts | 24 ++++++++++++--------- src/liveness.ts | 15 +++++++++++++ test/stale-session.test.ts | 44 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 73 insertions(+), 10 deletions(-) create mode 100644 src/liveness.ts create mode 100644 test/stale-session.test.ts diff --git a/index.ts b/index.ts index e698984..4b33b34 100644 --- a/index.ts +++ b/index.ts @@ -9,6 +9,7 @@ import type { SessionEntry, } from "@earendil-works/pi-coding-agent"; import { CLEARED_CONTEXT, nextContextRevision, REFRESHING_CONTEXT } from "./src/inject.ts"; +import { onLiveBranch } from "./src/liveness.ts"; import { configFromEnv, runObserver } from "./src/observer.ts"; import { isCommitmentEvent, liveThreads, observationChangesState, replay } from "./src/state.ts"; import type { CommitmentEvent, Registry } from "./src/types.ts"; @@ -66,18 +67,21 @@ export default function commitmentObserver(pi: ExtensionAPI): void { let pendingReconciliations = 0; let consecutiveFailures = 0; + // The failure path has no owner to catch it: a replaced session has nowhere to record the + // failure, and the queue that holds this work discards the rejection. const recordFailure = (ctx: ExtensionContext, messageIds: string[], error: unknown): void => { const text = (error instanceof Error ? error.message : String(error)).slice(0, 2_000); - appendEvent(pi, { kind: "observer_failure", messageIds, error: text, at: Date.now() }); - consecutiveFailures += 1; - if (consecutiveFailures === 1 && ctx.hasUI) { - ctx.ui.notify(`Commitment observer failed: ${text}`, "warning"); + try { + appendEvent(pi, { kind: "observer_failure", messageIds, error: text, at: Date.now() }); + consecutiveFailures += 1; + if (consecutiveFailures === 1 && ctx.hasUI) { + ctx.ui.notify(`Commitment observer failed: ${text}`, "warning"); + } + } catch { + // The session was replaced under this reconciliation. } }; - const branchContains = (ctx: ExtensionContext, entryId: string | null): boolean => - entryId === null || ctx.sessionManager.getBranch().some((entry) => entry.id === entryId); - const liveRevision = (registry: Registry): string => JSON.stringify( liveThreads(registry).map(({ id, canonicalRequest, status, updatedAt }) => ({ @@ -95,7 +99,7 @@ export default function commitmentObserver(pi: ExtensionAPI): void { targetMessageIds: ReadonlySet, boundaryEntryId: string | null, ): Promise => { - if (!branchContains(ctx, boundaryEntryId)) return; + if (!onLiveBranch(ctx, boundaryEntryId)) return; const registry = registryFrom(ctx); const currentTargets = new Set( registry.unreconciled @@ -105,7 +109,7 @@ export default function commitmentObserver(pi: ExtensionAPI): void { if (currentTargets.size === 0 && liveThreads(registry).length === 0) return; const revision = liveRevision(registry); const result = await runObserver(registry, messages, currentTargets, ctx.modelRegistry, cfg); - if (!branchContains(ctx, boundaryEntryId) || liveRevision(registryFrom(ctx)) !== revision) return; + if (!onLiveBranch(ctx, boundaryEntryId) || liveRevision(registryFrom(ctx)) !== revision) return; if (!result.ok) { recordFailure(ctx, [...currentTargets], result.error); return; @@ -130,7 +134,7 @@ export default function commitmentObserver(pi: ExtensionAPI): void { observerQueue = observerQueue .then(() => reconcile(ctx, messages, targetMessageIds, boundaryEntryId)) .catch((error) => { - if (branchContains(ctx, boundaryEntryId)) recordFailure(ctx, [...targetMessageIds], error); + if (onLiveBranch(ctx, boundaryEntryId)) recordFailure(ctx, [...targetMessageIds], error); }) .finally(() => { pendingReconciliations -= 1; diff --git a/src/liveness.ts b/src/liveness.ts new file mode 100644 index 0000000..fc96399 --- /dev/null +++ b/src/liveness.ts @@ -0,0 +1,15 @@ +import type { ExtensionContext } from "@earendil-works/pi-coding-agent"; + +/** + * A reconciliation outlives the session it began in, and every extension ctx accessor throws once + * that session is replaced or reloaded. Reading the branch is the only way to ask, so a replaced + * session answers "no" instead of throwing into callers that have nowhere to catch it. + */ +export function onLiveBranch(ctx: ExtensionContext, entryId: string | null): boolean { + try { + const branch = ctx.sessionManager.getBranch(); + return entryId === null || branch.some((entry) => entry.id === entryId); + } catch { + return false; + } +} diff --git a/test/stale-session.test.ts b/test/stale-session.test.ts new file mode 100644 index 0000000..f80b32e --- /dev/null +++ b/test/stale-session.test.ts @@ -0,0 +1,44 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import type { ExtensionContext, SessionEntry } from "@earendil-works/pi-coding-agent"; +import { onLiveBranch } from "../src/liveness.ts"; + +const STALE = "This extension ctx is stale after session replacement or reload."; + +const entry = (id: string): SessionEntry => + ({ + type: "custom", + id, + parentId: null, + timestamp: "2026-09-12T00:00:00.000Z", + customType: "unrelated", + data: {}, + }) as SessionEntry; + +const context = (branch: SessionEntry[] | (() => never)): ExtensionContext => + ({ + get sessionManager() { + return typeof branch === "function" + ? { + getBranch: branch, + } + : { getBranch: () => branch }; + }, + }) as unknown as ExtensionContext; + +test("a boundary still on the branch is live", () => { + assert.equal(onLiveBranch(context([entry("a"), entry("b")]), "a"), true); +}); + +test("a boundary that left the branch is not live", () => { + assert.equal(onLiveBranch(context([entry("a")]), "b"), false); +}); + +test("a replaced session is not a live branch and does not throw", () => { + const replaced = context(() => { + throw new Error(STALE); + }); + + assert.equal(onLiveBranch(replaced, "a"), false); + assert.equal(onLiveBranch(replaced, null), false); +}); -- 2.51.2