Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion packages/client-runtime/src/state/server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1310,7 +1310,8 @@ export function createServerEnvironmentAtoms<R, E>(
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,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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* () {
Expand All @@ -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(
Expand All @@ -855,6 +861,50 @@ 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("filters runner recipes that the environment cannot prepare", () => {
const runnerAgent = makeAgent({ npx: { package: "@example/acp@1.2.3" } });
const binaryAgent = {
Expand Down
28 changes: 25 additions & 3 deletions packages/provider-acp-registry/src/server/AcpRegistrySupport.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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<AcpRegistryIndex | undefined>(undefined);
// When the network last served registryRef; unset while it holds the disk cache.
const registryFetchedAtMs = yield* Ref.make<number | undefined>(undefined);
const registryRevision = yield* Ref.make(0);
const registrySemaphore = yield* Semaphore.make(1);
const installSemaphore = yield* Semaphore.make(1);
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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,
Expand Down
Loading