InputSource,
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";
let observerQueue = Promise.resolve();
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);
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");
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 }) => ({
id,
messages: AgentEndEvent["messages"],
targetMessageIds: ReadonlySet<string>,
boundaryEntryId: string | null,
): Promise<void> => {
if (!branchContains(ctx, boundaryEntryId)) return;
if (!onLiveBranch(ctx, boundaryEntryId)) return;
const registry = registryFrom(ctx);
const currentTargets = new Set(
registry.unreconciled
.map((message) => message.messageId)
);
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;
}
pendingReconciliations += 1;
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;
});