From c57071a1cecfa7d7ae1681017f7fc839db96a5a9 Mon Sep 17 00:00:00 2001 From: bk-agent-01 Date: Wed, 21 Oct 2026 14:45:33 +0000 Subject: [PATCH 1/2] fix(server): account policies no longer freeze the server reading the shell On bkt3 (2026-10-21) the server logged 98 event-loop stalls in 24 h once the Claude accounts hit their limits. ForkProviderSessions.listSessions read the full shell (active and archived) on the event loop and then one record read per thread, about 2.2 s, and maybeRebalance called it at least twice at every turn end of a thread on a full account before checking for room. - listSessions reads only the threads live sessions serve, from a new in-memory ProviderSessionManager.listLive (fork-marked, optional). - maybeRebalance checks for room elsewhere before any session read. - A released Claude session also settles its process-bound subagents when a replacement session already serves the thread (the hard-limit path places the replacement before the old release finishes; 9 cards stayed running). Co-Authored-By: Claude Opus 5.5 --- .../ClaudeAccountsService.test.ts | 70 ++++++- .../claudeAccounts/ClaudeAccountsService.ts | 12 +- ...nManager.releasedSubagents.expbkt3.test.ts | 126 ++++++++++-- .../ProviderSessionManager.ts | 23 +++ .../releasedSessionSubagents.expbkt3.ts | 70 +++++-- .../ForkProviderSessions.expbkt3.test.ts | 180 ++++++++++++++++++ .../provider/ForkProviderSessions.expbkt3.ts | 73 +++++-- 7 files changed, 492 insertions(+), 62 deletions(-) create mode 100644 apps/server/src/provider/ForkProviderSessions.expbkt3.test.ts diff --git a/apps/server/src/claudeAccounts/ClaudeAccountsService.test.ts b/apps/server/src/claudeAccounts/ClaudeAccountsService.test.ts index da2e70c1dfd4..ee05cfd5ad1a 100644 --- a/apps/server/src/claudeAccounts/ClaudeAccountsService.test.ts +++ b/apps/server/src/claudeAccounts/ClaudeAccountsService.test.ts @@ -201,6 +201,8 @@ const makeHarness = (options: HarnessOptions = {}) => const place = options.place ?? spreadPlace(status); const enabled = yield* Ref.make(options.enabled ?? true); const sessions = yield* Ref.make>([]); + /** How often the service listed the provider sessions, each a projection read in production. */ + const listSessionsCalls = yield* Ref.make(0); /** Threads whose live session still runs background work (a subagent) between turns. */ const backgroundWork = yield* Ref.make>(new Set()); const shellSession = yield* Ref.make(null); @@ -289,7 +291,10 @@ const makeHarness = (options: HarnessOptions = {}) => }), hasLiveBackgroundWork: ({ threadId }) => Ref.get(backgroundWork).pipe(Effect.map((threads) => threads.has(threadId))), - listSessions: () => Ref.get(sessions), + listSessions: () => + Ref.update(listSessionsCalls, (count) => count + 1).pipe( + Effect.andThen(Ref.get(sessions)), + ), streamEvents: Stream.fromSubscription(providerSubscription), }), ), @@ -333,6 +338,7 @@ const makeHarness = (options: HarnessOptions = {}) => dispatched, enabled, sessions, + listSessionsCalls, backgroundWork, shellSession, shellOwner, @@ -2792,6 +2798,68 @@ describe("ClaudeAccountsService rebalancing at turn boundaries", () => { ); } + { + const gate = gatedPlace(); + // tushar takes no new session, and every other account is at its fair share. + const fullStatus = (dir: DirFor) => ({ + ...switcherStatus([ + switcherProfile(dir, "tushar", { sessions: 3, fair_share: null, guard_ok: false, load: 3 }), + switcherProfile(dir, "agent", { + rank: 2, + elected: false, + sessions: 2, + fair_share: 2, + load: 2, + }), + switcherProfile(dir, "barsha", { + rank: 3, + elected: false, + sessions: 2, + fair_share: 2, + load: 2, + }), + ]), + place_rule: "space-share-v2", + }); + scenario( + "reads no sessions at a turn end on a full account while no other account has room", + { place: gate.place, status: fullStatus }, + (harness, service) => + Effect.gen(function* () { + yield* service.refreshStatus(); + const threadId = ThreadId.make("thread-rebalance-no-room"); + gate.state.open = ["tushar"]; + yield* service.resolveForSession(threadId); + gate.state.open = ["agent"]; + yield* Ref.set(harness.sessions, [liveSession(threadId, "ready")]); + const reads = yield* Ref.get(harness.listSessionsCalls); + const placements = harness.placeCalls.length; + + yield* endTurn(harness, threadId); + yield* drainDomainEvents(harness, service); + assert.equal(yield* Ref.get(harness.listSessionsCalls), reads); + assert.equal(harness.placeCalls.length, placements); + assert.deepEqual(harness.stoppedThreads, []); + assert.equal((yield* service.getThread(threadId)).resolvedProfile, "tushar"); + + // Once agent has room, the next turn end moves the thread as before. + yield* Ref.update(harness.status, (current) => ({ + ...current, + profiles: current.profiles.map((profile) => + profile.name === "agent" ? { ...profile, sessions: 0, load: 0 } : profile, + ), + })); + yield* service.refreshStatus(); + yield* endTurn(harness, threadId); + yield* drainDomainEvents(harness, service); + assert.isAbove(yield* Ref.get(harness.listSessionsCalls), reads); + assert.include(harness.placeCalls.at(-1)!.avoid, "tushar"); + assert.deepEqual(harness.stoppedThreads, [threadId]); + assert.equal((yield* service.getThread(threadId)).resolvedProfile, "agent"); + }), + ); + } + { const gate = gatedPlace(); scenario( diff --git a/apps/server/src/claudeAccounts/ClaudeAccountsService.ts b/apps/server/src/claudeAccounts/ClaudeAccountsService.ts index a0c36822d2b7..5b29ebb904fd 100644 --- a/apps/server/src/claudeAccounts/ClaudeAccountsService.ts +++ b/apps/server/src/claudeAccounts/ClaudeAccountsService.ts @@ -2088,11 +2088,9 @@ export const make = Effect.gen(function* () { if (!(yield* readSettings).enabled) return; const row = yield* repository.get(threadId); if (row.mode.kind !== "auto" || row.resolvedProfile !== current) return; - const live = (yield* providers.listSessions()).some( - (session) => session.threadId === threadId && session.status !== "closed", - ); - if (!live || !(yield* threadIsIdle(threadId, settled))) return; - // Without a subprocess: is there anywhere better at all? + // Without a subprocess or a session read: is there anywhere better at + // all? While every account is full, each turn end on a full one would + // otherwise read the sessions for nothing. yield* pruneExhausted; const { disallowed } = yield* disallowedFor(threadId); const pending = yield* pendingCounts; @@ -2108,6 +2106,10 @@ export const make = Effect.gen(function* () { (pending[profile.name] ?? 0), ); if (!roomElsewhere) return; + const live = (yield* providers.listSessions()).some( + (session) => session.threadId === threadId && session.status !== "closed", + ); + if (!live || !(yield* threadIsIdle(threadId, settled))) return; const target = yield* Semaphore.withPermit(placementLock)( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.releasedSubagents.expbkt3.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.releasedSubagents.expbkt3.test.ts index 5743e803b28d..9ff1d790d5ea 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.releasedSubagents.expbkt3.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.releasedSubagents.expbkt3.test.ts @@ -229,29 +229,34 @@ function fixture( options: { readonly projections?: ReadonlyMap; readonly failRecoveryRead?: boolean; + /** The released session's driver; Claude by default. */ + readonly driver?: ProviderDriverKind; } = {}, ) { const projections = options.projections ?? incidentProjections(); + const sessionDriver = options.driver ?? driver; const written: Array = []; /** Threads whose full recovery read ran, in order. */ const recoveryReads: Array = []; const adapter: ProviderAdapterV2["Service"] = { instanceId, - driver, + driver: sessionDriver, getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2), planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), openSession: (input) => Effect.gen(function* () { const events = yield* Queue.unbounded(); const unused = () => - Effect.fail(new ProviderAdapterProtocolError({ driver, detail: "unused" })); + Effect.fail( + new ProviderAdapterProtocolError({ driver: sessionDriver, detail: "unused" }), + ); return { instanceId, - driver, + driver: sessionDriver, providerSessionId: input.providerSessionId, providerSession: { id: input.providerSessionId, - driver, + driver: sessionDriver, providerInstanceId: instanceId, status: "ready" as const, cwd: "/tmp", @@ -413,19 +418,108 @@ it.effect("a detach that leaves the shared session open settles nothing", () => }).pipe(Effect.provide(f.layer)); }); -it.effect("a thread another live session still serves keeps its subagents", () => { - const f = fixture(); - return Effect.gen(function* () { - const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; - yield* manager.open(f.open(parentThreadId)); - yield* manager.open(f.open(parentThreadId, otherSessionId)); - yield* TestClock.adjust("10 minutes"); - yield* manager.release({ providerSessionId: releasedSessionId, reason: "manual_shutdown" }); +it.effect( + "a replacement Claude session on the thread does not keep the released process's subagents open", + () => { + const f = fixture(); + return Effect.gen(function* () { + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + yield* manager.open(f.open(parentThreadId)); + // The incident: a hard limit stopped the session, and the thread was + // placed on another account before that release finished. + yield* manager.open(f.open(parentThreadId, otherSessionId)); + yield* TestClock.adjust("10 minutes"); + yield* manager.release({ providerSessionId: releasedSessionId, reason: "manual_shutdown" }); - assert.deepEqual(f.cancelled("subagent.updated"), []); - assert.deepEqual(f.cancelled("node.updated"), []); - }).pipe(Effect.provide(f.layer)); -}); + assert.isTrue(Option.isSome(yield* manager.get(otherSessionId))); + // The process-bound subagent died with the released process. + assert.deepEqual(f.cancelled("subagent.updated"), ["subagent-running"]); + assert.sameMembers(f.cancelled("turn-item.updated"), [ + "card-subagent-running", + "child-command", + ]); + assert.sameMembers(f.cancelled("node.updated"), ["subagent-running", "child-root-running"]); + // One bound by its provider thread may run in the live session, and one + // started after the release is the replacement's own. + const touched = f.written.flatMap((event) => + event.type === "subagent.updated" || + event.type === "turn-item.updated" || + event.type === "node.updated" + ? [String(event.payload.id)] + : [], + ); + for (const name of ["other-session", "unloaded-thread", "started-later"]) { + assert.notInclude(touched, `subagent-${name}`); + assert.notInclude(touched, `card-subagent-${name}`); + assert.notInclude(touched, `child-root-${name}`); + } + }).pipe(Effect.provide(f.layer)); + }, +); + +it.effect( + "a subagent started after the release stays open while a replacement serves the thread", + () => { + const replacement = subagent("replacement", { startedAt: later, updatedAt: later }); + const f = fixture({ + projections: new Map([ + [ + parentThreadId, + projection(parentThreadId, { + subagents: [replacement], + turnItems: [subagentItem(replacement)] as never, + nodes: [node(replacement.id, parentThreadId, "subagent", later)], + }), + ], + ]), + }); + return Effect.gen(function* () { + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + yield* manager.open(f.open(parentThreadId)); + yield* manager.open(f.open(parentThreadId, otherSessionId)); + yield* TestClock.adjust("10 minutes"); + yield* manager.release({ providerSessionId: releasedSessionId, reason: "manual_shutdown" }); + + assert.deepEqual(f.cancelled("subagent.updated"), []); + assert.deepEqual(f.cancelled("turn-item.updated"), []); + assert.deepEqual(f.cancelled("node.updated"), []); + // The subagent rows alone showed nothing the released process ran. + assert.deepEqual(f.recoveryReads, []); + }).pipe(Effect.provide(f.layer)); + }, +); + +it.effect( + "a thread another live session still serves keeps its subagents for other drivers", + () => { + const codex = ProviderDriverKind.make("codex"); + const running = subagent("codex-running", { driver: codex }); + const f = fixture({ + driver: codex, + projections: new Map([ + [ + parentThreadId, + projection(parentThreadId, { + subagents: [running], + turnItems: [subagentItem(running)] as never, + nodes: [node(running.id, parentThreadId, "subagent")], + }), + ], + ]), + }); + return Effect.gen(function* () { + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + yield* manager.open(f.open(parentThreadId)); + yield* manager.open(f.open(parentThreadId, otherSessionId)); + yield* TestClock.adjust("10 minutes"); + yield* manager.release({ providerSessionId: releasedSessionId, reason: "manual_shutdown" }); + + assert.deepEqual(f.recoveryReads, []); + assert.deepEqual(f.cancelled("subagent.updated"), []); + assert.deepEqual(f.cancelled("node.updated"), []); + }).pipe(Effect.provide(f.layer)); + }, +); it.effect("a server shutdown leaves the subagents to startup recovery and its wake", () => { const f = fixture(); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 966ad2a3ce39..7ae7980e55a3 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -179,6 +179,18 @@ export interface ProviderSessionManagerV2Shape { ProviderSessionManagerV2Error >; readonly isLive: (providerSessionId: ProviderSessionId) => Effect.Effect; + // T3-CUSTOM(expbkt3): BEGIN the live sessions and the threads each serves, from memory. + /** + * Reads no projection and counts as no activity. Optional so test layers + * that build the full shape need not supply it. + */ + readonly listLive?: () => Effect.Effect< + ReadonlyArray<{ + readonly providerSessionId: ProviderSessionId; + readonly threadIds: ReadonlyArray; + }> + >; + // T3-CUSTOM(expbkt3): END readonly close: ( providerSessionId: ProviderSessionId, ) => Effect.Effect; @@ -2561,6 +2573,17 @@ export const layerWithOptions = ( Ref.get(sessions).pipe( Effect.map((current) => current.has(sessionKey(providerSessionId))), ), + // T3-CUSTOM(expbkt3): BEGIN account policies list live sessions without a shell read. + listLive: () => + Ref.get(sessions).pipe( + Effect.map((current) => + Array.from(current.values(), (entry) => ({ + providerSessionId: entry.runtime.providerSessionId, + threadIds: [...entry.attachedThreadIds], + })), + ), + ), + // T3-CUSTOM(expbkt3): END get: Effect.fn("ProviderSessionManager.get")( function* (providerSessionId) { const entry = (yield* Ref.get(sessions)).get(sessionKey(providerSessionId)); diff --git a/apps/server/src/orchestration-v2/releasedSessionSubagents.expbkt3.ts b/apps/server/src/orchestration-v2/releasedSessionSubagents.expbkt3.ts index 3d4c6ac33729..e20ee260ded7 100644 --- a/apps/server/src/orchestration-v2/releasedSessionSubagents.expbkt3.ts +++ b/apps/server/src/orchestration-v2/releasedSessionSubagents.expbkt3.ts @@ -1,13 +1,13 @@ // T3-CUSTOM(expbkt3): a released provider session settles the subagents it leaves open. -import type { - NodeId, - OrchestrationV2DomainEvent, - OrchestrationV2Subagent, - OrchestrationV2ThreadProjection, +import { + type NodeId, + type OrchestrationV2DomainEvent, + type OrchestrationV2Subagent, + type OrchestrationV2ThreadProjection, ProviderDriverKind, - ProviderInstanceId, - ProviderSessionId, - ThreadId, + type ProviderInstanceId, + type ProviderSessionId, + type ThreadId, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; @@ -21,6 +21,9 @@ import type { ProviderSessionReleaseReason } from "./ProviderSessionManager.ts"; type TurnItem = OrchestrationV2ThreadProjection["turnItems"][number]; type ExecutionNode = OrchestrationV2ThreadProjection["nodes"][number]; +/** Its subagents run inside the parent's process and carry no provider thread. */ +const CLAUDE_DRIVER = ProviderDriverKind.make("claudeAgent"); + export interface ReleasedSessionSubagentsInput { readonly projectionStore: ProjectionStoreV2Shape; readonly idAllocator: IdAllocatorV2["Service"]; @@ -31,7 +34,10 @@ export interface ReleasedSessionSubagentsInput { readonly providerInstanceId: ProviderInstanceId; /** The app threads the released process served. */ readonly threadIds: Iterable; - /** App threads a live session still serves. That session owns their subagents. */ + /** + * App threads a live session still serves. That session owns their + * subagents, except the Claude subagents the released process ran. + */ readonly liveThreadIds: Effect.Effect>; /** Work that started after the release belongs to a later session. */ readonly releasedAt: DateTime.Utc; @@ -66,8 +72,14 @@ const cancelledItem = (item: TurnItem, now: DateTime.Utc): TurnItem => * * Only open `provider_native` subagents of the released driver and instance, * started at or before the release, are settled. A delegate_task child has its - * own runs and is never touched, and neither is a thread a live session still - * serves. A later SendMessage resume reopens a settled subagent: its new events + * own runs and is never touched. A thread another live session still serves + * keeps its subagents, with one exception: a Claude subagent carries no + * provider thread and lives only in the process that started it, so a + * replacement session (placed on another account while this one was still + * releasing) cannot end it either. There only those process-bound subagents + * are settled; one bound to a provider thread may run in the live session. + * A later SendMessage resume reopens a settled subagent (it re-stamps + * `startedAt`, so it counts as started after the release): its new events * overwrite the cancelled status. * * A server shutdown is left to startup recovery, which also wakes a settled @@ -92,9 +104,12 @@ export const settleReleasedSessionSubagents: ( const at = startedAt ?? fallback; return at === null || DateTime.isLessThanOrEqualTo(at, input.releasedAt); }; + /** Another live session serves the threads of a released Claude session too. */ + const settlesServedThreads = input.driver === CLAUDE_DRIVER; const ranInReleasedSession = ( subagent: OrchestrationV2Subagent, projection: ProjectionRuntimeRecoveryState, + servedElsewhere: boolean, ) => { if ( subagent.origin !== "provider_native" || @@ -108,6 +123,8 @@ export const settleReleasedSessionSubagents: ( // Claude subagents carry no provider thread; their parent thread's // attachment is the link. One that does must not name another session. if (subagent.providerThreadId === null) return true; + // The live session may run it on that provider thread. + if (servedElsewhere) return false; const providerThread = projection.providerThreads.find( (candidate) => candidate.id === subagent.providerThreadId, ); @@ -152,7 +169,11 @@ export const settleReleasedSessionSubagents: ( * runless root turn and the items under it. Returns the child threads of the * subagents it settled, whose own work died with the same process. */ - const settleThread = Effect.fnUntraced(function* (threadId: ThreadId, isChild: boolean) { + const settleThread = Effect.fnUntraced(function* ( + threadId: ThreadId, + isChild: boolean, + servedElsewhere: boolean, + ) { // Most releases (idle ones) leave nothing open. The subagent rows alone // are cheap; the recovery read below can take seconds on a cold cache // for a thread with thousands of items, on the event loop. A failed @@ -168,7 +189,11 @@ export const settleReleasedSessionSubagents: ( subagent.origin === "provider_native" && isOpenStatus(subagent.status) && subagent.driver === input.driver && - subagent.providerInstanceId === input.providerInstanceId, + subagent.providerInstanceId === input.providerInstanceId && + // The live session's own subagents need no full read. + (!servedElsewhere || + (subagent.providerThreadId === null && + startedByRelease(subagent.startedAt, subagent.updatedAt))), ) ) { return []; @@ -184,7 +209,7 @@ export const settleReleasedSessionSubagents: ( const settledItemIds = new Set(); const childThreadIds: Array = []; for (const subagent of projection.subagents) { - if (!ranInReleasedSession(subagent, projection)) continue; + if (!ranInReleasedSession(subagent, projection, servedElsewhere)) continue; settledSubagentIds.push(subagent.id); events.push({ id: yield* allocateEventId(subagent.threadId), @@ -252,19 +277,26 @@ export const settleReleasedSessionSubagents: ( }); const visited = new Set(); - let pending: Array<{ readonly threadId: ThreadId; readonly isChild: boolean }> = []; + let pending: Array<{ + readonly threadId: ThreadId; + readonly isChild: boolean; + readonly servedElsewhere: boolean; + }> = []; for (const threadId of input.threadIds) { - if (!liveThreadIds.has(threadId)) pending.push({ threadId, isChild: false }); + const servedElsewhere = liveThreadIds.has(threadId); + if (!servedElsewhere || settlesServedThreads) { + pending.push({ threadId, isChild: false, servedElsewhere }); + } } while (pending.length > 0) { const next: typeof pending = []; - for (const { threadId, isChild } of pending) { + for (const { threadId, isChild, servedElsewhere } of pending) { if (visited.has(threadId)) continue; visited.add(threadId); - const childThreadIds = yield* settleThread(threadId, isChild); + const childThreadIds = yield* settleThread(threadId, isChild, servedElsewhere); for (const childThreadId of childThreadIds) { if (!liveThreadIds.has(childThreadId)) - next.push({ threadId: childThreadId, isChild: true }); + next.push({ threadId: childThreadId, isChild: true, servedElsewhere: false }); } } pending = next; diff --git a/apps/server/src/provider/ForkProviderSessions.expbkt3.test.ts b/apps/server/src/provider/ForkProviderSessions.expbkt3.test.ts new file mode 100644 index 000000000000..88b68b99ac41 --- /dev/null +++ b/apps/server/src/provider/ForkProviderSessions.expbkt3.test.ts @@ -0,0 +1,180 @@ +// T3-CUSTOM(expbkt3): account policies list live sessions without reading the full shell. +import { assert, it } from "@effect/vitest"; +import { + type OrchestrationV2ProviderSession, + type OrchestrationV2ThreadProjection, + ProviderDriverKind, + ProviderInstanceId, + ProviderSessionId, + ThreadId, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; + +import { ProviderSessionManagerV2 } from "../orchestration-v2/ProviderSessionManager.ts"; +import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts"; +import { ForkProviderSessions, layer } from "./ForkProviderSessions.expbkt3.ts"; + +const instanceId = ProviderInstanceId.make("claudeAgent"); +const shared = ProviderSessionId.make("session-shared"); +const solo = ProviderSessionId.make("session-solo"); +const stopped = ProviderSessionId.make("session-stopped"); +const first = ThreadId.make("thread-first"); +const second = ThreadId.make("thread-second"); +const archived = ThreadId.make("thread-archived"); +const deleted = ThreadId.make("thread-deleted"); +const missing = ThreadId.make("thread-missing"); +const createdAt = DateTime.makeUnsafe("2026-10-21T10:00:00Z"); +const updatedAt = DateTime.makeUnsafe("2026-10-21T10:05:00Z"); + +function providerSession( + id: ProviderSessionId, + overrides: Partial = {}, +): OrchestrationV2ProviderSession { + return { + id, + driver: ProviderDriverKind.make("claudeAgent"), + providerInstanceId: instanceId, + status: "ready", + cwd: "/repo", + model: "claude-opus-5-5", + capabilities: {}, + createdAt, + updatedAt, + lastError: null, + ...overrides, + } as unknown as OrchestrationV2ProviderSession; +} + +function records( + threadId: ThreadId, + providerSessions: ReadonlyArray, + thread: { + readonly runtimeMode?: string; + readonly archived?: boolean; + readonly deleted?: boolean; + }, +): OrchestrationV2ThreadProjection { + return { + thread: { + id: threadId, + runtimeMode: thread.runtimeMode ?? "full-access", + archivedAt: thread.archived === true ? updatedAt : null, + deletedAt: thread.deleted === true ? updatedAt : null, + }, + providerSessions, + } as unknown as OrchestrationV2ThreadProjection; +} + +/** + * Two live sessions: one shared by two threads, one serving an archived + * thread, a deleted thread and a thread with no projection. + */ +function fixture() { + const byThread = new Map([ + [ + first, + records( + first, + // A session the thread used before; it is no longer live. + [ + providerSession(stopped, { status: "stopped" }), + providerSession(shared, { status: "waiting" }), + ], + {}, + ), + ], + [ + second, + records( + second, + [ + providerSession(shared, { + status: "starting", + model: null, + driver: ProviderDriverKind.make("claude"), + }), + ], + { runtimeMode: "approval-required" }, + ), + ], + [archived, records(archived, [providerSession(solo)], { archived: true })], + [deleted, records(deleted, [providerSession(solo)], { deleted: true })], + ]); + const reads = { shell: 0, threads: [] as Array }; + const testLayer = layer.pipe( + Layer.provide( + // `get` stays unimplemented: listing sessions must not count as activity on them. + Layer.mock(ProviderSessionManagerV2)({ + listLive: () => + Effect.succeed([ + { providerSessionId: shared, threadIds: [first, second] }, + { providerSessionId: solo, threadIds: [archived, deleted, missing] }, + ]), + }), + ), + Layer.provide( + Layer.mock(ProjectionStore.ProjectionStoreV2)({ + getShellSnapshot: () => + Effect.sync(() => { + reads.shell += 1; + }).pipe(Effect.andThen(Effect.die("the full shell must not be read"))), + getThreadRecords: (threadId) => { + reads.threads.push(threadId); + const found = byThread.get(threadId); + return found === undefined + ? Effect.fail(new ProjectionStore.ProjectionStoreThreadNotFoundError({ threadId })) + : Effect.succeed(found); + }, + }), + ), + ); + return { layer: testLayer, reads }; +} + +it.effect("lists the live sessions from the threads they serve, without the full shell", () => { + const f = fixture(); + return Effect.gen(function* () { + const providers = yield* ForkProviderSessions; + const sessions = yield* providers.listSessions(); + + assert.deepEqual(sessions, [ + { + threadId: first, + provider: ProviderDriverKind.make("claudeAgent"), + providerInstanceId: instanceId, + status: "running", + runtimeMode: "full-access", + cwd: "/repo", + model: "claude-opus-5-5", + createdAt: "2026-10-21T10:00:00.000Z", + updatedAt: "2026-10-21T10:05:00.000Z", + }, + { + threadId: second, + provider: ProviderDriverKind.make("claudeAgent"), + providerInstanceId: instanceId, + status: "connecting", + runtimeMode: "approval-required", + cwd: "/repo", + createdAt: "2026-10-21T10:00:00.000Z", + updatedAt: "2026-10-21T10:05:00.000Z", + }, + { + threadId: archived, + provider: ProviderDriverKind.make("claudeAgent"), + providerInstanceId: instanceId, + status: "ready", + runtimeMode: "full-access", + cwd: "/repo", + model: "claude-opus-5-5", + createdAt: "2026-10-21T10:00:00.000Z", + updatedAt: "2026-10-21T10:05:00.000Z", + }, + ]); + assert.equal(f.reads.shell, 0); + // Only the threads a live session serves, each once. + assert.deepEqual(f.reads.threads, [first, second, archived, deleted, missing]); + }).pipe(Effect.provide(f.layer)); +}); diff --git a/apps/server/src/provider/ForkProviderSessions.expbkt3.ts b/apps/server/src/provider/ForkProviderSessions.expbkt3.ts index 7de336151773..8e7db6296ef6 100644 --- a/apps/server/src/provider/ForkProviderSessions.expbkt3.ts +++ b/apps/server/src/provider/ForkProviderSessions.expbkt3.ts @@ -1,6 +1,7 @@ // T3-CUSTOM(expbkt3): account policies observe and stop native V2 sessions. // This facade owns no execution state; V2 remains the source of truth. import { + type OrchestrationV2ProviderSession, ProviderDriverKind, type ProviderRuntimeEvent, type ProviderSession, @@ -16,6 +17,31 @@ import * as Stream from "effect/Stream"; import { ProviderSessionManagerV2 } from "../orchestration-v2/ProviderSessionManager.ts"; import { ProjectionStoreV2 } from "../orchestration-v2/ProjectionStore.ts"; +/** One live provider session, as the thread it serves sees it. */ +const toProviderSession = ( + thread: { readonly id: ThreadId; readonly runtimeMode: ProviderSession["runtimeMode"] }, + session: OrchestrationV2ProviderSession, +): ProviderSession => ({ + threadId: thread.id, + provider: ProviderDriverKind.make( + String(session.driver) === "claude" ? "claudeAgent" : String(session.driver), + ), + providerInstanceId: session.providerInstanceId, + status: + session.status === "stopped" + ? "closed" + : session.status === "starting" + ? "connecting" + : session.status === "waiting" + ? "running" + : session.status, + runtimeMode: thread.runtimeMode, + cwd: session.cwd, + ...(session.model === null ? {} : { model: session.model }), + createdAt: DateTime.formatIso(session.createdAt), + updatedAt: DateTime.formatIso(session.updatedAt), +}); + const listeners = new Set<(event: ProviderRuntimeEvent) => Effect.Effect>(); export const emitForkProviderEvent = (event: ProviderRuntimeEvent): Effect.Effect => Effect.forEach(listeners, (listener) => listener(event), { discard: true }); @@ -60,6 +86,7 @@ export const layer = Layer.effect( ForkProviderSessions, Effect.gen(function* () { const sessions = yield* ProviderSessionManagerV2; + const listLive = sessions.listLive; const projection = yield* ProjectionStoreV2; const events = yield* PubSub.unbounded(); const listener = (event: ProviderRuntimeEvent) => @@ -104,33 +131,37 @@ export const layer = Layer.effect( ), ), listSessions: Effect.fn("ForkProviderSessions.listSessions")(function* () { - const shell = yield* projection.getShellSnapshot(); const result: ProviderSession[] = []; + if (listLive !== undefined) { + // Only the threads a live session serves are read. The full shell + // (every active and archived thread) blocked the event loop for + // seconds, and account policies list sessions at every turn end. + for (const live of yield* listLive()) { + for (const threadId of live.threadIds) { + const records = yield* projection + .getThreadRecords(threadId, ["providerSessions"]) + .pipe( + Effect.catchTags({ + ProjectionStoreThreadNotFoundError: () => Effect.succeed(null), + }), + ); + // The shell never lists a deleted thread. + if (records === null || records.thread.deletedAt !== null) continue; + const session = records.providerSessions.find( + (candidate) => candidate.id === live.providerSessionId, + ); + if (session !== undefined) result.push(toProviderSession(records.thread, session)); + } + } + return result; + } + const shell = yield* projection.getShellSnapshot(); for (const thread of [...shell.threads, ...shell.archivedThreads]) { const records = yield* projection.getThreadRecords(thread.id, ["providerSessions"]); for (const session of records.providerSessions) { const live = yield* sessions.get(session.id); if (live._tag === "None") continue; - result.push({ - threadId: thread.id, - provider: ProviderDriverKind.make( - String(session.driver) === "claude" ? "claudeAgent" : String(session.driver), - ), - providerInstanceId: session.providerInstanceId, - status: - session.status === "stopped" - ? "closed" - : session.status === "starting" - ? "connecting" - : session.status === "waiting" - ? "running" - : session.status, - runtimeMode: thread.runtimeMode, - cwd: session.cwd, - ...(session.model === null ? {} : { model: session.model }), - createdAt: DateTime.formatIso(session.createdAt), - updatedAt: DateTime.formatIso(session.updatedAt), - }); + result.push(toProviderSession(thread, session)); } } return result; From 0c1f55d4f24748133373dad0b1c197bf0d6a4843 Mon Sep 17 00:00:00 2001 From: bk-agent-01 Date: Wed, 21 Oct 2026 14:55:38 +0000 Subject: [PATCH 2/2] fix(server): warn when listSessions falls back to the full shell read Review follow-up: a session manager without listLive (a wrapper or a Layer.mock) silently brought back the blocking full-shell path. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/provider/ForkProviderSessions.expbkt3.ts | 3 +++ 1 file changed, 3 insertions(+) diff --git a/apps/server/src/provider/ForkProviderSessions.expbkt3.ts b/apps/server/src/provider/ForkProviderSessions.expbkt3.ts index 8e7db6296ef6..017dfeff8afb 100644 --- a/apps/server/src/provider/ForkProviderSessions.expbkt3.ts +++ b/apps/server/src/provider/ForkProviderSessions.expbkt3.ts @@ -155,6 +155,9 @@ export const layer = Layer.effect( } return result; } + // Only a session manager without `listLive` gets here; this read + // blocks the event loop for seconds on a large database. + yield* Effect.logWarning("fork-provider-sessions.list-sessions-full-shell-fallback"); const shell = yield* projection.getShellSnapshot(); for (const thread of [...shell.threads, ...shell.archivedThreads]) { const records = yield* projection.getThreadRecords(thread.id, ["providerSessions"]);