authorgravatar for 92423096+nateEc@users.noreply.github.comYukun Shan <92423096+nateEc@users.noreply.github.com> 2026-09-02 08:43:42+08:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-01 17:43:42-07:00
log9a7b1e21e51609266adf657bcab0b43b6bcd445c
tree41cc333d213fe35f0dbd2a5e0fba1763a7651d5d
parent98725df00a729ba226ae8610a3488e5b9168029f
signature Signed by PGP key B5690EEEBB952194

perf(provider): bound persisted session lookups (#8909)

2 files changed, 71 insertions(+), 17 deletions(-)

apps/server/src/provider/Layers/ProviderService.test.ts+56-2
......@@ -284,7 +284,11 @@ const hasMetricSnapshot = (
284284 Object.entries(attributes).every(([key, value]) => snapshot.attributes?.[key] === value),
285285 );
286286
287function makeProviderServiceLayer() {
287function makeProviderServiceLayer(
288 input: {
289 readonly directory?: ProviderSessionDirectory.ProviderSessionDirectory["Service"];
290 } = {},
291) {
288292 const codex = makeFakeCodexAdapter();
289293 const claude = makeFakeCodexAdapter(CLAUDE_AGENT_DRIVER);
290294 const cursor = makeFakeCodexAdapter(CURSOR_DRIVER);
......@@ -301,7 +305,10 @@ function makeProviderServiceLayer() {
301305 const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe(
302306 Layer.provide(SqlitePersistenceMemory),
303307 );
304 const directoryLayer = ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer));
308 const directoryLayer =
309 input.directory === undefined
310 ? ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer))
311 : Layer.succeed(ProviderSessionDirectory.ProviderSessionDirectory, input.directory);
305312
306313 const layer = it.layer(
307314 Layer.mergeAll(
......@@ -2208,6 +2215,53 @@ validation.layer("ProviderServiceLive validation", (it) => {
22082215 );
22092216});
22102217
2218const activeSessionThreadId = asThreadId("thread-active-session");
2219const historicalSessionThreadId = asThreadId("thread-historical-session");
2220const listThreadIds = vi.fn(() =>
2221 Effect.succeed([activeSessionThreadId, historicalSessionThreadId]),
2222);
2223const getBinding = vi.fn((threadId: ThreadId) =>
2224 Effect.succeed(
2225 Option.some({
2226 threadId,
2227 provider: CODEX_DRIVER,
2228 providerInstanceId: codexInstanceId,
2229 }),
2230 ),
2231);
2232const boundedListing = makeProviderServiceLayer({
2233 directory: {
2234 upsert: () => Effect.void,
2235 getProvider: () => Effect.die("ProviderService.listSessions does not use getProvider"),
2236 getBinding,
2237 listThreadIds,
2238 listBindings: () => Effect.die("ProviderService.listSessions does not use listBindings"),
2239 },
2240});
2241
2242boundedListing.layer("ProviderServiceLive session listing", (it) => {
2243 it.effect("looks up bindings for active sessions without scanning historical threads", () =>
2244 Effect.gen(function* () {
2245 const provider = yield* ProviderService.ProviderService;
2246 yield* boundedListing.codex.startSession({
2247 provider: CODEX_DRIVER,
2248 providerInstanceId: codexInstanceId,
2249 threadId: activeSessionThreadId,
2250 cwd: "/tmp/project-active-session",
2251 runtimeMode: "full-access",
2252 });
2253 listThreadIds.mockClear();
2254 getBinding.mockClear();
2255
2256 const sessions = yield* provider.listSessions();
2257
2258 assert.equal(sessions.length, 1);
2259 assert.equal(listThreadIds.mock.calls.length, 0);
2260 assert.deepEqual(getBinding.mock.calls, [[activeSessionThreadId]]);
2261 }),
2262 );
2263});
2264
22112265describe("agent browser access", () => {
22122266 const revokedThreads: Array<ThreadId> = [];
22132267
apps/server/src/provider/Layers/ProviderService.ts+15-15
......@@ -996,21 +996,21 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
996996 ),
997997 );
998998 const activeSessions = sessionsByProvider.flatMap((sessions) => sessions);
999 const persistedBindings = yield* directory.listThreadIds().pipe(
1000 Effect.flatMap((threadIds) =>
1001 Effect.forEach(
1002 threadIds,
1003 (threadId) =>
1004 directory
1005 .getBinding(threadId)
1006 .pipe(
1007 Effect.orElseSucceed(() =>
1008 Option.none<ProviderSessionDirectory.ProviderRuntimeBinding>(),
1009 ),
1010 ),
1011 { concurrency: "unbounded" },
1012 ),
1013 ),
999 // Only live adapter sessions appear in this response. Resolving every
1000 // historical binding here makes each call scale with the full thread
1001 // history instead of the active session set.
1002 const persistedBindings = yield* Effect.forEach(
1003 [...new Set(activeSessions.map((session) => session.threadId))],
1004 (threadId) =>
1005 directory
1006 .getBinding(threadId)
1007 .pipe(
1008 Effect.orElseSucceed(() =>
1009 Option.none<ProviderSessionDirectory.ProviderRuntimeBinding>(),
1010 ),
1011 ),
1012 { concurrency: "unbounded" },
1013 ).pipe(
10141014 Effect.orElseSucceed(
10151015 () => [] as Array<Option.Option<ProviderSessionDirectory.ProviderRuntimeBinding>>,
10161016 ),