diff --git a/packages/client-runtime/src/state/server.ts b/packages/client-runtime/src/state/server.ts index 6b77c2a185fec..35a8a7ed2e956 100644 --- a/packages/client-runtime/src/state/server.ts +++ b/packages/client-runtime/src/state/server.ts @@ -1310,7 +1310,8 @@ export function createServerEnvironmentAtoms( searchAcpRegistry: createEnvironmentRpcQueryAtomFamily(runtime, { label: "environment-data:server:acp-registry:search", tag: WS_METHODS.serverSearchAcpRegistry, - // Reopening Add provider reuses recent results; submitting the same search refreshes it. + // Reopening Add provider reuses recent results. Submitting the same search asks the + // server again, which reuses a Registry index fetched in the last five minutes. staleTimeMs: 5 * 60_000, }), configProjection, diff --git a/packages/provider-acp-registry/src/server/AcpRegistrySupport.test.ts b/packages/provider-acp-registry/src/server/AcpRegistrySupport.test.ts index 7b11b520ea999..7cfc19e6fd37c 100644 --- a/packages/provider-acp-registry/src/server/AcpRegistrySupport.test.ts +++ b/packages/provider-acp-registry/src/server/AcpRegistrySupport.test.ts @@ -15,6 +15,7 @@ import * as Path from "effect/Path"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; +import * as TestClock from "effect/testing/TestClock"; import * as HttpClient from "effect/http/HttpClient"; import * as HttpClientResponse from "effect/http/HttpClientResponse"; import * as ChildProcess from "effect/process/ChildProcess"; @@ -821,7 +822,7 @@ describe("AcpRegistrySupport", () => { ); }); - it.effect("refreshes explicit searches and coalesces concurrent refreshes", () => { + it.effect("searches reuse an index fetched in the last five minutes", () => { const agent = makeAgent({ npx: { package: "@example/acp@1.2.3" } }); let requests = 0; return Effect.gen(function* () { @@ -840,6 +841,11 @@ describe("AcpRegistrySupport", () => { ); expect(requests).toBe(1); + yield* TestClock.adjust("4 minutes"); + yield* resolver.search({ query: "other" }); + expect(requests).toBe(1); + + yield* TestClock.adjust("1 minute"); yield* resolver.search({ query: "example" }); expect(requests).toBe(2); }).pipe( @@ -855,6 +861,94 @@ describe("AcpRegistrySupport", () => { ); }); + it.effect("searches retry the network after falling back to the cached index", () => { + const agent = makeAgent({ npx: { package: "@example/acp@1.2.3" } }); + let requests = 0; + return Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const cacheDir = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "t3-acp-registry-retry-", + }); + yield* fileSystem.makeDirectory(`${cacheDir}/acp-registry`, { recursive: true }); + yield* fileSystem.writeFileString( + `${cacheDir}/acp-registry/registry.json`, + makeRegistry(agent), + ); + const resolver = yield* AcpRegistrySupport.makeAcpRegistryCatalog({ + cacheDir, + toolsDir: `${cacheDir}/tools`, + registryUrl, + }); + const offline = yield* resolver.search({ query: "example" }); + expect(offline.agents.map((result) => result.id)).toEqual([agent.id]); + expect(requests).toBe(1); + + yield* resolver.search({ query: "example" }); + expect(requests).toBe(2); + yield* resolver.search({ query: "example" }); + expect(requests).toBe(2); + }).pipe( + Effect.scoped, + Effect.provide( + layerResolver((request) => { + requests += 1; + return Effect.succeed( + HttpClientResponse.fromWeb( + request, + requests === 1 + ? new Response("unavailable", { status: 503 }) + : new Response(makeRegistry(agent)), + ), + ); + }), + ), + ); + }); + + it.effect("searches retry the network after a refresh falls back within the reuse window", () => { + const agent = makeAgent({ npx: { package: "@example/acp@1.2.3" } }); + let requests = 0; + let online = true; + return Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const cacheDir = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "t3-acp-registry-refresh-fallback-", + }); + const resolver = yield* AcpRegistrySupport.makeAcpRegistryCatalog({ + cacheDir, + toolsDir: `${cacheDir}/tools`, + registryUrl, + }); + yield* resolver.search({ query: "example" }); + expect(requests).toBe(1); + + // Prepare always refreshes; an unknown agent stops it right after the refresh. + online = false; + const missing = yield* resolver.prepare({ agentId: "missing-agent" }).pipe(Effect.flip); + expect(missing.reason).toBe("agent_not_found"); + expect(requests).toBe(2); + + online = true; + yield* resolver.search({ query: "example" }); + expect(requests).toBe(3); + }).pipe( + Effect.scoped, + Effect.provide( + layerResolver((request) => { + requests += 1; + return Effect.succeed( + HttpClientResponse.fromWeb( + request, + online + ? new Response(makeRegistry(agent)) + : new Response("unavailable", { status: 503 }), + ), + ); + }), + ), + ); + }); + it.effect("filters runner recipes that the environment cannot prepare", () => { const runnerAgent = makeAgent({ npx: { package: "@example/acp@1.2.3" } }); const binaryAgent = { diff --git a/packages/provider-acp-registry/src/server/AcpRegistrySupport.ts b/packages/provider-acp-registry/src/server/AcpRegistrySupport.ts index d97c110d8611e..1d333ff2da6c0 100644 --- a/packages/provider-acp-registry/src/server/AcpRegistrySupport.ts +++ b/packages/provider-acp-registry/src/server/AcpRegistrySupport.ts @@ -55,6 +55,8 @@ const MAX_PACKAGE_MANIFEST_BYTES = 1024 * 1024; // Older clients omit the limit and validate responses against a 20-agent cap. const DEFAULT_SEARCH_LIMIT = 20; const REGISTRY_REQUEST_TIMEOUT = "30 seconds"; +// Searches reuse an index fetched this recently; prepare always fetches the latest. +const REGISTRY_SEARCH_MAX_AGE_MS = 5 * 60 * 1_000; const ARCHIVE_REQUEST_TIMEOUT = "20 minutes"; const PACKAGE_INSTALL_TIMEOUT = "20 minutes"; const PACKAGE_QUERY_TIMEOUT = "30 seconds"; @@ -659,6 +661,8 @@ export const makeAcpRegistryCatalog = Effect.fn("AcpRegistryCatalog.make")(funct : path.join(packageInstallRoot(agent, distribution), "bin"); const packageReceiptsDirectory = path.join(registryDirectory, "package-installs"); const registryRef = yield* Ref.make(undefined); + // When the network last served registryRef; unset while it holds the disk cache. + const registryFetchedAtMs = yield* Ref.make(undefined); const registryRevision = yield* Ref.make(0); const registrySemaphore = yield* Semaphore.make(1); const installSemaphore = yield* Semaphore.make(1); @@ -848,25 +852,43 @@ export const makeAcpRegistryCatalog = Effect.fn("AcpRegistryCatalog.make")(funct const current = yield* Ref.get(registryRef); if (current !== undefined && currentRevision !== observedRevision) return current; - const registry = yield* fetchRegistry().pipe( + const { registry, fetched } = yield* fetchRegistry().pipe( + Effect.map((registry) => ({ registry, fetched: true })), Effect.catch((networkError) => readCachedRegistry.pipe( Effect.flatMap( Option.match({ onNone: () => Effect.fail(networkError), - onSome: Effect.succeed, + onSome: (registry) => Effect.succeed({ registry, fetched: false }), }), ), ), ), ); yield* Ref.set(registryRef, registry); + yield* Ref.set(registryFetchedAtMs, fetched ? yield* Clock.currentTimeMillis : undefined); yield* Ref.update(registryRevision, (revision) => revision + 1); return registry; }), ); }); + const recentRegistry = Effect.fn("AcpRegistryCatalog.recentRegistry")(function* () { + const current = yield* Ref.get(registryRef); + const fetchedAtMs = yield* Ref.get(registryFetchedAtMs); + const now = yield* Clock.currentTimeMillis; + // A clock moved back makes the stamp untrustworthy, so it counts as stale. + if ( + current !== undefined && + fetchedAtMs !== undefined && + now >= fetchedAtMs && + now - fetchedAtMs < REGISTRY_SEARCH_MAX_AGE_MS + ) { + return current; + } + return yield* refreshRegistry(); + }); + const loadCachedRegistry = Effect.fn("AcpRegistryCatalog.loadCachedRegistry")(function* () { const current = yield* Ref.get(registryRef); if (current !== undefined) return current; @@ -1589,7 +1611,7 @@ export const makeAcpRegistryCatalog = Effect.fn("AcpRegistryCatalog.make")(funct const search: AcpRegistryCatalog["Service"]["search"] = Effect.fn("AcpRegistryCatalog.search")( function* (input) { - const registry = yield* refreshRegistry(); + const registry = yield* recentRegistry(); const ranked = registry.agents.flatMap((agent) => { const distribution = resolveAcpRegistryDistribution({ agent,