diff --git a/packages/pi-plugin/src/dreamer/index.test.ts b/packages/pi-plugin/src/dreamer/index.test.ts index 903d84e73..bc8084475 100644 --- a/packages/pi-plugin/src/dreamer/index.test.ts +++ b/packages/pi-plugin/src/dreamer/index.test.ts @@ -370,6 +370,32 @@ describe("Pi dreamer wiring", () => { expect(tasks[0].modelChainUnavailable).toBe(true); }); + test("re-probes a model that becomes registered after an early miss", () => { + // OMP queues a dynamic-only extension provider (`litellm`) during extension + // loading and merges its catalogue in a fire-and-forget boot refresh, so a + // miss at registration normally means "not hydrated yet". Caching the miss + // kept the historian chain empty for the whole process even after the model + // appeared -- observed in the field as `registered historian trigger: + // DISABLED` in two of four OMP processes started within one minute. + let hydrated = false; + const registry = { + find: (_provider: string, model: string) => + hydrated && model === "historian" ? {} : undefined, + }; + const task = { + task: "classify-memories" as const, + schedule: "0 4 * * *", + timeoutMinutes: 20, + model: "litellm/historian", + }; + expect(validatePiDreamerModels([task], registry)[0].model).toBeUndefined(); + hydrated = true; + expect(validatePiDreamerModels([task], registry)[0]).toMatchObject({ + model: "litellm/historian", + modelChainUnavailable: false, + }); + }); + test("does not register the filesystem root or home for dreaming", () => { db = createDb(); let starts = 0; diff --git a/packages/pi-plugin/src/dreamer/index.ts b/packages/pi-plugin/src/dreamer/index.ts index 9020c5f95..fdc56ddc7 100644 --- a/packages/pi-plugin/src/dreamer/index.ts +++ b/packages/pi-plugin/src/dreamer/index.ts @@ -155,7 +155,19 @@ export function validatePiDreamerModels( let available = resolved.get(model); if (available === undefined) { available = isPiModelRegistered(model, registry, harness); - resolved.set(model, available); + // Positives only. OMP hydrates discovery-backed providers such as + // `litellm` from a fire-and-forget boot refresh that the startup + // path never awaits (config/model-registry.ts + // `#refreshRuntimeDiscoveries`, main.ts `refreshInBackground`), and + // extension `registerProvider` calls are themselves queued until + // after extension loading -- so a miss during registration usually + // means "not yet", not "absent". Remembering the miss pinned the + // historian off for the whole process even after the model became + // visible; OMP's own consumers re-check after awaiting instead + // (session/agent-session.ts `#retryInactiveAdvisorAfterModelDiscovery`). + // A genuinely absent model costs one extra `find()` per re-resolution, + // which happens per transform pass. + if (available) resolved.set(model, available); } if (available) return true; if (!warnedUnknownModels.has(model)) { diff --git a/packages/pi-plugin/src/index.ts b/packages/pi-plugin/src/index.ts index a6a0f8ea4..814effc97 100644 --- a/packages/pi-plugin/src/index.ts +++ b/packages/pi-plugin/src/index.ts @@ -192,6 +192,7 @@ import { formatEmptyPiModelChain, formatEmptyPiModelChainsNotice, type PiModelRegistryLike, + waitForPiModelRegistryRefresh, } from "./model-chain-health"; import { bootPiRuntimeWithDeadline } from "./pi-boot-deadline"; import { @@ -1621,21 +1622,44 @@ async function startPiMagicContextRuntime( let activeModelRegistry: | { find(provider: string, modelId: string): unknown } | undefined; + // The latest session context Pi handed the extension, kept alongside the + // active registry. A report deferred behind a discovery refresh resolves + // its dependencies through this instead of replaying the context captured + // when the wait was scheduled: after a session switch that capture belongs + // to a dead session (its `ui` reaches nowhere, its registry is not the + // live catalogue), and after a config reload its project is stale. The + // context and its cwd stay paired: reports belong to the session's + // current project, never to a project merely cached in the process. + type PiModelChainReportContext = { + modelRegistry?: PiModelRegistryLike; + cwd?: string; + hasUI?: boolean; + ui?: { notify?: (message: string, level?: "warning") => unknown }; + }; + let activeModelChainContext: PiModelChainReportContext | undefined; // Config generation last checked per project directory. const modelChainsCheckedGeneration = new Map(); + // Project directories whose chain report already awaited the host's + // discovery-backed catalogue once. Set only when that await has settled. + const modelChainReadinessWaited = new Set(); + // Project directories with a discovery await still in flight. Reports + // defer to it while it is pending; it is the only thing allowed to + // schedule the settled re-check. + const modelChainRefreshPending = new Set(); /** * Validate the historian and dreamer chains against Pi's model registry: * log the historian chain that will actually run, and notify the session * once per process for each set of chains left empty (see * model-chain-health.ts). Runs at session start and after a config reload. + * An empty chain is first re-checked after one background-refresh await, + * because a registration-time miss on a discovery-backed provider means + * "not hydrated yet" far more often than "not configured". While that + * await is in flight every further report defers to it, and it re-checks + * through current lifecycle state, never the pass that scheduled it. */ function reportPiModelChains( - ctx: { - modelRegistry?: PiModelRegistryLike; - hasUI?: boolean; - ui?: { notify?: (message: string, level?: "warning") => unknown }; - }, + ctx: PiModelChainReportContext, project: ResolvedPiProjectDeps, ): void { const registry = ctx.modelRegistry; @@ -1654,6 +1678,51 @@ async function startPiMagicContextRuntime( registry, harness: PI_HARNESS_KIND, }); + + // Nothing is logged or notified from a pass that found an empty + // chain before the host's discovery has settled: on OMP the provider + // that owns these models is queued during extension loading and + // merged by a boot refresh nobody awaits, so announcing "will not + // run" from an unsettled catalogue is usually wrong and, once + // notified, sticky for the process through + // loggedPiModelChainLines/notifiedPiModelChainKeys. One await per + // project: any event arriving inside the refresh window -- a second + // session_start, or the reload-driven agent_start gate -- defers to + // the in-flight await instead of skipping the wait and reporting the + // miss. + if ( + empty.length > 0 && + !modelChainReadinessWaited.has(project.projectDir) + ) { + if (!modelChainRefreshPending.has(project.projectDir)) { + modelChainRefreshPending.add(project.projectDir); + const waitingDir = project.projectDir; + void waitForPiModelRegistryRefresh(registry).then(() => { + // Marked settled only now: the await has actually + // finished, so a miss remaining from here on is + // trustworthy and later passes may report it. + modelChainRefreshPending.delete(waitingDir); + modelChainReadinessWaited.add(waitingDir); + try { + reportPiModelChainsAfterRefresh(waitingDir); + } catch (err) { + // The re-check resolves deps and polls the config + // outside reportPiModelChains' own try/catch; a + // rejection on the void chain. The settled marker + // stands (the await really finished) and the last + // checked generation was not updated -- this throw + // happens before reportPiModelChains records it -- + // so the next session_start re-runs the check and + // reports directly, and a config reload that moves + // the generation re-runs it through agent_start's + // gate; with the config unchanged, agent_start + // alone does not re-report. + warn("model chain post-refresh check failed:", err); + } + }); + } + return; + } const emptyHistorian = empty.find((chain) => chain.owner === "historian"); const historianLine = historian ? `registered historian trigger (model=${historian.model}${historian.fallbackModels?.length ? `, fallbacks=${historian.fallbackModels.join(",")}` : ""}, executeThreshold=${formatExecuteThresholdForLog(historian.executeThresholdPercentage)})` @@ -1680,6 +1749,32 @@ async function startPiMagicContextRuntime( warn("model chain check failed:", err); } } + /** + * Re-run the chain report once a scheduled discovery refresh has settled. + * Nothing from the scheduled pass is replayed: the session may have been + * switched and the config reloaded while the await was open, so the + * report resolves the current session context and this directory's + * dependencies from the lifecycle state above. The context and project + * stay paired -- if the session switched to another project, this one's + * warning must not reach that project's UI, and the once-per-process + * notice must not be spent here; the next session_start of this + * directory re-checks (the settled-catalogue marker is process-wide, so + * it reports directly). A config whose generation moved past the last + * checked one was replaced while the await was open, and its report is + * left to the reload-aware path (agent_start's generation gate, or the + * next session_start) rather than announced from a superseded file. + */ + function reportPiModelChainsAfterRefresh(projectDir: string): void { + const ctx = activeModelChainContext; + const registry = ctx?.modelRegistry; + if (!ctx || !registry) return; + if (ctx.cwd !== undefined && ctx.cwd !== projectDir) return; + const project = resolveProjectDepsForDir(projectDir); + const generation = liveReaderFor(project.projectDir, project.config).poll() + .generation; + if (modelChainsCheckedGeneration.get(projectDir) !== generation) return; + reportPiModelChains(ctx, project); + } function resolveContextOptionsForProject( dir: string, ): PiContextHandlerOptions { @@ -1829,6 +1924,7 @@ async function startPiMagicContextRuntime( projectDepsByDir.delete(ctx.cwd); const current = resolveCurrentProjectDeps(ctx); activeModelRegistry = ctx.modelRegistry; + activeModelChainContext = ctx; reportPiModelChains(ctx, current); if (ctx.hasUI) syncDreamerProjectRegistration(current, ctx.modelRegistry); syncCtxMemoryToolEnabled(pi, current.config.memory.enabled); @@ -2252,6 +2348,7 @@ async function startPiMagicContextRuntime( // shared helper releases that ownership explicitly. try { activeModelRegistry = ctx.modelRegistry; + activeModelChainContext = ctx; if (ctx.hasUI) syncDreamerProjectRegistration( effectiveProjectDeps, diff --git a/packages/pi-plugin/src/model-chain-health-pi.test.ts b/packages/pi-plugin/src/model-chain-health-pi.test.ts index 3a47ab25d..5b8a3be87 100644 --- a/packages/pi-plugin/src/model-chain-health-pi.test.ts +++ b/packages/pi-plugin/src/model-chain-health-pi.test.ts @@ -2,6 +2,7 @@ import { afterEach, describe, expect, it, mock, spyOn } from "bun:test"; import { mkdirSync, writeFileSync } from "node:fs"; import { join } from "node:path"; import type { ExtensionAPI } from "@earendil-works/pi-coding-agent"; +import { LiveConfigReader } from "@magic-context/core/config/live-snapshot"; import type { MagicContextConfig } from "@magic-context/core/config/schema/magic-context"; import { clearSession, @@ -219,6 +220,35 @@ describe("Pi extension reports an empty historian chain at session start", () => return root; } + // A second, independently configured project: Pi hands every session + // context the operator's cwd, and the report machinery keys its reader and + // deps caches on that directory, so a switch between projects needs a + // distinct cwd -- and the project tier must sit under it, not under the + // temp root. + function projectCwd(root: string, name: string, config: unknown): string { + const cwd = join(root, name); + mkdirSync(join(cwd, ".cortexkit"), { recursive: true }); + writeFileSync( + join(cwd, ".cortexkit", "magic-context.jsonc"), + JSON.stringify(config), + ); + return cwd; + } + + // Only the historian trigger lines, so an assertion names what the report + // decided instead of incidental boot logging. + function triggerLines(logs: string[]): string[] { + return logs.filter((line) => line.includes("registered historian trigger")); + } + + // Yield until queued microtasks -- the settled refresh's re-check chain -- + // have run: setImmediate is the first macrotask after them. + function macrotask(): Promise { + const { promise, resolve } = Promise.withResolvers(); + setImmediate(resolve); + return promise; + } + it("/ctx-status lists only chains that can run, with suggestions and the inert drain latch", async () => { const root = isolatedConfig({ historian: { pi: HISTORIAN_CHAIN }, @@ -360,4 +390,434 @@ describe("Pi extension reports an empty historian chain at session start", () => "openai/gpt-6.1-sol (did you mean openai-codex/gpt-6.1-sol?)", ); }, 20_000); + + it("waits for the host's discovery refresh before declaring the chain dead", async () => { + // The OMP shape seen in the field: `litellm` is a dynamic-only extension + // provider, so its models exist only after the boot-time + // refreshInBackground() merge settles -- which nothing on the startup path + // awaits -- and extension registerProvider calls are queued until after + // extension loading. One synchronous find() there is a "not yet", not an + // "absent"; reporting it as `registered historian trigger: DISABLED` told + // operators their working historian config was broken. + const root = isolatedConfig({ + historian: { pi: { model: "litellm/google/gemini-3.1-flash-lite" } }, + dreamer: { disable: true }, + }); + const logs: string[] = []; + spyOn(loggerModule, "log").mockImplementation((message: unknown) => { + logs.push(String(message)); + }); + dreamerTest.setStartDreamScheduleTimerFactory(async () => () => {}); + + let hydrated = false; + const hydratedModels = [ + { provider: "litellm", id: "google/gemini-3.1-flash-lite" }, + ]; + const initialDiscovery = Promise.withResolvers(); + const lateRegistry = { + find: (provider: string, id: string) => + (hydrated ? hydratedModels : []).find( + (m) => m.provider === provider && m.id === id, + ), + getAll: () => (hydrated ? hydratedModels : []), + // The in-flight API is a no-op before OMP starts discovery. + awaitBackgroundRefresh: async () => {}, + awaitInitialBackgroundRefresh: () => initialDiscovery.promise, + }; + + const runtime = createPi(); + await magicContextPiExtension(runtime.pi); + const notify = mock((_message: string, _level?: string) => undefined); + const ctx = { + cwd: root, + hasUI: true, + modelRegistry: lateRegistry, + sessionManager: { getSessionId: () => "ses-hydrating" }, + ui: { notify, setStatus: () => undefined }, + }; + await runtime.emit("session_start", ctx); + await new Promise((resolve) => setImmediate(resolve)); + expect(notify).not.toHaveBeenCalled(); + expect( + logs.some((line) => line.includes("registered historian trigger")), + ).toBe(false); + hydrated = true; + initialDiscovery.resolve(); + const { promise: settled, resolve: resolveSettled } = + Promise.withResolvers(); + setImmediate(resolveSettled); + await settled; + + // Exactly one trigger line, and it is the running one: a first pass that + // announced DISABLED would leave a second element here. + const historianLines = logs.filter((line) => + line.includes("registered historian trigger"), + ); + expect(historianLines).toEqual([ + expect.stringContaining("(model=litellm/google/gemini-3.1-flash-lite"), + ]); + expect(notify).toHaveBeenCalledTimes(0); + }, 20_000); + + it("holds every report back while the discovery refresh is still pending", async () => { + // The second-event bypass the fix removes: the settled marker used to be + // set when the await was *scheduled*, so a session_start arriving + // inside the refresh window skipped the wait and pinned + // `registered historian trigger: DISABLED` for the process from a + // catalogue that was still hydrating. + const root = isolatedConfig({ + historian: { pi: { model: "litellm/google/gemini-3.1-flash-lite" } }, + dreamer: { disable: true }, + }); + const logs: string[] = []; + spyOn(loggerModule, "log").mockImplementation((message: unknown) => { + logs.push(String(message)); + }); + dreamerTest.setStartDreamScheduleTimerFactory(async () => () => {}); + + let hydrated = false; + let resolveRefresh!: () => void; + const refreshing = new Promise((resolve) => { + resolveRefresh = resolve; + }); + const lateRegistry = { + find: (provider: string, id: string) => + (hydrated + ? [{ provider: "litellm", id: "google/gemini-3.1-flash-lite" }] + : [] + ).find((model) => model.provider === provider && model.id === id), + getAll: () => + hydrated + ? [{ provider: "litellm", id: "google/gemini-3.1-flash-lite" }] + : [], + awaitBackgroundRefresh: () => refreshing, + }; + + const runtime = createPi(); + await magicContextPiExtension(runtime.pi); + const notify = mock((_message: string, _level?: string) => undefined); + const ctx = { + cwd: projectCwd(root, "repo-pending", { dreamer: { disable: true } }), + hasUI: true, + modelRegistry: lateRegistry, + sessionManager: { getSessionId: () => "ses-pending" }, + ui: { notify, setStatus: () => undefined }, + }; + // First pass schedules the await; a second session_start and a turn + // arrive while it is still in flight. + await runtime.emit("session_start", ctx); + await runtime.emit("session_start", ctx); + await runtime.agentTurn(ctx); + + expect(triggerLines(logs)).toEqual([]); + expect(notify).toHaveBeenCalledTimes(0); + + // The refresh settles: one report, from the settled catalogue. + hydrated = true; + resolveRefresh(); + await macrotask(); + + expect(triggerLines(logs)).toEqual([ + expect.stringContaining("(model=litellm/google/gemini-3.1-flash-lite"), + ]); + expect(notify).toHaveBeenCalledTimes(0); + }, 20_000); + + it("pairs the deferred report with the active session and never spends its notice budget", async () => { + // The cross-project leak the fix removes: the settled re-check combined + // the latest session's context with the *waiting* project's + // configuration, so a switch from A (awaiting discovery) to B announced + // A's missing models through B's UI -- even a B that has dreaming + // disabled -- and consumed A's once-per-process notice, which hid A's + // own warning when the session came back. The context and project are + // kept paired: an inactive project's deferred report is skipped, and + // the budget survives for the next session_start of that project. + // One operator config: historian model selection is user-tier by + // design (a repository cannot choose the historian's model), so both + // projects share it and differ by cwd and registry. + const root = isolatedConfig({ + historian: { pi: { model: "litellm/deepseek-v4-flash" } }, + dreamer: { + pi: { model: "litellm/curate-missing" }, + tasks: { curate: { schedule: "0 4 * * 0" } }, + }, + }); + const logs: string[] = []; + spyOn(loggerModule, "log").mockImplementation((message: unknown) => { + logs.push(String(message)); + }); + dreamerTest.setStartDreamScheduleTimerFactory(async () => () => {}); + + let resolveRefresh!: () => void; + const refreshing = new Promise((resolve) => { + resolveRefresh = resolve; + }); + // The awaiting session's catalogue never gains the models. + const stalledRegistry = { + find: () => undefined, + getAll: () => [], + awaitBackgroundRefresh: () => refreshing, + }; + // B's catalogue knows the historian but never the dreamer task: the + // waiting project's warning must not leak through this UI. + const partialRegistry = { + find: (provider: string, id: string) => + [{ provider: "litellm", id: "deepseek-v4-flash" }].find( + (model) => model.provider === provider && model.id === id, + ), + getAll: () => [{ provider: "litellm", id: "deepseek-v4-flash" }], + }; + + const runtime = createPi(); + await magicContextPiExtension(runtime.pi); + const staleNotify = mock(() => undefined); + // A schedules with the dreamer enabled (both projects read the same + // operator config); the return leg re-resolves A after B disabled it, + // so the per-directory deps cache must not still hold the old object. + const staleCwd = projectCwd(root, "repo-stale", { + dreamer: { disable: false }, + }); + await runtime.emit("session_start", { + cwd: staleCwd, + hasUI: true, + modelRegistry: stalledRegistry, + sessionManager: { getSessionId: () => "ses-stale" }, + ui: { notify: staleNotify, setStatus: () => undefined }, + }); + expect(triggerLines(logs)).toEqual([]); + + const liveNotify = mock(() => undefined); + const liveCwd = projectCwd(root, "repo-live", { + dreamer: { disable: true }, + }); + await runtime.emit("session_start", { + cwd: liveCwd, + hasUI: true, + modelRegistry: partialRegistry, + sessionManager: { getSessionId: () => "ses-live" }, + ui: { notify: liveNotify, setStatus: () => undefined }, + }); + + resolveRefresh(); + await macrotask(); + + // The switched-away project's report was skipped: nothing reached B's + // UI and A's UI was not replayed either; the one line logged is B's + // own session reporting its registered historian. Pre-fix, the settle + // pass announced A's chains (DISABLED for the unregistered dreamer + // task) through B's UI and spent A's once-per-process notice. + expect(staleNotify).toHaveBeenCalledTimes(0); + expect(liveNotify).toHaveBeenCalledTimes(0); + expect(triggerLines(logs)).toEqual([ + expect.stringContaining("(model=litellm/deepseek-v4-flash"), + ]); + + // A's notice budget survived: returning to A reports A's own still + // empty dreamer chain through A's UI (the historian line deduplicates, + // and A's settled-catalogue marker suppresses a second await). + await runtime.emit("session_start", { + cwd: staleCwd, + hasUI: true, + modelRegistry: partialRegistry, + sessionManager: { getSessionId: () => "ses-stale-again" }, + ui: { notify: staleNotify, setStatus: () => undefined }, + }); + expect(triggerLines(logs)).toEqual([ + expect.stringContaining("(model=litellm/deepseek-v4-flash"), + ]); + expect(staleNotify).toHaveBeenCalledTimes(1); + const notice = String(staleNotify.mock.calls[0]?.[0]); + expect(notice).toContain("curate"); + expect(notice).toContain("litellm/curate-missing"); + }, 20_000); + + it("leaves a replaced config's deferred report to the reload-aware path", async () => { + // Config generation verification: the refresh was awaited for one + // configuration, but the operator reloaded the file before it settled. + // Replaying the scheduled pass after the wait would warn about chains + // that no longer describe the config on disk; the generation gate owns + // the new file, and the next agent turn gets the single report. + const root = isolatedConfig({ + historian: { pi: { model: "litellm/google/gemini-3.1-flash-lite" } }, + dreamer: { disable: true }, + }); + const cwd = projectCwd(root, "repo-reload", { dreamer: { disable: true } }); + const logs: string[] = []; + spyOn(loggerModule, "log").mockImplementation((message: unknown) => { + logs.push(String(message)); + }); + dreamerTest.setStartDreamScheduleTimerFactory(async () => () => {}); + + // Unhydrated while the refresh is pending, so the scheduled pass finds + // the empty chain; the refresh settles the catalogue either way. + let hydrated = false; + let resolveRefresh!: () => void; + const refreshing = new Promise((resolve) => { + resolveRefresh = () => { + hydrated = true; + resolve(); + }; + }); + const lateRegistry = { + find: (provider: string, id: string) => + (hydrated + ? [ + { provider: "litellm", id: "google/gemini-3.1-flash-lite" }, + { provider: "litellm", id: "openai/gpt-4.1-mini" }, + ] + : [] + ).find((model) => model.provider === provider && model.id === id), + getAll: () => + hydrated + ? [ + { provider: "litellm", id: "google/gemini-3.1-flash-lite" }, + { provider: "litellm", id: "openai/gpt-4.1-mini" }, + ] + : [], + awaitBackgroundRefresh: () => refreshing, + }; + + const runtime = createPi(); + await magicContextPiExtension(runtime.pi); + const notify = mock((_message: string, _level?: string) => undefined); + const ctx = { + cwd, + hasUI: true, + modelRegistry: lateRegistry, + sessionManager: { getSessionId: () => "ses-reloaded" }, + ui: { notify, setStatus: () => undefined }, + }; + await runtime.emit("session_start", ctx); + + // The operator's reload moves both tiers: poll() re-reads the effective + // config from disk, and the generation the deferred report verifies + // against must have visibly moved. + writeFileSync( + join(root, "config", "cortexkit", "magic-context.jsonc"), + JSON.stringify({ + historian: { pi: { model: "litellm/openai/gpt-4.1-mini" } }, + dreamer: { disable: true }, + }), + ); + writeFileSync( + join(cwd, ".cortexkit", "magic-context.jsonc"), + JSON.stringify({ + historian: { pi: { model: "litellm/openai/gpt-4.1-mini" } }, + dreamer: { disable: true }, + }), + ); + + resolveRefresh(); + await macrotask(); + + // Nothing announced for the replaced configuration. + expect(triggerLines(logs)).toEqual([]); + expect(notify).toHaveBeenCalledTimes(0); + + // The reload-aware path (agent_start's generation gate) reports the + // current file once. + await runtime.agentTurn(ctx); + expect(triggerLines(logs)).toEqual([ + expect.stringContaining("(model=litellm/openai/gpt-4.1-mini"), + ]); + expect(notify).toHaveBeenCalledTimes(0); + }, 20_000); + + it("warns instead of rejecting when the settled re-check throws", async () => { + // The settled callback runs resolveProjectDepsForDir and poll() + // outside reportPiModelChains' own try/catch; a throw there used to + // escape as an unhandled rejection on the void chain. The re-check is + // guarded, the settled marker stands (the await really finished), and + // the next session_start reports directly. + const root = isolatedConfig({ + historian: { pi: { model: "litellm/google/gemini-3.1-flash-lite" } }, + dreamer: { disable: true }, + }); + const logs: string[] = []; + spyOn(loggerModule, "log").mockImplementation((message: unknown) => { + logs.push(String(message)); + }); + dreamerTest.setStartDreamScheduleTimerFactory(async () => () => {}); + + let hydrated = false; + let resolveRefresh!: () => void; + const refreshing = new Promise((resolve) => { + resolveRefresh = resolve; + }); + const lateRegistry = { + find: (provider: string, id: string) => + (hydrated + ? [{ provider: "litellm", id: "google/gemini-3.1-flash-lite" }] + : [] + ).find((model) => model.provider === provider && model.id === id), + getAll: () => + hydrated + ? [{ provider: "litellm", id: "google/gemini-3.1-flash-lite" }] + : [], + awaitBackgroundRefresh: () => refreshing, + }; + + const runtime = createPi(); + await magicContextPiExtension(runtime.pi); + const notify = mock((_message: string, _level?: string) => undefined); + const cwd = projectCwd(root, "repo-settle-throw", { + dreamer: { disable: true }, + }); + const ctx = { + cwd, + hasUI: true, + modelRegistry: lateRegistry, + sessionManager: { getSessionId: () => "ses-settle-throw" }, + ui: { notify, setStatus: () => undefined }, + }; + await runtime.emit("session_start", ctx); + + // Make the settle-time re-check throw: reportPiModelChainsAfterRefresh + // calls liveReaderFor(...).poll() outside reportPiModelChains' own + // try/catch, so this is the call that used to escape as an unhandled + // rejection on the void chain. The spy goes in after the schedule + // pass (which polls too, but inside its guard) so only the settle + // trips it. + const pollSpy = spyOn( + LiveConfigReader.prototype, + "poll", + ).mockImplementation(() => { + throw new Error("settle-time config load failure"); + }); + + resolveRefresh(); + await macrotask(); + pollSpy.mockRestore(); + + // The run survived the settle: no trigger line from the crashed + // re-check, and the failure was warned through the logger, not thrown + // (bun fails a test on an unhandled rejection, so the guard is also + // proven by this test completing). + expect(triggerLines(logs)).toEqual([]); + expect( + logs.some((line) => + line.includes("WARN model chain post-refresh check failed"), + ), + ).toBe(true); + + // The recovery limit the guard's comment states: with the config + // unchanged the generation gate matches the last successful pass, so + // an agent turn does NOT re-run the check -- recovery is the next + // session_start below. + await runtime.agentTurn(ctx); + await macrotask(); + expect(triggerLines(logs)).toEqual([]); + + // The miss was reported later instead of lost: hydrate the catalogue + // and a fresh session_start of the waiting project reports directly + // (the settled marker stands -- the await really finished -- so zero + // lines here would mean the notice died waiting again forever). + hydrated = true; + await runtime.emit("session_start", ctx); + await macrotask(); + expect(triggerLines(logs)).toEqual([ + expect.stringContaining("(model=litellm/google/gemini-3.1-flash-lite"), + ]); + expect(notify).toHaveBeenCalledTimes(0); + }, 20_000); }); diff --git a/packages/pi-plugin/src/model-chain-health.ts b/packages/pi-plugin/src/model-chain-health.ts index b46048520..0f70deb7c 100644 --- a/packages/pi-plugin/src/model-chain-health.ts +++ b/packages/pi-plugin/src/model-chain-health.ts @@ -29,6 +29,9 @@ type PiHarness = "pi" | "omp"; export interface PiModelRegistryLike { find(provider: string, modelId: string): unknown; getAll?: () => ReadonlyArray<{ provider?: unknown; id?: unknown }>; + /** OMP only: resolves once the boot-time discovery merge has settled. */ + awaitBackgroundRefresh?: () => Promise; + awaitInitialBackgroundRefresh?: () => Promise; } export interface DroppedPiModel { @@ -70,6 +73,47 @@ export function isPiModelRegistered( ); } +/** + * Wait for the host's discovery-backed catalogue to settle before treating a + * `find()` miss as proof a model does not exist. + * + * OMP composes its catalogue in two phases: the `ModelRegistry` constructor + * loads only the synchronous layers, and a dynamic-only extension provider such + * as `litellm` appears only after the boot-time `refreshInBackground()` merge + * runs -- which OMP deliberately never awaits on the startup path, and whose + * extension `registerProvider` calls are themselves queued until after + * extension loading. So a synchronous lookup during extension registration can + * miss a model that exists, and it does so intermittently (observed: four OMP + * processes inside one minute, two registering the historian and two + * reporting `registered historian trigger: DISABLED` for the same config). + * + * OMP's own consumers re-check after awaiting this (session/agent-session.ts + * `#retryInactiveAdvisorAfterModelDiscovery`, task/executor.ts). An extension + * that trusts one synchronous miss disables the feature for the life of the + * process. Prefer the initial-discovery barrier: it remains pending until the + * host completes discovery, including when discovery has not started yet. + * Older hosts expose only the in-flight barrier, which resolves immediately + * when idle. Pi without either API is checked synchronously. Refresh failures + * are swallowed so the caller can re-read the settled catalogue. + * Do not time out into an absence report: elapsed time is not evidence that + * a model is missing from a catalogue that has not finished loading. + */ +export async function waitForPiModelRegistryRefresh( + registry: Pick< + PiModelRegistryLike, + "awaitBackgroundRefresh" | "awaitInitialBackgroundRefresh" + >, +): Promise { + const refresh = + registry.awaitInitialBackgroundRefresh ?? registry.awaitBackgroundRefresh; + if (typeof refresh !== "function") return; + try { + await refresh.call(registry); + } catch { + // Swallowed on purpose: the caller re-reads the registry either way. + } +} + function registeredModels( registry: PiModelRegistryLike, ): Array<{ provider: string; id: string }> {