import type { ExtensionAPI, ExtensionContext } from "@earendil-works/pi-coding-agent"; type MemoryContext = Pick; export type EnsureMemoryReady = (ctx: MemoryContext, signal?: AbortSignal) => Promise; /** Cancel one waiter without cancelling initialization shared with other users. */ function waitForReady(promise: Promise, signal?: AbortSignal): Promise { if (!signal) return promise; return new Promise((resolve, reject) => { const abort = () => reject(signal.reason ?? new Error("Memory operation aborted")); signal.addEventListener("abort", abort, { once: true }); if (signal.aborted) abort(); promise.then(resolve, reject).finally(() => signal.removeEventListener("abort", abort)); }); } /** Own initialization, per-use preparation and operations through shutdown. */ export function createMemoryInitializer( initialize: () => Promise, prepare?: (ctx: MemoryContext) => Promise, ) { let pending: Promise | undefined; let initialized = false; let ready = false; let closed = false; const active = new Set>(); function assertOpen(): void { if (closed) throw new Error("Memory session has shut down"); } function track(work: Promise): Promise { active.add(work); void work.then(() => active.delete(work), () => active.delete(work)); return work; } async function ensure(ctx?: MemoryContext, signal?: AbortSignal): Promise { assertOpen(); signal?.throwIfAborted(); if (!initialized) { pending ??= Promise.resolve().then(initialize).then(() => { initialized = true; }).finally(() => { pending = undefined; }); } // Track the underlying preparation even when an individual waiter cancels. const preparation = track((pending ?? Promise.resolve()).then(async () => { if (ctx) await prepare?.(ctx); ready = true; })); await waitForReady(preparation, signal); signal?.throwIfAborted(); assertOpen(); } return { isReady: () => ready, ensure, run(ctx: MemoryContext, work: () => Promise, signal?: AbortSignal): Promise { return track(ensure(ctx, signal).then(() => { assertOpen(); signal?.throwIfAborted(); return work(); })); }, async close(): Promise { closed = true; // No new work is accepted. Join loads, project binds, backfill and callers // already executing before the owner checkpoints/closes its database. await Promise.allSettled([...active]); }, }; } /** Guard memory entry points without changing schemas, rendering or events. */ export function withMemoryInitialization( pi: ExtensionAPI, initialization: Pick, "run">, ): ExtensionAPI { return { ...pi, registerTool(tool) { pi.registerTool({ ...tool, async execute(id, params, signal, onUpdate, ctx) { return initialization.run(ctx, () => tool.execute(id, params, signal, onUpdate, ctx), signal); }, }); }, registerCommand(name, options) { pi.registerCommand(name, { ...options, async handler(args, ctx) { return initialization.run(ctx, () => options.handler(args, ctx)); }, }); }, }; }