diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 0a71a1751..4ae58835f 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -100,6 +100,7 @@ - Fixed `setToolsExpanded(false)` to be a no-op when tool output is already collapsed, avoiding redundant `Tool output: collapsed` startup notices from extensions ([#7292](https://github.com/earendil-works/pi/issues/7292)). - Fixed extension-driven model calls in custom compaction, handoff, and Q&A examples to dispatch through the coding-agent model runtime so custom providers and resolved auth options are preserved ([#7325](https://github.com/earendil-works/pi/pull/7325)). - Fixed long-running sessions using stale credentials after another process updates `auth.json` without serializing concurrent credential reads and delaying startup ([#7319](https://github.com/earendil-works/pi/issues/7319)). +- Fixed concurrent `models-store.json` reads forming a file-lock convoy and delaying startup. - Updated the packaged `brace-expansion` dependency to 5.0.8 to address GHSA-mh99-v99m-4gvg ([#7316](https://github.com/earendil-works/pi/issues/7316)). - Fixed forced model availability refreshes remaining blocked behind a stalled earlier refresh ([#7301](https://github.com/earendil-works/pi/issues/7301), [#7421](https://github.com/earendil-works/pi/pull/7421) by [@a-yeyang](https://github.com/a-yeyang)). - Fixed `/model` catalog refresh failures to identify every catalog that failed. diff --git a/packages/coding-agent/src/core/auth-storage.ts b/packages/coding-agent/src/core/auth-storage.ts index 0b7e96392..5433c6e7e 100644 --- a/packages/coding-agent/src/core/auth-storage.ts +++ b/packages/coding-agent/src/core/auth-storage.ts @@ -4,13 +4,13 @@ */ import type { AuthOperationOptions, Credential, CredentialInfo, CredentialStore } from "@earendil-works/pi-ai"; -import { chmodSync, existsSync, mkdirSync, readFileSync, statSync, writeFileSync } from "fs"; +import { chmodSync, existsSync, mkdirSync, readFileSync, writeFileSync } from "fs"; import { dirname, join } from "path"; import lockfile from "proper-lockfile"; import { setTimeout as sleep } from "timers/promises"; import { getAgentDir } from "../config.ts"; import { raceWithAbortSignal } from "../utils/abort.ts"; -import { normalizePath } from "../utils/paths.ts"; +import { getFileRevision, normalizePath } from "../utils/paths.ts"; import { resolveConfigValue } from "./resolve-config-value.ts"; type AuthStorageData = Record; @@ -30,15 +30,6 @@ type AuthFileReadState = { let sharedAuthFileReadState: { authPath: string; readState: AuthFileReadState } | undefined; -function getFileRevision(path: string): string | undefined { - try { - const stats = statSync(path, { bigint: true }); - return `${stats.dev}:${stats.ino}:${stats.size}:${stats.mtimeNs}:${stats.ctimeNs}`; - } catch { - return undefined; - } -} - export interface AuthStorageBackend { withLock(fn: (current: string | undefined) => LockResult): T; withLockAsync( @@ -241,12 +232,16 @@ export class AuthStorage implements CredentialStore { private constructor(storage: AuthStorageBackend, authPath?: string) { this.storage = storage; - this.authPath = authPath; - this.readState = - authPath && sharedAuthFileReadState?.authPath === authPath ? sharedAuthFileReadState.readState : { data: {} }; - if (authPath) { + this.readState = { data: {} }; + if (authPath && !sharedAuthFileReadState) { + this.authPath = authPath; sharedAuthFileReadState = { authPath, readState: this.readState }; - const revision = getFileRevision(authPath); + } else if (authPath && sharedAuthFileReadState?.authPath === authPath) { + this.authPath = authPath; + this.readState = sharedAuthFileReadState.readState; + } + if (this.authPath) { + const revision = getFileRevision(this.authPath); if (revision !== undefined && revision === this.readState.revision) return; } this.reload(); @@ -308,11 +303,13 @@ export class AuthStorage implements CredentialStore { private readLatestData(options?: AuthOperationOptions): Promise { options?.signal?.throwIfAborted(); - if (this.authPath) { - const revision = getFileRevision(this.authPath); - if (revision !== undefined && revision === this.readState.revision) { - return Promise.resolve(this.readState.data); - } + if (!this.authPath) { + const reload = this.reloadFromStorageAsync(options); + return options?.signal ? reload : reload.catch(() => this.readState.data); + } + const revision = getFileRevision(this.authPath); + if (revision !== undefined && revision === this.readState.revision) { + return Promise.resolve(this.readState.data); } if (options?.signal) return this.reloadFromStorageAsync(options); if (!this.readState.reload) { diff --git a/packages/coding-agent/src/core/models-store.ts b/packages/coding-agent/src/core/models-store.ts index 649e54054..eb497c845 100644 --- a/packages/coding-agent/src/core/models-store.ts +++ b/packages/coding-agent/src/core/models-store.ts @@ -1,10 +1,27 @@ import { join } from "node:path"; import type { ModelsStore, ModelsStoreEntry, ModelsStoreOperationOptions } from "@earendil-works/pi-ai"; import { getAgentDir } from "../config.ts"; +import { raceWithAbortSignal } from "../utils/abort.ts"; +import { getFileRevision, normalizePath } from "../utils/paths.ts"; import { type AuthStorageBackend, FileAuthStorageBackend } from "./auth-storage.ts"; type StoredModels = Record; +type ModelsFileReload = { + controller: AbortController; + promise: Promise; + readers: number; +}; + +type ModelsFileReadState = { + data: StoredModels; + revision?: string; + reload?: ModelsFileReload; +}; + +// Optimize the common path without retaining an unbounded set of custom paths. +let sharedModelsFileReadState: { path: string; readState: ModelsFileReadState } | undefined; + export class InMemoryCodingAgentModelsStore implements ModelsStore { private readonly entries = new Map(); @@ -28,35 +45,109 @@ export class InMemoryCodingAgentModelsStore implements ModelsStore { /** Locked JSON-backed storage for dynamically refreshed provider catalogs. */ export class FileModelsStore implements ModelsStore { private readonly storage: AuthStorageBackend; + private readonly path: string; + private readonly readState: ModelsFileReadState | undefined; constructor(path: string = join(getAgentDir(), "models-store.json")) { - this.storage = new FileAuthStorageBackend(path); + this.path = normalizePath(path); + this.storage = new FileAuthStorageBackend(this.path); + if (!sharedModelsFileReadState) { + this.readState = { data: {} }; + sharedModelsFileReadState = { path: this.path, readState: this.readState }; + } else if (sharedModelsFileReadState.path === this.path) { + this.readState = sharedModelsFileReadState.readState; + } } private parse(content: string | undefined): StoredModels { return content ? (JSON.parse(content) as StoredModels) : {}; } + private updateReadState(readState: ModelsFileReadState, data: StoredModels, revision?: string): void { + readState.data = data; + readState.revision = revision; + } + + private reloadFromStorage( + readState: ModelsFileReadState, + options?: ModelsStoreOperationOptions, + ): Promise { + return this.storage.withLockAsync(async (content) => { + const data = this.parse(content); + this.updateReadState(readState, data, getFileRevision(this.path)); + return { result: data }; + }, options); + } + + private async readLatest( + readState: ModelsFileReadState, + options?: ModelsStoreOperationOptions, + ): Promise { + options?.signal?.throwIfAborted(); + const revision = getFileRevision(this.path); + if (revision !== undefined && revision === readState.revision) return readState.data; + if (!readState.reload) { + const controller = new AbortController(); + const reload: ModelsFileReload = { + controller, + promise: this.reloadFromStorage(readState, { signal: controller.signal }), + readers: 0, + }; + readState.reload = reload; + void reload.promise.then( + () => { + if (readState.reload === reload) readState.reload = undefined; + }, + () => { + if (readState.reload === reload) readState.reload = undefined; + }, + ); + } + + const reload = readState.reload; + reload.readers++; + try { + return await raceWithAbortSignal(reload.promise, options?.signal); + } finally { + reload.readers--; + if (reload.readers === 0 && readState.reload === reload) { + readState.reload = undefined; + reload.controller.abort(); + } + } + } + async read(providerId: string, options?: ModelsStoreOperationOptions): Promise { - return this.storage.withLockAsync( - async (content) => ({ result: structuredClone(this.parse(content)[providerId]) }), - options, - ); + if (!this.readState) { + return this.storage.withLockAsync( + async (content) => ({ result: structuredClone(this.parse(content)[providerId]) }), + options, + ); + } + const entry = (await this.readLatest(this.readState, options))[providerId]; + options?.signal?.throwIfAborted(); + return entry ? structuredClone(entry) : undefined; } async write(providerId: string, entry: ModelsStoreEntry, options?: ModelsStoreOperationOptions): Promise { + let latest: StoredModels | undefined; await this.storage.withLockAsync(async (content) => { const current = this.parse(content); current[providerId] = structuredClone(entry); + latest = current; return { result: undefined, next: JSON.stringify(current, null, 2) }; }, options); + if (this.readState && latest) this.updateReadState(this.readState, latest); } async delete(providerId: string, options?: ModelsStoreOperationOptions): Promise { + let latest: StoredModels | undefined; await this.storage.withLockAsync(async (content) => { const current = this.parse(content); delete current[providerId]; + latest = current; return { result: undefined, next: JSON.stringify(current, null, 2) }; }, options); + if (this.readState && latest) this.updateReadState(this.readState, latest); } } diff --git a/packages/coding-agent/src/utils/paths.ts b/packages/coding-agent/src/utils/paths.ts index 7f1013446..5e6533c48 100644 --- a/packages/coding-agent/src/utils/paths.ts +++ b/packages/coding-agent/src/utils/paths.ts @@ -1,4 +1,4 @@ -import { realpathSync } from "node:fs"; +import { realpathSync, statSync } from "node:fs"; import { homedir } from "node:os"; import { isAbsolute, join, resolve as nodeResolvePath, relative, sep } from "node:path"; import { fileURLToPath } from "node:url"; @@ -33,6 +33,15 @@ export function canonicalizePath(path: string): string { } } +export function getFileRevision(path: string): string | undefined { + try { + const stats = statSync(path, { bigint: true }); + return `${stats.dev}:${stats.ino}:${stats.size}:${stats.mtimeNs}:${stats.ctimeNs}`; + } catch { + return undefined; + } +} + /** * Returns true if the value is NOT a package source (npm:, git:, etc.) * or a remote URL protocol. Bare names, relative paths, and file: URLs diff --git a/packages/coding-agent/test/auth-storage.test.ts b/packages/coding-agent/test/auth-storage.test.ts index 67cb5edf3..8736562b8 100644 --- a/packages/coding-agent/test/auth-storage.test.ts +++ b/packages/coding-agent/test/auth-storage.test.ts @@ -7,13 +7,12 @@ import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; import { AuthStorage, FileAuthStorageBackend } from "../src/core/auth-storage.ts"; describe("AuthStorage", () => { - let tempDir: string; - let authJsonPath: string; + const tempDir = join(tmpdir(), `pi-test-auth-storage-${Date.now()}-${Math.random().toString(36).slice(2)}`); + const authJsonPath = join(tempDir, "auth.json"); beforeEach(() => { - tempDir = join(tmpdir(), `pi-test-auth-storage-${Date.now()}-${Math.random().toString(36).slice(2)}`); + if (existsSync(tempDir)) rmSync(tempDir, { recursive: true }); mkdirSync(tempDir, { recursive: true }); - authJsonPath = join(tempDir, "auth.json"); }); afterEach(() => { @@ -97,9 +96,21 @@ describe("AuthStorage", () => { await expect(second.read("anthropic")).resolves.toEqual({ type: "api_key", key: "new" }); expect(lockSpy).toHaveBeenCalledTimes(1); + const otherPath = join(tempDir, "other-auth.json"); + writeFileSync(otherPath, JSON.stringify({ other: { type: "api_key", key: "other-key" } })); + const otherFirst = AuthStorage.create(otherPath); + const otherSecond = AuthStorage.create(otherPath); + await otherFirst.read("other"); + await otherSecond.read("other"); + await otherFirst.list(); + expect(lockSpy).toHaveBeenCalledTimes(4); + + const third = AuthStorage.create(authJsonPath); writeAuthJson({ anthropic: { type: "api_key", key: "newest" } }); - await expect(first.read("anthropic")).resolves.toEqual({ type: "api_key", key: "newest" }); - expect(lockSpy).toHaveBeenCalledTimes(2); + const [firstReload, thirdReload] = await Promise.all([first.read("anthropic"), third.read("anthropic")]); + expect(firstReload).toEqual({ type: "api_key", key: "newest" }); + expect(thirdReload).toEqual({ type: "api_key", key: "newest" }); + expect(lockSpy).toHaveBeenCalledTimes(5); }); test("modify persists a credential while preserving unrelated external edits", async () => { diff --git a/packages/coding-agent/test/models-store.test.ts b/packages/coding-agent/test/models-store.test.ts index 2d42211e1..975309cfe 100644 --- a/packages/coding-agent/test/models-store.test.ts +++ b/packages/coding-agent/test/models-store.test.ts @@ -3,15 +3,22 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import type { Model } from "@earendil-works/pi-ai"; import lockfile from "proper-lockfile"; -import { afterEach, describe, expect, it } from "vitest"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { FileModelsStore } from "../src/core/models-store.ts"; -const tempDirs: string[] = []; +const sharedTempDir = join(tmpdir(), `pi-models-store-shared-${Date.now()}-${Math.random().toString(36).slice(2)}`); +const sharedModelsPath = join(sharedTempDir, "models-store.json"); + +beforeAll(() => { + mkdirSync(sharedTempDir, { recursive: true }); +}); + +afterAll(() => { + if (existsSync(sharedTempDir)) rmSync(sharedTempDir, { recursive: true }); +}); afterEach(() => { - for (const path of tempDirs.splice(0)) { - if (existsSync(path)) rmSync(path, { recursive: true }); - } + vi.restoreAllMocks(); }); function model(provider: string, id: string): Model<"openai-completions"> { @@ -31,16 +38,12 @@ function model(provider: string, id: string): Model<"openai-completions"> { describe("FileModelsStore", () => { it("persists provider catalogs without replacing unrelated providers", async () => { - const dir = join(tmpdir(), `pi-models-store-${Date.now()}-${Math.random().toString(36).slice(2)}`); - tempDirs.push(dir); - mkdirSync(dir, { recursive: true }); - const path = join(dir, "models-store.json"); - const store = new FileModelsStore(path); + const store = new FileModelsStore(sharedModelsPath); await store.write("one", { models: [model("one", "m1")], checkedAt: 100 }); await store.write("two", { models: [model("two", "m2")], checkedAt: 200 }); - const reloaded = new FileModelsStore(path); + const reloaded = new FileModelsStore(sharedModelsPath); expect((await reloaded.read("one"))?.models.map((entry) => entry.id)).toEqual(["m1"]); expect((await reloaded.read("one"))?.checkedAt).toBe(100); expect((await reloaded.read("two"))?.models.map((entry) => entry.id)).toEqual(["m2"]); @@ -50,14 +53,73 @@ describe("FileModelsStore", () => { expect((await reloaded.read("two"))?.models.map((entry) => entry.id)).toEqual(["m2"]); }); + it("coalesces file reloads across concurrent readers and interleaved storage instances", async () => { + writeFileSync( + sharedModelsPath, + JSON.stringify({ + one: { models: [model("one", "old")] }, + two: { models: [model("two", "m2")] }, + }), + ); + const first = new FileModelsStore(sharedModelsPath); + const second = new FileModelsStore(sharedModelsPath); + const lockSpy = vi.spyOn(lockfile, "lock"); + + const [one, two, missing] = await Promise.all([ + first.read("one", { signal: new AbortController().signal }), + second.read("two", { signal: new AbortController().signal }), + first.read("missing", { signal: new AbortController().signal }), + ]); + expect(one?.models.map((entry) => entry.id)).toEqual(["old"]); + expect(two?.models.map((entry) => entry.id)).toEqual(["m2"]); + expect(missing).toBeUndefined(); + expect(lockSpy).toHaveBeenCalledTimes(1); + + await expect(second.read("one")).resolves.toMatchObject({ models: [{ id: "old" }] }); + expect(lockSpy).toHaveBeenCalledTimes(1); + + const otherPath = join(sharedTempDir, "other-models-store.json"); + writeFileSync(otherPath, "{}"); + const other = new FileModelsStore(otherPath); + await expect(other.read("one")).resolves.toBeUndefined(); + await expect(other.read("one")).resolves.toBeUndefined(); + const third = new FileModelsStore(sharedModelsPath); + writeFileSync(sharedModelsPath, JSON.stringify({ one: { models: [model("one", "newest-model")] } })); + const [firstReload, thirdReload] = await Promise.all([first.read("one"), third.read("one")]); + expect(firstReload).toMatchObject({ models: [{ id: "newest-model" }] }); + expect(thirdReload).toMatchObject({ models: [{ id: "newest-model" }] }); + expect(lockSpy).toHaveBeenCalledTimes(4); + }); + + it("keeps a coalesced reload alive while another reader is still waiting", async () => { + writeFileSync(sharedModelsPath, JSON.stringify({ one: { models: [model("one", "stored")] } })); + const store = new FileModelsStore(sharedModelsPath); + let grantLock: (() => void) | undefined; + const lockGranted = new Promise((resolve) => { + grantLock = resolve; + }); + const release = vi.fn(async () => {}); + const lockSpy = vi.spyOn(lockfile, "lock").mockImplementation(async () => { + await lockGranted; + return release; + }); + const firstController = new AbortController(); + const secondController = new AbortController(); + const first = store.read("one", { signal: firstController.signal }); + const second = store.read("one", { signal: secondController.signal }); + + firstController.abort(); + await expect(first).rejects.toMatchObject({ name: "AbortError" }); + grantLock?.(); + await expect(second).resolves.toMatchObject({ models: [{ id: "stored" }] }); + expect(lockSpy).toHaveBeenCalledTimes(1); + expect(release).toHaveBeenCalledTimes(1); + }); + it("cancels a catalog write waiting for a held file lock without writing later", async () => { - const dir = join(tmpdir(), `pi-models-store-${Date.now()}-${Math.random().toString(36).slice(2)}`); - tempDirs.push(dir); - mkdirSync(dir, { recursive: true }); - const path = join(dir, "models-store.json"); - writeFileSync(path, JSON.stringify({ one: { models: [model("one", "existing")] } })); - const store = new FileModelsStore(path); - const release = await lockfile.lock(path, { realpath: false }); + writeFileSync(sharedModelsPath, JSON.stringify({ one: { models: [model("one", "existing")] } })); + const store = new FileModelsStore(sharedModelsPath); + const release = await lockfile.lock(sharedModelsPath, { realpath: false }); const controller = new AbortController(); const pending = store.write("two", { models: [model("two", "cancelled")] }, { signal: controller.signal }); @@ -67,7 +129,7 @@ describe("FileModelsStore", () => { await release(); await new Promise((resolve) => setTimeout(resolve, 150)); - const stored = JSON.parse(readFileSync(path, "utf8")) as Record; + const stored = JSON.parse(readFileSync(sharedModelsPath, "utf8")) as Record; expect(stored.one).toBeDefined(); expect(stored.two).toBeUndefined(); });