diff --git a/apps/server/src/claudeAccounts/ClaudeAccountsService.test.ts b/apps/server/src/claudeAccounts/ClaudeAccountsService.test.ts index da2e70c1dfd40..ee05cfd5ad1af 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 a0c36822d2b7c..5b29ebb904fd5 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 5743e803b28d2..9ff1d790d5eab 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 966ad2a3ce39a..7ae7980e55a3b 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 3d4c6ac337298..e20ee260ded77 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 0000000000000..88b68b99ac41b --- /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 7de3361517732..017dfeff8afbe 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,40 @@ 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; + } + // 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"]); 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;