diff --git a/apps/server/src/claudeAccounts/ClaudeAccountsService.test.ts b/apps/server/src/claudeAccounts/ClaudeAccountsService.test.ts index 6a59576d32be2..da2e70c1dfd40 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>([]); + /** 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); /** The thread shell's owner; `undefined` means no shell (a draft). */ const shellOwner = yield* Ref.make(owner); @@ -285,6 +287,8 @@ const makeHarness = (options: HarnessOptions = {}) => Effect.sync(() => { stoppedThreads.push(input.threadId); }), + hasLiveBackgroundWork: ({ threadId }) => + Ref.get(backgroundWork).pipe(Effect.map((threads) => threads.has(threadId))), listSessions: () => Ref.get(sessions), streamEvents: Stream.fromSubscription(providerSubscription), }), @@ -329,6 +333,7 @@ const makeHarness = (options: HarnessOptions = {}) => dispatched, enabled, sessions, + backgroundWork, shellSession, shellOwner, shellFails, @@ -985,6 +990,63 @@ describe("ClaudeAccountsService mode changes on a live session", () => { }), ); + scenario( + "waits for a running subagent between turns and restarts on the status poll once it ends", + {}, + (harness, service) => + Effect.gen(function* () { + const threadId = ThreadId.make("thread-background-switch"); + yield* Ref.set(harness.sessions, [liveSession(threadId, "ready")]); + yield* Ref.set(harness.backgroundWork, new Set([threadId])); + yield* service.refreshStatus(); + yield* service.resolveForSession(threadId); + const view = yield* service.setThreadMode({ + threadId, + mode: { kind: "profile", profile: "agent" }, + }); + // No turn runs, but a stop now would kill the subagent. + assert.equal(view.pendingRestart, true); + assert.deepEqual(harness.stoppedThreads, []); + + yield* service.refreshStatus(); + assert.deepEqual(harness.stoppedThreads, []); + + // The subagent ends between turns: no turn end follows, the poll restarts. + yield* Ref.set(harness.backgroundWork, new Set()); + yield* service.refreshStatus(); + assert.deepEqual(harness.stoppedThreads, [threadId]); + assert.notEqual((yield* service.getThread(threadId)).pendingRestart, true); + }), + ); + + scenario( + "restarts anyway once background work has held a deferred restart for an hour", + {}, + (harness, service) => + Effect.gen(function* () { + const threadId = ThreadId.make("thread-background-cap"); + yield* Ref.set(harness.sessions, [liveSession(threadId, "ready")]); + yield* Ref.set(harness.backgroundWork, new Set([threadId])); + yield* service.refreshStatus(); + yield* service.resolveForSession(threadId); + yield* service.setThreadMode({ threadId, mode: { kind: "profile", profile: "agent" } }); + assert.deepEqual(harness.stoppedThreads, []); + // The background poll stays idle while the clock moves; each status + // read below is the one the poll would make. + yield* Ref.set(harness.enabled, false); + + yield* TestClock.adjust("59 minutes"); + yield* service.refreshStatus(); + assert.deepEqual(harness.stoppedThreads, []); + assert.equal((yield* service.getThread(threadId)).pendingRestart, true); + + yield* TestClock.adjust("2 minutes"); + yield* service.refreshStatus(); + assert.deepEqual(harness.stoppedThreads, [threadId]); + assert.notEqual((yield* service.getThread(threadId)).pendingRestart, true); + }), + ); + scenario( "keeps the current account when Auto is chosen from a pin and it is still usable", {}, @@ -2052,6 +2114,8 @@ describe("ClaudeAccountsService access per user: owners, probing, hard limits, r liveSession(busy, "running"), liveSession(untouched, "ready"), ]); + // Background work never holds a session on an account its owner lost. + yield* Ref.set(harness.backgroundWork, new Set([idle])); yield* service.setAccess({ profile: "tushar", userIds: [colleague], actorUserId: null }); assert.deepEqual(harness.stoppedThreads, [idle]); @@ -2661,6 +2725,40 @@ describe("ClaudeAccountsService rebalancing at turn boundaries", () => { ); } + { + const gate = gatedPlace(); + scenario( + "leaves an idle thread alone while its session still runs a background subagent", + { place: gate.place, status: sharedStatus }, + (harness, service) => + Effect.gen(function* () { + yield* service.refreshStatus(); + const threadId = ThreadId.make("thread-rebalance-background"); + gate.state.open = ["tushar"]; + yield* service.resolveForSession(threadId); + gate.state.open = ["agent"]; + yield* Ref.set(harness.sessions, [liveSession(threadId, "ready")]); + yield* Ref.set(harness.backgroundWork, new Set([threadId])); + const placements = harness.placeCalls.length; + + yield* endTurn(harness, threadId); + yield* drainDomainEvents(harness, service); + assert.equal(harness.placeCalls.length, placements); + assert.deepEqual(harness.stoppedThreads, []); + const view = yield* service.getThread(threadId); + assert.equal(view.resolvedProfile, "tushar"); + assert.notEqual(view.pendingRestart, true); + + // Once the work ended, the next turn boundary moves the thread as usual. + yield* Ref.set(harness.backgroundWork, new Set()); + yield* endTurn(harness, threadId); + yield* drainDomainEvents(harness, service); + 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 78c11b3fad3aa..b542a7ed4d710 100644 --- a/apps/server/src/claudeAccounts/ClaudeAccountsService.ts +++ b/apps/server/src/claudeAccounts/ClaudeAccountsService.ts @@ -114,6 +114,11 @@ export const REBALANCE_PROBE_BACKOFF_MS = 5 * 60 * 1000; export const RESET_WAIT_THRESHOLD_MS = 5 * 60 * 1000; /** Margin after the reset so the status poll sees the account free before the thread continues. */ export const RESET_WAIT_BUFFER_MS = 60 * 1000; +/** + * How long a deferred account restart waits for the session's background work + * (a subagent, a background command) before it restarts the session anyway. + */ +export const BACKGROUND_WORK_RESTART_CAP_MS = 60 * 60 * 1000; const SNAPSHOT_CHANGE = "\u0000snapshot"; /** The fields the CLI's `sessions/.json` carries that the live-session index needs. */ const decodeSessionFile = Schema.decodeUnknownEffect( @@ -358,6 +363,19 @@ export const make = Effect.gen(function* () { const threadProfiles = new Map(); const notices = new Map(); const pendingRestart = new Set(); + /** + * When each deferred restart first found the session's background work + * running. Its wait is capped at BACKGROUND_WORK_RESTART_CAP_MS from here. + */ + const backgroundWaitSince = new Map(); + /** Deferred restarts that wait for the turn only, never for background work (access revoked). */ + const restartIgnoresBackgroundWork = new Set(); + /** Forgets a deferred restart and its wait. True when one was pending. */ + const forgetPendingRestart = (threadId: ThreadId): boolean => { + backgroundWaitSince.delete(threadId); + restartIgnoresBackgroundWork.delete(threadId); + return pendingRestart.delete(threadId); + }; /** Threads with a turn in flight, as seen on the provider event bus. */ const activeTurns = new Set(); /** @@ -852,6 +870,8 @@ export const make = Effect.gen(function* () { yield* expireNotices; // Accounts may have room now: continue the threads waiting for one. yield* revivePass; + // Restarts deferred for background work have no turn end to wake them. + yield* recheckBackgroundWaits; }); // Poll only while the setting is on; a disabled feature costs nothing but @@ -1241,7 +1261,7 @@ export const make = Effect.gen(function* () { const row = rowResult.success; yield* pruneExhausted; // A spawn is the restart a deferred mode change was waiting for. - if (pendingRestart.delete(threadId)) yield* announce(threadId); + if (forgetPendingRestart(threadId)) yield* announce(threadId); if (row.mode.kind === "profile") { const pinned = row.mode.profile; @@ -1351,10 +1371,7 @@ export const make = Effect.gen(function* () { * shell (which marks a turn active as soon as its command is accepted, before * the provider has started it) and the provider session itself. */ - const threadIsIdle = ( - threadId: ThreadId, - settled?: OrchestrationSession, - ): Effect.Effect => + const turnIsIdle = (threadId: ThreadId, settled?: OrchestrationSession): Effect.Effect => Effect.gen(function* () { if (activeTurns.has(threadId)) return false; // A `thread.session-set` event carries the session as just applied; the @@ -1381,39 +1398,146 @@ export const make = Effect.gen(function* () { return live === undefined || (live.status !== "running" && live.activeTurnId === undefined); }); - /** Restarts the thread's provider session so its next turn spawns on the new account. */ - const restartForNewAccount = (threadId: ThreadId, why = "mode-changed") => + /** + * Whether the thread can lose its provider session now: between turns, and + * with no background work (a subagent, a background command) still running + * in that session, which a stop would kill. + */ + const threadIsIdle = ( + threadId: ThreadId, + settled?: OrchestrationSession, + ): Effect.Effect => + Effect.gen(function* () { + if (!(yield* turnIsIdle(threadId, settled))) return false; + return !(yield* providers.hasLiveBackgroundWork({ threadId })); + }); + + /** + * Whether a deferred restart keeps waiting for the session's background + * work. The wait counts from the first check that finds work. Past + * BACKGROUND_WORK_RESTART_CAP_MS the restart runs anyway, and the released + * session marks the subagents it leaves open as cancelled. + */ + const waitsForBackgroundWork = (threadId: ThreadId): Effect.Effect => + Effect.gen(function* () { + if (restartIgnoresBackgroundWork.has(threadId)) return false; + if (!(yield* providers.hasLiveBackgroundWork({ threadId }))) return false; + const nowMs = yield* Clock.currentTimeMillis; + const since = backgroundWaitSince.get(threadId); + if (since === undefined) { + backgroundWaitSince.set(threadId, nowMs); + return true; + } + if (nowMs - since < BACKGROUND_WORK_RESTART_CAP_MS) return true; + yield* Effect.logWarning("claude.account.restart-background-wait-expired", { + threadId, + waitedMs: nowMs - since, + }); + return false; + }); + + /** + * Restarts the thread's provider session so its next turn spawns on the new + * account. A running turn defers the restart to its end, and so does the + * session's background work, unless `waitForBackgroundWork` is false. + */ + const restartForNewAccount = ( + threadId: ThreadId, + why = "mode-changed", + options: { readonly waitForBackgroundWork?: boolean } = {}, + ) => Effect.gen(function* () { const sessions = yield* providers.listSessions(); const live = sessions.find( (session) => session.threadId === threadId && session.status !== "closed", ); if (live === undefined) return; - if (!(yield* threadIsIdle(threadId))) { + const ignoresBackgroundWork = options.waitForBackgroundWork === false; + const turnIdle = yield* turnIsIdle(threadId); + const waitingForBackgroundWork = + turnIdle && !ignoresBackgroundWork && (yield* waitsForBackgroundWork(threadId)); + if (!turnIdle || waitingForBackgroundWork) { pendingRestart.add(threadId); - yield* Effect.logInfo("claude.account.restart-deferred", { threadId }); + if (ignoresBackgroundWork) restartIgnoresBackgroundWork.add(threadId); + yield* Effect.logInfo("claude.account.restart-deferred", { + threadId, + waitingFor: turnIdle ? "background-work" : "turn", + }); return; } + backgroundWaitSince.delete(threadId); yield* stopSessionQuietly(threadId, why); }); /** - * The turn a mode change waited for has ended. Stop only if nothing new has - * started since (a queued message may already be on its way); otherwise - * keep waiting for the next completion. + * The turn or the background work a mode change waited for has ended. Stop + * only if nothing new has started since (a queued message may already be on + * its way); otherwise keep waiting for the next completion. */ const completePendingRestart = (threadId: ThreadId, settled?: OrchestrationSession) => Effect.gen(function* () { if (!pendingRestart.has(threadId)) return; - if (!(yield* threadIsIdle(threadId, settled))) { - yield* Effect.logInfo("claude.account.restart-still-deferred", { threadId }); + if (!(yield* turnIsIdle(threadId, settled))) { + yield* Effect.logInfo("claude.account.restart-still-deferred", { + threadId, + waitingFor: "turn", + }); + return; + } + if (yield* waitsForBackgroundWork(threadId)) { + // Another fiber completed the restart while this check ran: its wait + // start must not outlive it. + if (!pendingRestart.has(threadId)) { + backgroundWaitSince.delete(threadId); + return; + } + yield* Effect.logInfo("claude.account.restart-still-deferred", { + threadId, + waitingFor: "background-work", + }); return; } - pendingRestart.delete(threadId); + // Another fiber may have completed this restart while the checks ran. + if (!forgetPendingRestart(threadId)) return; yield* stopSessionQuietly(threadId, "mode-changed-after-turn"); yield* announce(threadId); }); + /** + * A restart deferred for background work has no turn end to wake it when + * that work ends between turns, so the status poll checks each such thread. + * The cheap check reads only the thread's own sessions: the full idle check + * reads every session, and each read starts that session's idle release + * again. + */ + const recheckBackgroundWaits: Effect.Effect = Effect.suspend(() => + Effect.forEach( + [...backgroundWaitSince], + ([threadId, since]) => + Effect.gen(function* () { + if (!pendingRestart.has(threadId)) { + backgroundWaitSince.delete(threadId); + return; + } + // A running turn re-checks the restart itself when it ends. + if (activeTurns.has(threadId)) return; + const nowMs = yield* Clock.currentTimeMillis; + if ( + nowMs - since < BACKGROUND_WORK_RESTART_CAP_MS && + (yield* providers.hasLiveBackgroundWork({ threadId })) + ) { + return; + } + yield* completePendingRestart(threadId); + }), + { discard: true }, + ), + ).pipe( + Effect.catchCause((cause) => + Effect.logWarning("claude.account.background-wait-recheck-failed", { cause }), + ), + ); + // ------------------------------------------------------------ revive queue /** Forgets the rejections already handled for a thread, so the next one is judged afresh. */ @@ -1571,7 +1695,7 @@ export const make = Effect.gen(function* () { if (wouldChange) { yield* restartForNewAccount(threadId); } else { - pendingRestart.delete(threadId); + forgetPendingRestart(threadId); } // Choosing an account for a thread that stopped mid-turn to wait is // "continue now on that account": its session already stopped. @@ -2149,7 +2273,7 @@ export const make = Effect.gen(function* () { projectedRunning.delete(threadId); threadProfiles.delete(threadId); notices.delete(threadId); - pendingRestart.delete(threadId); + forgetPendingRestart(threadId); activeTurns.delete(threadId); movedAwaitingTurn.delete(threadId); yield* repository @@ -2203,7 +2327,9 @@ export const make = Effect.gen(function* () { if (mayUse(access, owner, profile)) continue; if (row.mode.kind === "profile") setNotice(threadId, noAccessDetail(profile)); yield* Effect.logInfo("claude.account.access-revoked", { threadId, profile }); - yield* restartForNewAccount(threadId); + // The session must leave an account its owner may no longer use, even + // while its background work runs. + yield* restartForNewAccount(threadId, "mode-changed", { waitForBackgroundWork: false }); yield* announce(threadId); } }).pipe( diff --git a/apps/server/src/orchestration-v2/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration-v2/Layers/ProjectionSnapshotQuery.ts index 8772f12b0509f..044f25e2fce00 100644 --- a/apps/server/src/orchestration-v2/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration-v2/Layers/ProjectionSnapshotQuery.ts @@ -12,6 +12,7 @@ import { OrchestrationV2PlanArtifact, OrchestrationV2TurnItemJson, } from "@t3tools/contracts"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; @@ -53,7 +54,12 @@ const make = Effect.gen(function* () { const shell = (location: "active" | "archive" = "active", unsettledOnly = false) => wrap( Effect.gen(function* () { - const snapshot = yield* projections.getShellSnapshot({ location, unsettledOnly }); + // The full shell read takes seconds on a large database: run it on the + // SQLite reader worker, off the event loop and the writer. Every caller + // reads committed state; inside an open transaction it stays on the writer. + const snapshot = yield* projections + .getShellSnapshot({ location, unsettledOnly }) + .pipe(NodeSqliteClient.readOnly); // The V2 store returns archived rows in `archivedThreads`, never in `threads`. const threads = location === "archive" ? snapshot.archivedThreads : snapshot.threads; const names = new Map( diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.releasedSubagents.expbkt3.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.releasedSubagents.expbkt3.test.ts new file mode 100644 index 0000000000000..b1c892f4c90fc --- /dev/null +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.releasedSubagents.expbkt3.test.ts @@ -0,0 +1,473 @@ +// T3-CUSTOM(expbkt3): a released provider session settles the subagents it leaves open. +import { assert, it } from "@effect/vitest"; +import { + NodeId, + type OrchestrationV2DomainEvent, + type OrchestrationV2Subagent, + type OrchestrationV2ThreadProjection, + ProviderDriverKind, + ProviderInstanceId, + ProviderSessionId, + ProviderThreadId, + ThreadId, + TurnItemId, +} from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Queue from "effect/Queue"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; +import * as McpSessionRegistry from "../mcp/McpSessionRegistry.ts"; +import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; +import * as EventSink from "./EventSink.ts"; +import * as IdAllocator from "./IdAllocator.ts"; +import * as ProjectionStore from "./ProjectionStore.ts"; +import { + ProviderAdapterProtocolError, + type ProviderAdapterV2Event, + type ProviderAdapterV2Shape, +} from "./ProviderAdapter.ts"; +import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; +import * as ProviderEventIngestor from "./ProviderEventIngestor.ts"; +import * as ProviderSessionManager from "./ProviderSessionManager.ts"; + +const driver = ProviderDriverKind.make("claudeAgent"); +const instanceId = ProviderInstanceId.make("claudeAgent"); +const parentThreadId = ThreadId.make("released-parent-thread"); +const childThreadId = (name: string) => ThreadId.make(`released-child-${name}`); +const releasedSessionId = ProviderSessionId.make("released-session"); +const otherSessionId = ProviderSessionId.make("other-session"); +const otherProviderThreadId = ProviderThreadId.make("other-session-provider-thread"); +const epoch = DateTime.makeUnsafe("1970-01-01T00:00:00Z"); +/** The release happens ten minutes in (TestClock), after `epoch`, before `later`. */ +const later = DateTime.makeUnsafe("1970-01-01T00:20:00Z"); + +type Projection = ProjectionStore.ProjectionRuntimeRecoveryState; + +function subagent( + name: string, + overrides: Partial = {}, +): OrchestrationV2Subagent { + return { + id: NodeId.make(`subagent-${name}`), + threadId: parentThreadId, + runId: null, + parentNodeId: NodeId.make("parent-root"), + origin: "provider_native", + createdBy: "agent", + driver, + providerInstanceId: instanceId, + providerThreadId: null, + childThreadId: childThreadId(name), + nativeTaskRef: null, + prompt: `review ${name}`, + title: null, + model: null, + status: "running", + result: null, + startedAt: epoch, + completedAt: null, + updatedAt: epoch, + ...overrides, + }; +} + +/** The parent's card for a subagent. */ +function subagentItem(task: OrchestrationV2Subagent) { + return { + id: TurnItemId.make(`card-${task.id}`), + threadId: task.threadId, + runId: task.runId, + nodeId: task.id, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: task.status, + title: null, + startedAt: task.startedAt, + completedAt: null, + updatedAt: task.updatedAt, + type: "subagent" as const, + subagentId: task.id, + origin: task.origin, + driver: task.driver, + providerInstanceId: task.providerInstanceId, + childThreadId: task.childThreadId, + prompt: task.prompt, + result: null, + }; +} + +function node( + id: string, + threadId: ThreadId, + kind: "subagent" | "root_turn", + startedAt: DateTime.Utc = epoch, +) { + return { + id: NodeId.make(id), + threadId, + runId: null, + parentNodeId: null, + rootNodeId: NodeId.make(id), + kind, + status: "running" as const, + countsForRun: false, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt, + completedAt: null, + }; +} + +function projection(threadId: ThreadId, records: Partial): Projection { + return { + thread: { id: threadId }, + runs: [], + attempts: [], + nodes: [], + subagents: [], + providerSessions: [], + providerThreads: [], + providerTurns: [], + runtimeRequests: [], + messages: [], + turnItems: [], + ...records, + } as unknown as Projection; +} + +/** + * The incident's shape: a parent whose turn completed while its background + * subagent kept working in a child thread (a runless root turn with a running + * command), next to subagents the release must leave alone. + */ +function incidentProjections(): ReadonlyMap { + const running = subagent("running"); + const startedLater = subagent("started-later", { startedAt: later, updatedAt: later }); + const otherSession = subagent("other-session", { providerThreadId: otherProviderThreadId }); + const delegated = subagent("delegated", { origin: "app_owned" }); + const otherDriver = subagent("other-driver", { driver: ProviderDriverKind.make("codex") }); + const finished = subagent("finished", { status: "completed", completedAt: epoch }); + // Its provider thread was not loaded: it may run in another shared session. + const unloadedThread = subagent("unloaded-thread", { + providerThreadId: ProviderThreadId.make("unloaded-provider-thread"), + }); + const tasks = [ + running, + startedLater, + otherSession, + delegated, + otherDriver, + finished, + unloadedThread, + ]; + const childRoot = node("child-root-running", childThreadId("running"), "root_turn"); + // The last thing the stopped subagent ran, still open under its root turn. + const childCommand = { + id: TurnItemId.make("child-command"), + threadId: childThreadId("running"), + runId: null, + nodeId: childRoot.id, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 2, + status: "running" as const, + title: null, + startedAt: epoch, + completedAt: null, + updatedAt: epoch, + type: "command_execution" as const, + input: "cat notes.md", + }; + return new Map([ + [ + parentThreadId, + projection(parentThreadId, { + subagents: tasks, + turnItems: tasks.map(subagentItem) as never, + nodes: tasks.map((task) => node(task.id, parentThreadId, "subagent")), + providerThreads: [ + { id: otherProviderThreadId, providerSessionId: otherSessionId } as never, + ], + }), + ], + [ + childThreadId("running"), + projection(childThreadId("running"), { + nodes: [childRoot], + turnItems: [childCommand] as never, + }), + ], + // Reached only if the release wrongly settled their subagents. + ...["started-later", "other-session", "delegated", "other-driver", "unloaded-thread"].map( + (name) => + [ + childThreadId(name), + projection(childThreadId(name), { + nodes: [node(`child-root-${name}`, childThreadId(name), "root_turn")], + }), + ] as const, + ), + ]); +} + +function fixture( + options: { + readonly projections?: ReadonlyMap; + readonly failRecoveryRead?: boolean; + } = {}, +) { + const projections = options.projections ?? incidentProjections(); + const written: Array = []; + /** Threads whose full recovery read ran, in order. */ + const recoveryReads: Array = []; + const adapter: ProviderAdapterV2Shape = { + instanceId, + driver, + 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" })); + return { + instanceId, + driver, + providerSessionId: input.providerSessionId, + providerSession: { + id: input.providerSessionId, + driver, + providerInstanceId: instanceId, + status: "ready" as const, + cwd: "/tmp", + model: "test", + capabilities: CodexProviderCapabilitiesV2, + createdAt: epoch, + updatedAt: epoch, + lastError: null, + }, + events: Stream.fromQueue(events), + ensureThread: unused, + resumeThread: (request) => Effect.succeed(request.providerThread), + forkThread: unused, + readThreadSnapshot: unused, + rollbackThread: unused, + startTurn: () => Effect.void, + steerTurn: () => Effect.void, + interruptTurn: () => Effect.void, + respondToRuntimeRequest: () => Effect.void, + }; + }), + }; + const layer = ProviderSessionManager.layerWithOptions({ + // The clock moves in these tests; idle release must not run on its own. + idleTimeoutMs: 24 * 60 * 60 * 1000, + configureMcp: false, + }).pipe( + Layer.provide( + Layer.mergeAll( + IdAllocator.layer, + FileSystem.layerNoop({}), + ProviderAdapterRegistry.layerSingle(adapter), + Layer.mock(EventSink.EventSinkV2)({ + write: (input) => + Effect.sync(() => { + written.push(...input.events); + return []; + }), + }), + Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({}), + Layer.mock(ProjectionStore.ProjectionStoreV2)({ + getThreadRecords: (threadId) => + Effect.succeed({ + thread: { id: threadId }, + messages: [], + runtimeRequests: [], + nodes: [], + turnItems: [], + providerThreads: [], + providerTurns: [], + subagents: projections.get(threadId)?.subagents ?? [], + } as unknown as OrchestrationV2ThreadProjection), + getRuntimeRecoveryProjection: (threadId) => { + recoveryReads.push(threadId); + if (options.failRecoveryRead === true) { + return Effect.fail( + new ProjectionStore.ProjectionStoreReadError({ threadId, cause: "boom" }), + ); + } + const found = projections.get(threadId); + return found === undefined + ? Effect.fail(new ProjectionStore.ProjectionStoreThreadNotFoundError({ threadId })) + : Effect.succeed(found); + }, + }), + Layer.mock(McpSessionRegistry.McpSessionRegistry)({ + revokeThread: () => Effect.void, + revokeProviderSession: () => Effect.void, + }), + ), + ), + ); + const open = (threadId: ThreadId, providerSessionId = releasedSessionId) => ({ + threadId, + providerSessionId, + modelSelection: { instanceId, model: "test" }, + runtimePolicy: { cwd: null, runtimeMode: "full-access", interactionMode: "default" } as const, + }); + /** The cancellations the release wrote, by event type. */ + const cancelled = (type: "subagent.updated" | "turn-item.updated" | "node.updated") => + written.flatMap((event) => { + if ( + event.type !== type || + (event.type !== "subagent.updated" && + event.type !== "turn-item.updated" && + event.type !== "node.updated") + ) { + return []; + } + return event.payload.status === "cancelled" ? [String(event.payload.id)] : []; + }); + return { layer, written, recoveryReads, open, cancelled }; +} + +it.effect( + "a released session cancels the subagent it left running, its card and its child thread", + () => { + const f = fixture(); + return Effect.gen(function* () { + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + yield* manager.open(f.open(parentThreadId)); + yield* TestClock.adjust("10 minutes"); + yield* manager.release({ providerSessionId: releasedSessionId, reason: "manual_shutdown" }); + + assert.isTrue(Option.isNone(yield* manager.get(releasedSessionId))); + 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"]); + // Started after the release, on another session, a delegate_task child, + // another provider, already finished, on an unloaded provider thread: + // none of them is touched. + 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 [ + "started-later", + "other-session", + "delegated", + "other-driver", + "finished", + "unloaded-thread", + ]) { + 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 detach that leaves the shared session open settles nothing", () => { + const f = fixture(); + return Effect.gen(function* () { + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + yield* manager.open(f.open(parentThreadId)); + yield* manager.open(f.open(ThreadId.make("released-sibling-thread"))); + yield* TestClock.adjust("10 minutes"); + yield* manager.detach({ providerSessionId: releasedSessionId, threadId: parentThreadId }); + + assert.isTrue(Option.isSome(yield* manager.get(releasedSessionId))); + assert.deepEqual(f.cancelled("subagent.updated"), []); + assert.deepEqual(f.cancelled("turn-item.updated"), []); + assert.deepEqual(f.cancelled("node.updated"), []); + }).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" }); + + 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(); + return Effect.gen(function* () { + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + yield* manager.open(f.open(parentThreadId)); + yield* TestClock.adjust("10 minutes"); + yield* manager.release({ providerSessionId: releasedSessionId, reason: "server_shutdown" }); + + assert.deepEqual(f.cancelled("subagent.updated"), []); + assert.deepEqual(f.cancelled("node.updated"), []); + }).pipe(Effect.provide(f.layer)); +}); + +it.effect("a release with no open subagent skips the full recovery read", () => { + const finished = subagent("finished", { status: "completed", completedAt: epoch }); + const f = fixture({ + projections: new Map([ + [ + parentThreadId, + projection(parentThreadId, { + subagents: [finished], + turnItems: [subagentItem(finished)] as never, + }), + ], + ]), + }); + return Effect.gen(function* () { + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + yield* manager.open(f.open(parentThreadId)); + yield* TestClock.adjust("10 minutes"); + yield* manager.release({ providerSessionId: releasedSessionId, reason: "idle_timeout" }); + + assert.isTrue(Option.isNone(yield* manager.get(releasedSessionId))); + assert.deepEqual(f.recoveryReads, []); + assert.deepEqual(f.cancelled("subagent.updated"), []); + }).pipe(Effect.provide(f.layer)); +}); + +it.effect("a failed subagent read never fails the release", () => { + const f = fixture({ failRecoveryRead: true }); + return Effect.gen(function* () { + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + yield* manager.open(f.open(parentThreadId)); + yield* manager.release({ providerSessionId: releasedSessionId, reason: "runtime_error" }); + + assert.isTrue(Option.isNone(yield* manager.get(releasedSessionId))); + assert.isTrue( + f.written.some( + (event) => event.type === "provider-session.updated" && event.payload.status === "error", + ), + ); + assert.deepEqual(f.cancelled("subagent.updated"), []); + }).pipe(Effect.provide(f.layer)); +}); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 30f86c8c2f378..49e10783bd416 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -69,6 +69,8 @@ import { import { SourceControlProfileService } from "../sourceControl/SourceControlProfileService.ts"; import type { ProviderSessionExecutionOptions } from "../provider/sessionExecutionOptions.expbkt3.ts"; import { makeBackgroundTurnGuard } from "./backgroundTurnGrant.expbkt3.ts"; +// T3-CUSTOM(expbkt3): a released session settles the subagents it leaves open. +import { settleReleasedSessionSubagents } from "./releasedSessionSubagents.expbkt3.ts"; const DEFAULT_IDLE_TIMEOUT_MS = 30 * 60 * 1000; const DEFAULT_MAX_IDLE_PIN_MS = 4 * 60 * 60 * 1000; @@ -925,6 +927,28 @@ export const layerWithOptions = ( input.entry.requestEventPermit.withPermits(1), ), ), + // T3-CUSTOM(expbkt3): BEGIN the released process can no longer end its subagents. + // Never fails, so it never fails or retries the release. + Effect.exit( + settleReleasedSessionSubagents({ + projectionStore, + idAllocator, + eventSink, + providerSessionId: input.entry.runtime.providerSessionId, + reason: input.reason, + driver: input.entry.runtime.driver, + providerInstanceId: input.entry.runtime.instanceId, + threadIds: input.entry.attachedThreadIds, + liveThreadIds: Ref.get(sessions).pipe( + Effect.map( + (current) => + new Set([...current.values()].flatMap((live) => [...live.attachedThreadIds])), + ), + ), + releasedAt: input.releasedAt, + }), + ), + // T3-CUSTOM(expbkt3): END ], { concurrency: 1 }, ).pipe(Effect.flatMap(Exit.asVoidAll)); diff --git a/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.test.ts b/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.test.ts new file mode 100644 index 0000000000000..d44943873ab82 --- /dev/null +++ b/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.test.ts @@ -0,0 +1,529 @@ +// T3-CUSTOM(expbkt3): provider-native subagents follow their parent into and out of the archive. +import { assert, it } from "@effect/vitest"; +import { + CommandId, + EventId, + NodeId, + type OrchestrationV2AppThread, + type OrchestrationV2StoredEvent, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + ThreadId, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; + +import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import { EventSinkV2 } from "./EventSink.ts"; +import * as Orchestrator from "./Orchestrator.ts"; +import { ProviderAdapterRegistryV2 } from "./ProviderAdapterRegistry.ts"; +import * as SubagentArchiveCascade from "./SubagentArchiveCascade.expbkt3.ts"; +import { layerWithRegistry } from "./testkit/ProviderReplayHarness.ts"; + +const layerTest = layerWithRegistry( + { name: "SubagentArchiveCascade.expbkt3.test.ts" }, + Layer.mock(ProviderAdapterRegistryV2)({ + get: () => Effect.die("Provider execution is outside this test."), + list: () => Effect.succeed([ProviderInstanceId.make("codex")]), + }), + { runEffectWorker: false, databaseLayer: SqlitePersistence.layerMemory }, +).pipe(Layer.provideMerge(SqlitePersistence.layerMemory)); + +const projectId = ProjectId.make("project:subagent-archive-cascade"); +const modelSelection = { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }; +const driver = ProviderDriverKind.make("codex"); + +const makeThread = ( + id: string, + now: DateTime.Utc, + overrides: Partial = {}, +): OrchestrationV2AppThread => { + const threadId = ThreadId.make(id); + return { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId, + title: id, + providerInstanceId: modelSelection.instanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + ...overrides, + }; +}; + +/** A child as each creator records it: a provider's native subagent, delegate_task, or a fork. */ +const makeChild = ( + kind: "provider-subagent" | "delegated" | "fork", + parentThreadId: ThreadId, + id: string, + now: DateTime.Utc, + overrides: Partial = {}, +) => + makeThread(id, now, { + createdBy: kind === "fork" ? "user" : "agent", + creationSource: + kind === "provider-subagent" ? "provider" : kind === "delegated" ? "mcp" : "web", + lineage: { + parentThreadId, + relationshipToParent: kind === "fork" ? "fork" : "subagent", + rootThreadId: parentThreadId, + }, + forkedFrom: kind === "fork" ? null : { type: "node", nodeId: NodeId.make(`node:${id}`) }, + ...overrides, + }); + +const seed = (threads: ReadonlyArray) => + Effect.gen(function* () { + const eventSink = yield* EventSinkV2; + yield* eventSink.write({ + events: threads.map((thread) => ({ + id: EventId.make(`event:${thread.id}:created`), + type: "thread.created" as const, + threadId: thread.id, + occurredAt: thread.createdAt, + payload: thread, + })), + }); + }); + +/** + * The parent's record of a provider-native subagent, as the adapters write it: + * a native subagent's thread has no run, so this record says whether it is live. + */ +const seedSubagentRecord = ( + child: OrchestrationV2AppThread, + status: "running" | "completed", + now: DateTime.Utc, +) => + Effect.gen(function* () { + const eventSink = yield* EventSinkV2; + const parentThreadId = child.lineage.parentThreadId!; + const subagentId = NodeId.make(`subagent:${child.id}`); + yield* eventSink.write({ + events: [ + { + id: EventId.make(`event:${subagentId}:${status}`), + type: "subagent.updated", + threadId: parentThreadId, + nodeId: subagentId, + driver, + providerInstanceId: modelSelection.instanceId, + occurredAt: now, + payload: { + id: subagentId, + threadId: parentThreadId, + runId: null, + parentNodeId: NodeId.make(`node:${parentThreadId}:root`), + origin: "provider_native", + createdBy: "agent", + driver, + providerInstanceId: modelSelection.instanceId, + providerThreadId: null, + childThreadId: child.id, + nativeTaskRef: null, + prompt: "Explore the repository", + title: null, + model: null, + status, + result: status === "completed" ? "Done." : null, + startedAt: now, + completedAt: status === "completed" ? now : null, + updatedAt: now, + }, + }, + ], + }); + }); + +/** The next stored event of this type on the thread, even one that lands before the read subscribes. */ +const nextEvent = (threadId: ThreadId, eventType: "thread.archived" | "thread.unarchived") => + Effect.gen(function* () { + const eventSink = yield* EventSinkV2; + const afterSequence = yield* eventSink.latestSequence(); + return yield* eventSink + .stream({ threadId, afterSequence, eventType }) + .pipe(Stream.runHead, Effect.forkChild); + }); + +const commandIdOf = (stored: Option.Option) => + String(Option.getOrNull(Option.map(stored, (event) => event.commandId))); + +const isArchived = (threadId: ThreadId) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const shell = yield* orchestrator.getThreadShell(threadId); + return shell !== null && shell.archivedAt !== null; + }); + +/** When the thread was archived, on the test clock. */ +const archivedAtMillis = (threadId: ThreadId) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const shell = yield* orchestrator.getThreadShell(threadId); + if (shell?.archivedAt == null) return yield* Effect.die(`${threadId} is not archived.`); + return DateTime.toEpochMillis(shell.archivedAt); + }); + +it.effect( + "archiving a parent archives its provider subagents, not its delegated tasks, forks or other threads", + () => + Effect.scoped( + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const now = yield* DateTime.now; + const parent = makeThread("thread:parent", now); + const other = makeThread("thread:other", now); + const subagent = makeChild("provider-subagent", parent.id, "thread:parent:subagent", now); + const delegated = makeChild("delegated", parent.id, "thread:parent:delegated", now); + const fork = makeChild("fork", parent.id, "thread:parent:fork", now); + const otherSubagent = makeChild( + "provider-subagent", + other.id, + "thread:other:subagent", + now, + ); + yield* seed([parent, other, subagent, delegated, fork, otherSubagent]); + + const cascade = yield* SubagentArchiveCascade.make({ + startupPause: Duration.zero, + cascadePause: Duration.zero, + }); + yield* cascade.start(); + // Every parent is active, so the startup pass archives nothing. + yield* cascade.drain; + assert.isFalse(yield* isArchived(subagent.id)); + + const subagentArchived = yield* nextEvent(subagent.id, "thread.archived"); + yield* orchestrator.dispatch({ + type: "thread.archive", + commandId: CommandId.make("command:archive-parent"), + threadId: parent.id, + }); + const stored = yield* Fiber.join(subagentArchived); + yield* cascade.drain; + + assert.isTrue(commandIdOf(stored).startsWith("server:subagent-archive-cascade:")); + assert.isTrue(yield* isArchived(parent.id)); + assert.isTrue(yield* isArchived(subagent.id)); + assert.isFalse(yield* isArchived(delegated.id)); + assert.isFalse(yield* isArchived(fork.id)); + assert.isFalse(yield* isArchived(other.id)); + assert.isFalse(yield* isArchived(otherSubagent.id)); + }), + ).pipe(Effect.provide(layerTest)), +); + +it.effect("deleting a parent archives its provider subagents", () => + Effect.scoped( + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const now = yield* DateTime.now; + const parent = makeThread("thread:deleted-parent", now); + const subagent = makeChild( + "provider-subagent", + parent.id, + "thread:deleted-parent:subagent", + now, + ); + yield* seed([parent, subagent]); + + const cascade = yield* SubagentArchiveCascade.make({ + startupPause: Duration.zero, + cascadePause: Duration.zero, + }); + yield* cascade.start(); + yield* cascade.drain; + + const subagentArchived = yield* nextEvent(subagent.id, "thread.archived"); + yield* orchestrator.dispatch({ + type: "thread.delete", + commandId: CommandId.make("command:delete-parent"), + threadId: parent.id, + }); + const stored = yield* Fiber.join(subagentArchived); + yield* cascade.drain; + + assert.isTrue(commandIdOf(stored).startsWith("server:subagent-archive-cascade:")); + assert.isNull(yield* orchestrator.getThreadShell(parent.id)); + assert.isTrue(yield* isArchived(subagent.id)); + }), + ).pipe(Effect.provide(layerTest)), +); + +it.effect("unarchiving the parent unarchives its provider subagents and nothing else", () => + Effect.scoped( + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const now = yield* DateTime.now; + const parent = makeThread("thread:round-trip", now); + const subagent = makeChild("provider-subagent", parent.id, "thread:round-trip:subagent", now); + // Archived by the user before the parent was, so the parent's unarchive must not restore it. + const delegated = makeChild("delegated", parent.id, "thread:round-trip:delegated", now, { + archivedAt: now, + }); + yield* seed([parent, subagent, delegated]); + + const cascade = yield* SubagentArchiveCascade.make({ + startupPause: Duration.zero, + cascadePause: Duration.zero, + }); + yield* cascade.start(); + yield* cascade.drain; + + const subagentArchived = yield* nextEvent(subagent.id, "thread.archived"); + yield* orchestrator.dispatch({ + type: "thread.archive", + commandId: CommandId.make("command:archive-round-trip"), + threadId: parent.id, + }); + yield* Fiber.join(subagentArchived); + yield* cascade.drain; + assert.isTrue(yield* isArchived(subagent.id)); + + const subagentUnarchived = yield* nextEvent(subagent.id, "thread.unarchived"); + yield* orchestrator.dispatch({ + type: "thread.unarchive", + commandId: CommandId.make("command:unarchive-round-trip"), + threadId: parent.id, + }); + const stored = yield* Fiber.join(subagentUnarchived); + yield* cascade.drain; + + assert.isTrue(commandIdOf(stored).startsWith("server:subagent-unarchive-cascade:")); + assert.isFalse(yield* isArchived(parent.id)); + assert.isFalse(yield* isArchived(subagent.id)); + assert.isTrue(yield* isArchived(delegated.id)); + }), + ).pipe(Effect.provide(layerTest)), +); + +it.effect( + "the startup pass archives subagents whose parent is archived, deleted or gone, and only those", + () => + Effect.scoped( + Effect.gen(function* () { + const now = yield* DateTime.now; + const archivedParent = makeThread("thread:archived-parent", now, { archivedAt: now }); + const deletedParent = makeThread("thread:gone-parent", now, { deletedAt: now }); + const activeParent = makeThread("thread:active-parent", now); + const ofArchived = makeChild( + "provider-subagent", + archivedParent.id, + "thread:archived-parent:subagent", + now, + ); + const ofDeleted = makeChild( + "provider-subagent", + deletedParent.id, + "thread:gone-parent:subagent", + now, + ); + const ofMissing = makeChild( + "provider-subagent", + ThreadId.make("thread:never-recorded"), + "thread:never-recorded:subagent", + now, + ); + const stillRunning = makeChild( + "provider-subagent", + archivedParent.id, + "thread:archived-parent:running-subagent", + now, + ); + const delegated = makeChild( + "delegated", + archivedParent.id, + "thread:archived-parent:delegated", + now, + ); + const ofActive = makeChild( + "provider-subagent", + activeParent.id, + "thread:active-parent:subagent", + now, + ); + yield* seed([ + archivedParent, + deletedParent, + activeParent, + ofArchived, + ofDeleted, + ofMissing, + stillRunning, + delegated, + ofActive, + ]); + yield* seedSubagentRecord(stillRunning, "running", now); + + const cascade = yield* SubagentArchiveCascade.make({ + startupPause: Duration.zero, + cascadePause: Duration.zero, + }); + yield* cascade.start(); + yield* cascade.drain; + + assert.isTrue(yield* isArchived(ofArchived.id)); + assert.isTrue(yield* isArchived(ofDeleted.id)); + assert.isTrue(yield* isArchived(ofMissing.id)); + assert.isFalse(yield* isArchived(stillRunning.id)); + assert.isFalse(yield* isArchived(delegated.id)); + assert.isFalse(yield* isArchived(ofActive.id)); + assert.isFalse(yield* isArchived(activeParent.id)); + }), + ).pipe(Effect.provide(layerTest)), +); + +it.effect("the startup pass archives one orphan per pause", () => + Effect.scoped( + Effect.gen(function* () { + const now = yield* DateTime.now; + const parent = makeThread("thread:paced-parent", now, { archivedAt: now }); + const first = makeChild("provider-subagent", parent.id, "thread:paced-parent:1", now); + const second = makeChild("provider-subagent", parent.id, "thread:paced-parent:2", now); + const third = makeChild("provider-subagent", parent.id, "thread:paced-parent:3", now); + yield* seed([parent, first, second, third]); + const firstArchived = yield* nextEvent(first.id, "thread.archived"); + const secondArchived = yield* nextEvent(second.id, "thread.archived"); + const thirdArchived = yield* nextEvent(third.id, "thread.archived"); + + const cascade = yield* SubagentArchiveCascade.make({ + startupPause: Duration.seconds(2), + cascadePause: Duration.zero, + }); + yield* cascade.start(); + + // The first orphan goes at once; each later one waits for the clock to move a pause. + yield* Fiber.join(firstArchived); + assert.isFalse(yield* isArchived(second.id)); + yield* TestClock.adjust("2 seconds"); + yield* Fiber.join(secondArchived); + assert.isFalse(yield* isArchived(third.id)); + yield* TestClock.adjust("2 seconds"); + yield* Fiber.join(thirdArchived); + yield* cascade.drain; + + const start = DateTime.toEpochMillis(now); + assert.deepEqual( + [ + yield* archivedAtMillis(first.id), + yield* archivedAtMillis(second.id), + yield* archivedAtMillis(third.id), + ], + [start, start + 2_000, start + 4_000], + ); + }), + ).pipe(Effect.provide(layerTest)), +); + +it.effect("a subagent still running when its parent is archived follows once it finishes", () => + Effect.scoped( + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const now = yield* DateTime.now; + const parent = makeThread("thread:busy-parent", now); + // Read before its sibling, so the sibling's archive shows the cascade has passed it. + const running = makeChild( + "provider-subagent", + parent.id, + "thread:busy-parent:a-running", + now, + ); + const done = makeChild("provider-subagent", parent.id, "thread:busy-parent:b-done", now); + yield* seed([parent, running, done]); + yield* seedSubagentRecord(running, "running", now); + yield* seedSubagentRecord(done, "completed", now); + + const cascade = yield* SubagentArchiveCascade.make({ + startupPause: Duration.zero, + cascadePause: Duration.zero, + }); + yield* cascade.start(); + yield* cascade.drain; + + const doneArchived = yield* nextEvent(done.id, "thread.archived"); + yield* orchestrator.dispatch({ + type: "thread.archive", + commandId: CommandId.make("command:archive-busy-parent"), + threadId: parent.id, + }); + yield* Fiber.join(doneArchived); + yield* cascade.drain; + assert.isFalse(yield* isArchived(running.id)); + + const runningArchived = yield* nextEvent(running.id, "thread.archived"); + yield* seedSubagentRecord(running, "completed", now); + const stored = yield* Fiber.join(runningArchived); + yield* cascade.drain; + + assert.isTrue(commandIdOf(stored).startsWith("server:subagent-archive-cascade:")); + assert.isTrue(yield* isArchived(running.id)); + }), + ).pipe(Effect.provide(layerTest)), +); + +it.effect("a parent's subagents follow it one per pause", () => + Effect.scoped( + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const now = yield* DateTime.now; + const parent = makeThread("thread:paced-cascade", now); + const first = makeChild("provider-subagent", parent.id, "thread:paced-cascade:1", now); + const second = makeChild("provider-subagent", parent.id, "thread:paced-cascade:2", now); + const third = makeChild("provider-subagent", parent.id, "thread:paced-cascade:3", now); + yield* seed([parent, first, second, third]); + + const cascade = yield* SubagentArchiveCascade.make({ + startupPause: Duration.zero, + cascadePause: Duration.millis(500), + }); + yield* cascade.start(); + yield* cascade.drain; + + const firstArchived = yield* nextEvent(first.id, "thread.archived"); + const thirdArchived = yield* nextEvent(third.id, "thread.archived"); + yield* orchestrator.dispatch({ + type: "thread.archive", + commandId: CommandId.make("command:archive-paced-cascade"), + threadId: parent.id, + }); + yield* Fiber.join(firstArchived); + // The clock has not moved, so the next child still waits for its pause. + assert.isFalse(yield* isArchived(second.id)); + + // Move the clock in small steps until the last child lands; the bound turns a + // stuck cascade into a failure instead of a hang. + for (let step = 0; step < 1_000 && thirdArchived.pollUnsafe() === undefined; step += 1) { + yield* TestClock.adjust("50 millis"); + } + assert.isDefined(thirdArchived.pollUnsafe()); + yield* cascade.drain; + + const firstAt = yield* archivedAtMillis(first.id); + const secondAt = yield* archivedAtMillis(second.id); + const thirdAt = yield* archivedAtMillis(third.id); + assert.isAtLeast(secondAt - firstAt, 500); + assert.isAtLeast(thirdAt - secondAt, 500); + }), + ).pipe(Effect.provide(layerTest)), +); diff --git a/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.ts b/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.ts new file mode 100644 index 0000000000000..a1af2fb4be24d --- /dev/null +++ b/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.ts @@ -0,0 +1,359 @@ +/** + * T3-CUSTOM(expbkt3): a provider-native subagent thread follows its parent into + * and out of the archive. + * + * Claude's Task/Agent tool children, and other providers' native subagents, are + * mirrored as threads (`isProviderNativeSubagentThread`) that every client + * hides. Archiving or deleting the parent never touched them, so a finished + * subagent stayed in every active shell read for good: on bkt3 on 2026-10-19, + * 318 of the 1,144 active threads were subagents of an archived parent. + * + * - `thread.archived` or `thread.deleted` of a parent archives its + * provider-native subagent children, and `thread.unarchived` unarchives them. + * - A child whose subagent record is still pending, running or waiting is not + * archived. It follows its parent when that record stops being live. + * - Once per start, after activation, every provider-native subagent whose + * parent is archived, deleted or gone is archived. That pass is the cleanup of + * the orphans that predate this rule, and heals a cascade a restart cut short. + * + * Every archive also starts a history export (`ArchiveExportReactor`) that runs + * git in the worktree, so nothing here moves children in a burst: live moves are + * at least `cascadePause` apart, and the startup pass queues one orphan per + * `startupPause` between them. delegate_task children (`creationSource: "mcp"`) + * and forks keep their own lifecycle. Each child is re-read just before it + * moves, so a later event that reverses an earlier one wins. + */ +import { + CommandId, + isProviderNativeSubagentThread, + type OrchestrationV2DomainEvent, + type OrchestrationV2ServerCommand, + type ProjectId, + ThreadId, +} from "@t3tools/contracts"; +import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; +import * as Cause from "effect/Cause"; +import * as Clock from "effect/Clock"; +import * as Deferred from "effect/Deferred"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as SqlClient from "effect/sql/SqlClient"; +import * as Stream from "effect/Stream"; + +import { forkParked, ServerActivation } from "../serverActivation.ts"; +import { EventSinkV2 } from "./EventSink.ts"; +import * as Orchestrator from "./Orchestrator.ts"; +import { randomUuidV4 } from "./RandomUuid.ts"; + +/** ~318 startup archives on bkt3 spread over ~11 minutes, with their history exports. */ +const DEFAULT_STARTUP_PAUSE = Duration.seconds(2); + +/** A parent with 63 subagents (bkt3's largest) archives them over ~30 seconds. */ +const DEFAULT_CASCADE_PAUSE = Duration.millis(500); + +/** The startup pass logs its progress once per this many orphans. */ +const STARTUP_PROGRESS_EVERY = 50; + +type CascadeDirection = "archive" | "unarchive"; + +type CascadeRequest = + /** A parent moved: its children follow. */ + | { + readonly type: "follow"; + readonly direction: CascadeDirection; + readonly parentThreadId: ThreadId; + readonly projectId: ProjectId; + } + /** A subagent record stopped being live: its thread follows an archived parent. */ + | { readonly type: "finished"; readonly threadId: ThreadId } + | { readonly type: "startup-orphan"; readonly threadId: ThreadId } + | { readonly type: "startup-summary" }; + +/** Which children a read covers: every one, one parent's, or a single child. */ +type ChildScope = + | { readonly kind: "all" } + | { readonly kind: "parent"; readonly parentThreadId: ThreadId; readonly projectId: ProjectId } + | { readonly kind: "child"; readonly threadId: ThreadId }; + +interface ChildRow { + readonly thread_id: string; +} + +/** The work a domain event starts, or null when no child has anything to follow. */ +function cascadeRequestFor(event: OrchestrationV2DomainEvent): CascadeRequest | null { + if (event.type === "subagent.updated") { + const subagent = event.payload; + const live = + subagent.status === "pending" || + subagent.status === "running" || + subagent.status === "waiting"; + return subagent.origin === "provider_native" && subagent.childThreadId !== null && !live + ? { type: "finished", threadId: subagent.childThreadId } + : null; + } + if ( + event.type !== "thread.archived" && + event.type !== "thread.unarchived" && + event.type !== "thread.deleted" + ) { + return null; + } + // A provider-native subagent has no provider-native children, so its own + // archive (most are sent from here) needs no lookup. + if (isProviderNativeSubagentThread(event.payload)) return null; + return { + type: "follow", + direction: event.type === "thread.unarchived" ? "unarchive" : "archive", + parentThreadId: event.threadId, + projectId: event.payload.projectId, + }; +} + +export const make = ( + options: { + readonly startupPause?: Duration.Input; + readonly cascadePause?: Duration.Input; + } = {}, +) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSinkV2; + const sql = yield* SqlClient.SqlClient; + const activation = (yield* ServerActivation) ?? Effect.void; + const startupPause = Duration.fromInputUnsafe(options.startupPause ?? DEFAULT_STARTUP_PAUSE); + const cascadePauseMillis = Duration.toMillis(options.cascadePause ?? DEFAULT_CASCADE_PAUSE); + // Completed once the startup pass has queued its last orphan. + const startupQueued = yield* Deferred.make(); + const startupPass = { candidates: 0, processed: 0, archived: 0 }; + let lastLiveMoveAt: number | null = null; + + /** + * Provider-native subagents in `scope` that are not where their parent is. + * To archive: the child is unarchived, its parent is archived, deleted or + * gone, and its subagent record is not live. To unarchive: the child is + * archived and its parent is active. A child shares its parent's project, so + * a parent's read stays on the project index; a child's is a key lookup. + */ + const selectChildren = (direction: CascadeDirection, scope: ChildScope) => + sql` + SELECT child.thread_id AS thread_id + FROM orchestration_v2_projection_threads AS child + LEFT JOIN orchestration_v2_projection_threads AS parent + ON parent.thread_id = CASE WHEN json_valid(child.payload_json) + THEN json_extract(child.payload_json, '$.lineage.parentThreadId') END + WHERE child.deleted_at IS NULL${ + scope.kind === "parent" + ? sql` AND child.project_id = ${scope.projectId}` + : scope.kind === "child" + ? sql` AND child.thread_id = ${scope.threadId}` + : sql`` + } + AND ${direction === "archive" ? sql`child.archived_at IS NULL` : sql`child.archived_at IS NOT NULL`} + AND CASE WHEN json_valid(child.payload_json) THEN + json_extract(child.payload_json, '$.lineage.parentThreadId') ${ + scope.kind === "parent" ? sql`= ${scope.parentThreadId}` : sql`IS NOT NULL` + } + AND json_extract(child.payload_json, '$.lineage.relationshipToParent') = 'subagent' + AND json_extract(child.payload_json, '$.creationSource') = 'provider' + ELSE 0 END + AND ${ + direction === "archive" + ? sql`( + parent.thread_id IS NULL + OR parent.archived_at IS NOT NULL + OR parent.deleted_at IS NOT NULL + ) + AND NOT EXISTS ( + SELECT 1 + FROM orchestration_v2_projection_subagents AS subagent + WHERE subagent.child_thread_id = child.thread_id + AND subagent.status IN ('pending', 'running', 'waiting') + )` + : sql`( + parent.thread_id IS NOT NULL + AND parent.archived_at IS NULL + AND parent.deleted_at IS NULL + )` + } + ORDER BY child.created_at ASC, child.thread_id ASC + `; + + /** Waits until the last live move is `cascadePause` old. */ + const paceLiveMove = Effect.gen(function* () { + if (lastLiveMoveAt === null) return; + const wait = lastLiveMoveAt + cascadePauseMillis - (yield* Clock.currentTimeMillis); + if (wait > 0) yield* Effect.sleep(Duration.millis(wait)); + }); + + const isDue = (direction: CascadeDirection, threadId: ThreadId) => + selectChildren(direction, { kind: "child", threadId }).pipe( + Effect.map((rows) => rows.length > 0), + ); + + /** + * Re-reads one child and moves it when it still needs to; true when the + * orchestrator accepted the command. A live move waits for its pace, then + * reads the child again, so a check with nothing to do never waits. + */ + const moveChild = ( + direction: CascadeDirection, + threadId: ThreadId, + source: "cascade" | "startup", + ) => + Effect.gen(function* () { + if (!(yield* isDue(direction, threadId))) return false; + if (source === "cascade" && lastLiveMoveAt !== null && cascadePauseMillis > 0) { + yield* paceLiveMove; + if (!(yield* isDue(direction, threadId))) return false; + } + const uuid = yield* randomUuidV4; + const commandId = CommandId.make( + `server:subagent-${direction}-${source}:${threadId}:${uuid}`, + ); + const command: OrchestrationV2ServerCommand = + direction === "archive" + ? { type: "thread.archive", commandId, threadId } + : { type: "thread.unarchive", commandId, threadId }; + yield* orchestrator.dispatch(command); + if (source === "cascade") lastLiveMoveAt = yield* Clock.currentTimeMillis; + return true; + }).pipe( + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.failCause(cause) + : Effect.logWarning("subagent archive cascade: a child did not follow its parent", { + threadId, + direction, + cause: Cause.pretty(cause), + }).pipe(Effect.as(false)), + ), + ); + + const processRequest = Effect.fn("SubagentArchiveCascade.process")(function* ( + request: CascadeRequest, + ) { + switch (request.type) { + case "follow": { + const children = yield* selectChildren(request.direction, { + kind: "parent", + parentThreadId: request.parentThreadId, + projectId: request.projectId, + }); + if (children.length === 0) return; + let moved = 0; + for (const child of children) { + if (yield* moveChild(request.direction, ThreadId.make(child.thread_id), "cascade")) { + moved += 1; + } + } + yield* Effect.logInfo("subagent archive cascade: subagents followed their parent", { + parentThreadId: request.parentThreadId, + direction: request.direction, + candidates: children.length, + moved, + }); + return; + } + case "finished": + if (yield* moveChild("archive", request.threadId, "cascade")) { + yield* Effect.logInfo( + "subagent archive cascade: a finished subagent followed its archived parent", + { threadId: request.threadId }, + ); + } + return; + case "startup-orphan": { + if (yield* moveChild("archive", request.threadId, "startup")) { + startupPass.archived += 1; + } + startupPass.processed += 1; + if ( + startupPass.processed % STARTUP_PROGRESS_EVERY === 0 && + startupPass.processed < startupPass.candidates + ) { + yield* Effect.logInfo("subagent archive cascade: startup pass progress", { + ...startupPass, + }); + } + return; + } + case "startup-summary": + yield* Effect.logInfo( + "subagent archive cascade: archived the subagents of archived or deleted parents", + { ...startupPass }, + ); + return; + } + }); + + // Items wait for activation. Live work and startup orphans share this one + // worker, so they never race each other on the same child. + const worker = yield* makeDrainableWorker((request: CascadeRequest) => + activation.pipe( + Effect.andThen(processRequest(request)), + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.failCause(cause) + : Effect.logWarning("subagent archive cascade: request failed", { + request, + cause: Cause.pretty(cause), + }), + ), + ), + ); + + /** Queues the orphans one per pause, so live work runs between them. */ + const queueStartupPass = Effect.gen(function* () { + const orphans = yield* selectChildren("archive", { kind: "all" }); + startupPass.candidates = orphans.length; + for (const [index, orphan] of orphans.entries()) { + if (index > 0 && !Duration.isZero(startupPause)) yield* Effect.sleep(startupPause); + yield* worker.enqueue({ + type: "startup-orphan", + threadId: ThreadId.make(orphan.thread_id), + }); + } + }).pipe( + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.failCause(cause) + : Effect.logWarning("subagent archive cascade: startup pass failed", { + cause: Cause.pretty(cause), + }), + ), + Effect.andThen(worker.enqueue({ type: "startup-summary" })), + Effect.ensuring(Deferred.succeed(startupQueued, undefined)), + ); + + const start = Effect.fn("SubagentArchiveCascade.start")(function* () { + // Read from the current head: an event that lands before the parked + // stream subscribes is replayed rather than lost. + const afterSequence = yield* eventSink.latestSequence().pipe(Effect.orDie); + yield* forkParked( + eventSink.stream({ afterSequence }).pipe( + Stream.runForEach((stored) => { + const request = cascadeRequestFor(stored.event); + return request === null ? Effect.void : worker.enqueue(request); + }), + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.failCause(cause) + : Effect.logWarning("subagent archive cascade: event stream failed", { + cause: Cause.pretty(cause), + }), + ), + ), + ); + yield* forkParked(queueStartupPass); + }); + + return { + start, + /** After `start`: waits for the whole startup pass, then for every queued item. */ + drain: Deferred.await(startupQueued).pipe(Effect.andThen(worker.drain)), + }; + }); + +/** Starts the cascade with the server; one marked line in `server.ts` provides it. */ +export const layer = Layer.effectDiscard(Effect.flatMap(make(), (cascade) => cascade.start())); diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts index 3fb5d2794748e..1506f04a9f12a 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts @@ -2308,6 +2308,55 @@ it.effect("cancels tracked setup before provider work is released", () => }), ); +it.effect("stops an in-flight branch rename when tracked setup is cancelled", () => + Effect.gen(function* () { + const setupEntered = yield* Deferred.make(); + const branchNameStarted = yield* Deferred.make(); + const allowBranchName = yield* Deferred.make(); + const branchNameInterrupted = yield* Deferred.make(); + const renameAttempted = yield* Deferred.make(); + const harness = makeHarness({ + renameBranch: (input) => + Deferred.succeed(renameAttempted, undefined).pipe(Effect.as({ branch: input.newBranch })), + runSetup: () => Deferred.succeed(setupEntered, undefined).pipe(Effect.andThen(Effect.never)), + generateBranchName: () => + Deferred.succeed(branchNameStarted, undefined).pipe( + Effect.andThen(Deferred.await(allowBranchName)), + Effect.as({ branch: "generated-branch" }), + Effect.onInterrupt(() => Deferred.succeed(branchNameInterrupted, undefined)), + ), + }); + yield* Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const tracker = yield* WorktreeSetupTracker.WorktreeSetupTracker; + const threads = yield* ThreadManagement.ThreadManagementService; + const launched = yield* launches.launch( + launchInput({ + command: "launch:cancel-renaming", + thread: "thread:cancel-renaming", + message: "Start", + workspace: { type: "worktree", baseRef: "main" }, + }), + ); + yield* Deferred.await(setupEntered); + yield* Deferred.await(branchNameStarted); + assert.isTrue(yield* tracker.cancel(launched.threadId)); + // Let a rename that survived the cancel finish, then check it never ran. + yield* Deferred.succeed(allowBranchName, undefined); + yield* Effect.raceFirst( + Deferred.await(branchNameInterrupted), + Deferred.await(renameAttempted), + ); + assert.isTrue( + yield* Deferred.isDone(branchNameInterrupted), + "the cancelled setup's branch rename kept running", + ); + assert.equal(harness.renameBranch.mock.calls.length, 0); + assert.isNull((yield* threads.getThreadProjection(launched.threadId)).thread.worktreePath); + }).pipe(Effect.provide(harness.layer)); + }), +); + it.effect.each([0, 1])("releases an async setup before its completion with exit %s", (exitCode) => Effect.gen(function* () { const completion = yield* Deferred.make<{ exitCode: number | null; durationMs: number }>(); diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.ts index 97e7b574ad9cc..9ac2d5159926e 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.ts @@ -23,6 +23,7 @@ import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; @@ -259,6 +260,9 @@ const make = Effect.gen(function* () { let createdWorktreePath: string | null = null; let setupTerminalId: string | null = null; let workspaceRecorded = false; + // The background branch rename records the worktree again when it lands, + // so cleanup stops it before forgetting a removed worktree. + let branchRename: Fiber.Fiber | null = null; if (input.workspaceStrategy.type === "worktree") { yield* setupTracker.begin({ threadId, @@ -446,7 +450,7 @@ const make = Effect.gen(function* () { ) { const oldBranch = branch; const worktreeCwd = worktreePath; - yield* generateBranchNameFor(worktreeCwd, initialMessage).pipe( + branchRename = yield* generateBranchNameFor(worktreeCwd, initialMessage).pipe( Effect.flatMap(({ branch: newBranch, exactName }) => git.renameBranch({ cwd: worktreeCwd, @@ -586,6 +590,7 @@ const make = Effect.gen(function* () { yield* terminals .close({ threadId, terminalId: setupTerminalId, deleteHistory: true }) .pipe(Effect.ignore); + if (branchRename !== null) yield* Fiber.interrupt(branchRename); const removedPath = createdWorktreePath; // The thread forgets the worktree only once it is gone; a failed // removal leaves the directory for the user to clean up rather than diff --git a/apps/server/src/orchestration-v2/ThreadPullRequestService.ts b/apps/server/src/orchestration-v2/ThreadPullRequestService.ts index a1dabb267b2f8..a7c38e244c11b 100644 --- a/apps/server/src/orchestration-v2/ThreadPullRequestService.ts +++ b/apps/server/src/orchestration-v2/ThreadPullRequestService.ts @@ -11,6 +11,7 @@ import { type ThreadLinkedPullRequest, } from "@t3tools/contracts"; import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; import * as Cause from "effect/Cause"; // T3-CUSTOM(expbkt3): the idle-thread rotation reads the clock. import * as Clock from "effect/Clock"; @@ -133,10 +134,12 @@ export const make = Effect.gen(function* () { */ const readThreadSnapshot = ({ threadId, backfill }: RefreshRequest) => threadId === null - ? orchestrator.getShellSnapshot({ - location: "active", - unsettledOnly: !(backfill || pendingBackfill.size > 0), - }) + ? orchestrator + .getShellSnapshot({ + location: "active", + unsettledOnly: !(backfill || pendingBackfill.size > 0), + }) + .pipe(NodeSqliteClient.readOnly) : Effect.gen(function* () { // Read the sequence first. The thread is then at least this new, so a // sync guarded by the sequence is rejected rather than missing a change. diff --git a/apps/server/src/orchestration-v2/http.ts b/apps/server/src/orchestration-v2/http.ts index 478947320fc1c..b50ddb17d3756 100644 --- a/apps/server/src/orchestration-v2/http.ts +++ b/apps/server/src/orchestration-v2/http.ts @@ -9,6 +9,8 @@ import { TurnItemId, type OrchestrationProjectShell, } from "@t3tools/contracts"; +// T3-CUSTOM(expbkt3): shell snapshot reads run on the SQLite reader worker. +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; import * as Effect from "effect/Effect"; // T3-CUSTOM(expbkt3): resolve the bound operator. import * as Option from "effect/Option"; @@ -145,16 +147,20 @@ export const layer = HttpApiBuilder.group( ); const loadShellSnapshot = Effect.fn("http.orchestration.loadShellSnapshot")(function* () { - const base = yield* sql.withTransaction( - Effect.gen(function* () { - const threads = yield* threadManagement.getShellSnapshot({ location: "active" }); - return buildActiveShellSnapshot({ - projects: yield* projectStore.listShells(), - threads, - snapshotSequence: yield* applicationEvents.latestApplicationSequence, - }); - }), - ); + // T3-CUSTOM(expbkt3): read on the SQLite reader worker, off the event loop and the + // writer (upstream does this in loadShellSnapshotParts, which stage does not have). + const base = yield* sql + .withTransaction( + Effect.gen(function* () { + const threads = yield* threadManagement.getShellSnapshot({ location: "active" }); + return buildActiveShellSnapshot({ + projects: yield* projectStore.listShells(), + threads, + snapshotSequence: yield* applicationEvents.latestApplicationSequence, + }); + }), + ) + .pipe(NodeSqliteClient.readOnly); const projects = yield* enrichProjectShells(base.projects); // T3-CUSTOM(expbkt3): HTTP shell visibility matches the native socket stream. return filterNativeShell({ ...base, projects }, yield* currentActorUserId); diff --git a/apps/server/src/orchestration-v2/releasedSessionSubagents.expbkt3.ts b/apps/server/src/orchestration-v2/releasedSessionSubagents.expbkt3.ts new file mode 100644 index 0000000000000..bb4ed0f661a61 --- /dev/null +++ b/apps/server/src/orchestration-v2/releasedSessionSubagents.expbkt3.ts @@ -0,0 +1,288 @@ +// T3-CUSTOM(expbkt3): a released provider session settles the subagents it leaves open. +import type { + NodeId, + OrchestrationV2DomainEvent, + OrchestrationV2Subagent, + OrchestrationV2ThreadProjection, + ProviderDriverKind, + ProviderInstanceId, + ProviderSessionId, + ThreadId, +} from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; + +import type { EventSinkV2Shape } from "./EventSink.ts"; +import type { IdAllocatorV2Shape } from "./IdAllocator.ts"; +import type { ProjectionRuntimeRecoveryState, ProjectionStoreV2Shape } from "./ProjectionStore.ts"; +import type { ProviderSessionReleaseReason } from "./ProviderSessionManager.ts"; + +type TurnItem = OrchestrationV2ThreadProjection["turnItems"][number]; +type ExecutionNode = OrchestrationV2ThreadProjection["nodes"][number]; + +export interface ReleasedSessionSubagentsInput { + readonly projectionStore: ProjectionStoreV2Shape; + readonly idAllocator: IdAllocatorV2Shape; + readonly eventSink: EventSinkV2Shape; + readonly providerSessionId: ProviderSessionId; + readonly reason: ProviderSessionReleaseReason; + readonly driver: ProviderDriverKind; + 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. */ + readonly liveThreadIds: Effect.Effect>; + /** Work that started after the release belongs to a later session. */ + readonly releasedAt: DateTime.Utc; +} + +const isOpenStatus = (status: string): boolean => + status === "pending" || status === "running" || status === "waiting"; + +/** Same as startup recovery: a closed card also closes its form and stops streaming. */ +const cancelledItem = (item: TurnItem, now: DateTime.Utc): TurnItem => + item.type === "secret_request" && item.secretStatus === "pending" + ? { ...item, status: "cancelled", secretStatus: "cancelled", completedAt: now, updatedAt: now } + : { + ...item, + status: "cancelled", + completedAt: now, + updatedAt: now, + ...(item.type === "reasoning" || item.type === "assistant_message" + ? { streaming: false } + : {}), + }; + +/** + * Settles the provider-native subagents a released provider session leaves open. + * + * A subagent runs inside its parent's provider process. When that process goes + * away between turns (an account rebalance or mode change, a hard limit, a + * crash), nothing reports the subagent's end: its record, the parent's card and + * its child thread stay running until a server restart. Startup recovery + * (ProviderRuntimeRecoveryService) settles them after a restart; this settles + * them for one released session, with the same events and the same reads. + * + * 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 + * overwrite the cancelled status. + * + * A server shutdown is left to startup recovery, which also wakes a settled + * run whose background work the restart cancelled; settled here, that work + * would no longer look lost to it. + * + * Never fails: the release must complete either way, and startup recovery is + * the fallback for a write that did not land. + */ +export const settleReleasedSessionSubagents = ( + input: ReleasedSessionSubagentsInput, +): Effect.Effect => + Effect.gen(function* () { + if (input.reason === "server_shutdown") return; + const liveThreadIds = yield* input.liveThreadIds; + const now = yield* DateTime.now; + const events: Array = []; + const settledSubagentIds: Array = []; + const allocateEventId = (threadId: ThreadId) => + input.idAllocator.allocate.event({ threadId, providerSessionId: input.providerSessionId }); + const startedByRelease = (startedAt: DateTime.Utc | null, fallback: DateTime.Utc | null) => { + const at = startedAt ?? fallback; + return at === null || DateTime.isLessThanOrEqualTo(at, input.releasedAt); + }; + const ranInReleasedSession = ( + subagent: OrchestrationV2Subagent, + projection: ProjectionRuntimeRecoveryState, + ) => { + if ( + subagent.origin !== "provider_native" || + !isOpenStatus(subagent.status) || + subagent.driver !== input.driver || + subagent.providerInstanceId !== input.providerInstanceId || + !startedByRelease(subagent.startedAt, subagent.updatedAt) + ) { + return false; + } + // 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; + const providerThread = projection.providerThreads.find( + (candidate) => candidate.id === subagent.providerThreadId, + ); + // The recovery read loads only live-looking provider threads: one it did + // not load may belong to another shared session that still runs it. + return ( + providerThread !== undefined && + (providerThread.providerSessionId == null || + providerThread.providerSessionId === input.providerSessionId) + ); + }; + + const settleNode = Effect.fnUntraced(function* (node: ExecutionNode) { + events.push({ + id: yield* allocateEventId(node.threadId), + type: "node.updated", + threadId: node.threadId, + ...(node.runId === null ? {} : { runId: node.runId }), + nodeId: node.id, + driver: input.driver, + providerInstanceId: input.providerInstanceId, + occurredAt: now, + payload: { ...node, status: "cancelled", completedAt: now }, + }); + }); + const settleItem = Effect.fnUntraced(function* (item: TurnItem) { + events.push({ + id: yield* allocateEventId(item.threadId), + type: "turn-item.updated", + threadId: item.threadId, + ...(item.runId === null ? {} : { runId: item.runId }), + ...(item.nodeId === null ? {} : { nodeId: item.nodeId }), + driver: input.driver, + providerInstanceId: input.providerInstanceId, + occurredAt: now, + payload: cancelledItem(item, now), + }); + }); + + /** + * Settles one thread's subagents, and for a provider-native child thread its + * 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) { + // 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 + // check falls through to the full read. + if (!isChild) { + const records = yield* input.projectionStore + .getThreadRecords(threadId, ["subagents"]) + .pipe(Effect.option); + if ( + records._tag === "Some" && + !records.value.subagents.some( + (subagent) => + subagent.origin === "provider_native" && + isOpenStatus(subagent.status) && + subagent.driver === input.driver && + subagent.providerInstanceId === input.providerInstanceId, + ) + ) { + return []; + } + } + // The thread's open subagents, nodes and items, as startup recovery reads them. + const projection = yield* input.projectionStore + .getRuntimeRecoveryProjection(threadId) + .pipe(Effect.catchTags({ ProjectionStoreThreadNotFoundError: () => Effect.succeed(null) })); + // A deleted thread has nothing left to show as running. + if (projection === null) return []; + const settledNodeIds = new Set(); + const settledItemIds = new Set(); + const childThreadIds: Array = []; + for (const subagent of projection.subagents) { + if (!ranInReleasedSession(subagent, projection)) continue; + settledSubagentIds.push(subagent.id); + events.push({ + id: yield* allocateEventId(subagent.threadId), + type: "subagent.updated", + threadId: subagent.threadId, + ...(subagent.runId === null ? {} : { runId: subagent.runId }), + nodeId: subagent.id, + driver: subagent.driver, + providerInstanceId: subagent.providerInstanceId, + occurredAt: now, + payload: { ...subagent, status: "cancelled", completedAt: now, updatedAt: now }, + }); + // The parent's card for this subagent. + for (const item of projection.turnItems) { + if ( + item.type !== "subagent" || + item.subagentId !== subagent.id || + !isOpenStatus(item.status) || + settledItemIds.has(item.id) + ) { + continue; + } + settledItemIds.add(item.id); + yield* settleItem(item); + } + const node = projection.nodes.find( + (candidate) => candidate.id === subagent.id && isOpenStatus(candidate.status), + ); + if (node !== undefined && !settledNodeIds.has(node.id)) { + settledNodeIds.add(node.id); + yield* settleNode(node); + } + if (subagent.childThreadId !== null) childThreadIds.push(subagent.childThreadId); + } + if (isChild) { + // A provider-native child thread has no runs: its work is a runless root + // turn (shown as Working) and the items under it. + for (const node of projection.nodes) { + if ( + node.kind !== "root_turn" || + node.runId !== null || + !isOpenStatus(node.status) || + settledNodeIds.has(node.id) || + !startedByRelease(node.startedAt, null) + ) { + continue; + } + settledNodeIds.add(node.id); + yield* settleNode(node); + for (const item of projection.turnItems) { + if ( + item.nodeId !== node.id || + item.runId !== null || + !isOpenStatus(item.status) || + settledItemIds.has(item.id) + ) { + continue; + } + settledItemIds.add(item.id); + yield* settleItem(item); + } + } + } + return childThreadIds; + }); + + const visited = new Set(); + let pending: Array<{ readonly threadId: ThreadId; readonly isChild: boolean }> = []; + for (const threadId of input.threadIds) { + if (!liveThreadIds.has(threadId)) pending.push({ threadId, isChild: false }); + } + while (pending.length > 0) { + const next: typeof pending = []; + for (const { threadId, isChild } of pending) { + if (visited.has(threadId)) continue; + visited.add(threadId); + const childThreadIds = yield* settleThread(threadId, isChild); + for (const childThreadId of childThreadIds) { + if (!liveThreadIds.has(childThreadId)) + next.push({ threadId: childThreadId, isChild: true }); + } + } + pending = next; + } + + if (events.length === 0) return; + yield* input.eventSink.write({ events }); + yield* Effect.logInfo("orchestration-v2.provider-session-release-subagents-cancelled", { + providerSessionId: input.providerSessionId, + subagentIds: settledSubagentIds, + }); + }).pipe( + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.interrupt + : Effect.logWarning("orchestration-v2.provider-session-release-subagents-failed", { + providerSessionId: input.providerSessionId, + cause: Cause.pretty(cause), + }), + ), + ); diff --git a/apps/server/src/persistence/Sqlite.ts b/apps/server/src/persistence/Sqlite.ts index 0f248b5a23877..142ff045946ef 100644 --- a/apps/server/src/persistence/Sqlite.ts +++ b/apps/server/src/persistence/Sqlite.ts @@ -35,6 +35,7 @@ export const layerFromPath = Effect.fn("makeSqlitePersistenceLive")(function* (d layerSetup, NodeSqliteClient.layer({ filename: dbPath, + readerWorker: true, spanAttributes: { "db.name": path.basename(dbPath), "service.name": "t3code-server", diff --git a/apps/server/src/provider/ForkProviderSessions.expbkt3.ts b/apps/server/src/provider/ForkProviderSessions.expbkt3.ts index aca78e36042aa..dfae2b42b5cbd 100644 --- a/apps/server/src/provider/ForkProviderSessions.expbkt3.ts +++ b/apps/server/src/provider/ForkProviderSessions.expbkt3.ts @@ -6,6 +6,7 @@ import { type ProviderSession, type ThreadId, } from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; @@ -37,6 +38,19 @@ export class ForkProviderSessions extends Context.Service< ForkProviderSessions, { readonly stopSession: (input: { readonly threadId: ThreadId }) => Effect.Effect; + /** + * Whether a live session of the thread still runs background work between + * turns: a subagent, a background command or monitor, or a wake not yet + * delivered. Stopping that session ends the work. A failed check answers + * true, as the session manager's own configuration refresh does: a restart + * can wait, but killed work does not come back. + * + * Each check counts as activity on the thread's own sessions (their idle + * release starts again), so poll it only for threads that wait on it. + */ + readonly hasLiveBackgroundWork: (input: { + readonly threadId: ThreadId; + }) => Effect.Effect; readonly listSessions: () => Effect.Effect>; readonly streamEvents: Stream.Stream; } @@ -70,6 +84,26 @@ export const layer = Layer.effect( yield* sessions.release({ providerSessionId, reason: "manual_shutdown" }); } }).pipe(Effect.orDie), + hasLiveBackgroundWork: ({ threadId }) => + Effect.gen(function* () { + const { providerThreads } = yield* projection.getThreadRecords(threadId, [ + "providerThreads", + ]); + for (const providerSessionId of new Set( + providerThreads.flatMap((thread) => + thread.providerSessionId === null ? [] : [thread.providerSessionId], + ), + )) { + const live = yield* sessions.get(providerSessionId); + if (live._tag === "None" || live.value.hasPendingBackgroundWork === undefined) continue; + if (yield* live.value.hasPendingBackgroundWork) return true; + } + return false; + }).pipe( + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) ? Effect.interrupt : Effect.succeed(true), + ), + ), listSessions: () => Effect.gen(function* () { const shell = yield* projection.getShellSnapshot(); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index b1383df1e47c0..b959420cdbc10 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -43,6 +43,7 @@ import * as SessionArchiveSweeper from "./sessionArchive/SessionArchiveSweeper.t import * as ThreadWorkspaceGroups from "./persistence/ThreadWorkspaceGroups.ts"; import { ArchiveExportReactorLive } from "./orchestration-v2/Layers/ArchiveExportReactor.ts"; import { ArchiveExportReactor } from "./orchestration-v2/Services/ArchiveExportReactor.ts"; +import * as SubagentArchiveCascade from "./orchestration-v2/SubagentArchiveCascade.expbkt3.ts"; import { ProjectionSnapshotQueryLive } from "./orchestration-v2/Layers/ProjectionSnapshotQuery.ts"; import { OrchestrationEngineLive } from "./orchestration-v2/Layers/OrchestrationEngine.ts"; import * as TurnStartBootstrap from "./orchestration-v2/turnStartBootstrap.expbkt3.ts"; @@ -658,6 +659,7 @@ const layerRuntimeCoreDependenciesBase = Layer.mergeAll( layerProviderInstallationRefresh, ReplayMarkers.layer, SessionActivityReactor.layer, // T3-CUSTOM(expbkt3): session time-in-state metrics (TEC-1631). + SubagentArchiveCascade.layer, // T3-CUSTOM(expbkt3): provider subagents follow their parent's archive. ).pipe( // Core Services Layer.provideMerge(layerOrchestrationApplication), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index decc4ed912d76..0c424bc476d19 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -143,6 +143,8 @@ import { FORK_WS_RPCS, } from "@t3tools/contracts"; import { resolveServerBackgroundActivitySettings } from "@t3tools/shared/backgroundActivitySettings"; +// T3-CUSTOM(expbkt3): shell reads run on the SQLite reader worker. +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; import { HttpRouter, HttpServerRequest, @@ -1015,16 +1017,20 @@ export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subsc }, ); const loadSnapshot = Effect.fn("ws.orchestrationV2.loadShellSnapshot")(function* () { - const base = yield* sql.withTransaction( - Effect.gen(function* () { - const threads = yield* threadManagement.getShellSnapshot({ location: "active" }); - return buildActiveShellSnapshot({ - projects: yield* projects.listShells(), - threads, - snapshotSequence: yield* applicationEvents.latestApplicationSequence, - }); - }), - ); + // T3-CUSTOM(expbkt3): read on the SQLite reader worker, off the event loop and the + // writer (upstream does this in loadShellSnapshotParts, which stage does not have). + const base = yield* sql + .withTransaction( + Effect.gen(function* () { + const threads = yield* threadManagement.getShellSnapshot({ location: "active" }); + return buildActiveShellSnapshot({ + projects: yield* projects.listShells(), + threads, + snapshotSequence: yield* applicationEvents.latestApplicationSequence, + }); + }), + ) + .pipe(NodeSqliteClient.readOnly); const enriched = yield* enrichProjectShells(base.projects); return { snapshot: { ...base, projects: enriched.projects } as OrchestrationV2ShellSnapshot, @@ -1923,6 +1929,8 @@ const layerWsRpc = ( }), ) .pipe( + // T3-CUSTOM(expbkt3): read on the SQLite reader worker; enrichment runs after it. + NodeSqliteClient.readOnly, Effect.flatMap((snapshot) => enrichProjectShells(snapshot.projects).pipe( Effect.map(({ projects }) => ({ ...snapshot, projects })), diff --git a/packages/shared/src/nodeSqliteClient.test.ts b/packages/shared/src/nodeSqliteClient.test.ts index 83fe851ac8ad5..040baa7ba1304 100644 --- a/packages/shared/src/nodeSqliteClient.test.ts +++ b/packages/shared/src/nodeSqliteClient.test.ts @@ -1,14 +1,67 @@ import * as NodeSqlite from "node:sqlite"; +import type * as NodeWorkerThreads from "node:worker_threads"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; +import * as Logger from "effect/Logger"; import * as Path from "effect/Path"; import * as SqlClient from "effect/sql/SqlClient"; +import * as TestClock from "effect/testing/TestClock"; +import { vi } from "vite-plus/test"; import * as SqliteClient from "./nodeSqliteClient.ts"; +const startedWorkers = vi.hoisted((): Array => []); +// When on, workers run a script that never announces, like a worker stuck opening the file. +const silentWorkers = vi.hoisted(() => ({ on: false, readyOnTerminate: false })); + +vi.mock("node:worker_threads", async (importOriginal) => { + const actual = await importOriginal(); + class RecordedWorker extends actual.Worker { + constructor(...args: ConstructorParameters) { + super( + ...((silentWorkers.on + ? ["setInterval(() => {}, 1 << 30);", { eval: true }] + : args) as ConstructorParameters), + ); + startedWorkers.push(this); + } + override terminate() { + // A "ready" already queued when termination starts reaches the parent before "exit". + if (silentWorkers.readyOnTerminate) this.emit("message", { type: "ready" }); + return super.terminate(); + } + } + return { ...actual, Worker: RecordedWorker }; +}); + +const withReaderClient = ( + use: (sql: SqlClient.SqlClient) => Effect.Effect, +) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-sqlite-reader-" }); + const filename = path.join(directory, "state.sqlite"); + return yield* Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql`PRAGMA journal_mode = WAL`; + yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`; + yield* sql`INSERT INTO entries VALUES ('first')`; + return yield* use(sql); + }).pipe(Effect.provide(SqliteClient.layer({ filename, readerWorker: true }))); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)); + +const countEntries = (sql: SqlClient.SqlClient) => + Effect.map( + sql<{ count: number }>`SELECT COUNT(*) AS count FROM entries`, + (rows) => rows[0]?.count, + ); + const layer = it.layer(SqliteClient.layer({ filename: ":memory:" })); layer("NodeSqliteClient", (it) => { @@ -182,3 +235,318 @@ it.effect( }).pipe(Effect.provide(SqliteClient.layer({ filename }))); }).pipe(Effect.provide(NodeServices.layer)), ); + +it.effect("lets the writer commit while a read-only transaction is open", () => + withReaderClient((sql) => + Effect.gen(function* () { + const snapshotTaken = yield* Deferred.make(); + const writer = yield* Effect.forkChild( + Deferred.await(snapshotTaken).pipe( + Effect.andThen(sql`INSERT INTO entries VALUES ('second')`), + ), + ); + const counts = yield* sql + .withTransaction( + Effect.gen(function* () { + const before = yield* countEntries(sql); + yield* Deferred.succeed(snapshotTaken, undefined); + // Holding the writer lock here would deadlock this join. + yield* Fiber.join(writer); + const after = yield* countEntries(sql); + return { before, after }; + }), + ) + .pipe(SqliteClient.readOnly); + + // The reader keeps the snapshot it began with, and later reads see the commit. + assert.deepEqual(counts, { before: 1, after: 1 }); + assert.equal(yield* countEntries(sql).pipe(SqliteClient.readOnly), 2); + }), + ), +); + +it.effect("rejects a write in a read-only transaction", () => + withReaderClient((sql) => + Effect.gen(function* () { + const error = yield* sql + .withTransaction(sql`INSERT INTO entries VALUES ('blocked')`) + .pipe(SqliteClient.readOnly, Effect.flip); + + assert.equal(error._tag, "SqlError"); + assert.include(String(error.reason.cause), "readonly database"); + assert.equal(yield* countEntries(sql), 1); + }), + ), +); + +it.effect("keeps the reader read-only even if a statement turns query_only off", () => + withReaderClient((sql) => + Effect.gen(function* () { + const error = yield* sql + .withTransaction( + Effect.gen(function* () { + yield* sql`PRAGMA query_only = OFF`; + yield* sql`INSERT INTO entries VALUES ('blocked')`; + }), + ) + .pipe(SqliteClient.readOnly, Effect.flip); + + assert.equal(error._tag, "SqlError"); + assert.include(String(error.reason.cause), "readonly database"); + assert.equal(yield* countEntries(sql), 1); + }), + ), +); + +it.effect("keeps read-only work inside a write transaction on the writer", () => + withReaderClient((sql) => + sql.withTransaction( + Effect.gen(function* () { + yield* sql`INSERT INTO entries VALUES ('uncommitted')`; + // Only the writer can see its own uncommitted row. + const count = yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly); + assert.equal(count, 2); + }), + ), + ), +); + +it.effect("fails a read-only transaction whose reader stops, then starts a new reader", () => + withReaderClient((sql) => + Effect.gen(function* () { + const startedBefore = startedWorkers.length; + const error = yield* sql + .withTransaction( + Effect.gen(function* () { + yield* countEntries(sql); + yield* Effect.promise(() => startedWorkers.at(-1)!.terminate()); + return yield* countEntries(sql); + }), + ) + .pipe(SqliteClient.readOnly, Effect.flip); + + assert.equal(error._tag, "SqlError"); + assert.include(String(error.reason.cause), "Database reader exited"); + assert.equal(yield* countEntries(sql).pipe(SqliteClient.readOnly), 1); + assert.equal(startedWorkers.length, startedBefore + 2); + }), + ), +); + +it.effect("uses the writer for read-only work on an in-memory database", () => + Effect.gen(function* () { + const startedBefore = startedWorkers.length; + const sql = yield* SqlClient.SqlClient; + yield* sql`CREATE TABLE memory_entries(name TEXT NOT NULL)`; + yield* sql`INSERT INTO memory_entries VALUES ('kept')`; + + const rows = yield* sql + .withTransaction(sql<{ name: string }>`SELECT name FROM memory_entries`) + .pipe(SqliteClient.readOnly); + + assert.deepEqual(rows, [{ name: "kept" }]); + assert.equal(startedWorkers.length, startedBefore); + }).pipe(Effect.provide(SqliteClient.layer({ filename: ":memory:", readerWorker: true }))), +); + +it.effect("keeps a nested transaction inside a read-only transaction on the reader", () => + withReaderClient((sql) => + Effect.gen(function* () { + const snapshotTaken = yield* Deferred.make(); + const writer = yield* Effect.forkChild( + Deferred.await(snapshotTaken).pipe( + Effect.andThen(sql`INSERT INTO entries VALUES ('second')`), + ), + ); + // Callers wrap a store read that opens its own transaction, as the + // shell snapshot routes do around getShellSnapshot. + const counts = yield* sql + .withTransaction( + Effect.gen(function* () { + const outer = yield* countEntries(sql); + yield* Deferred.succeed(snapshotTaken, undefined); + yield* Fiber.join(writer); + const nested = yield* sql.withTransaction(countEntries(sql)); + return { outer, nested }; + }), + ) + .pipe(SqliteClient.readOnly); + + assert.deepEqual(counts, { outer: 1, nested: 1 }); + }), + ), +); + +const withFileClient = ( + setup: ( + sql: SqlClient.SqlClient, + filename: string, + ) => Effect.Effect, +) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-sqlite-reader-off-" }); + const filename = path.join(directory, "state.sqlite"); + return yield* Effect.gen(function* () { + return yield* setup(yield* SqlClient.SqlClient, filename); + }).pipe(Effect.provide(SqliteClient.layer({ filename, readerWorker: true }))); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)); + +it.effect("uses the writer when the reader cannot open the database", () => + withFileClient((sql, filename) => + Effect.gen(function* () { + yield* sql`PRAGMA journal_mode = WAL`; + yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`; + yield* sql`INSERT INTO entries VALUES ('kept')`; + // The writer keeps its open file; a read-only open of the path now fails. + yield* (yield* FileSystem.FileSystem).remove(filename); + const startedBefore = startedWorkers.length; + + assert.equal(yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), 1); + assert.equal(yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), 1); + assert.equal(startedWorkers.length, startedBefore + 1); + }), + ), +); + +it.effect("uses the writer for a database that is not in WAL mode", () => + withFileClient((sql) => + Effect.gen(function* () { + yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`; + const startedBefore = startedWorkers.length; + + // A write inside readOnly only succeeds on the writer: the reader is read-only. + yield* sql + .withTransaction(sql`INSERT INTO entries VALUES ('on the writer')`) + .pipe(SqliteClient.readOnly); + assert.equal(yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), 1); + assert.equal(startedWorkers.length, startedBefore + 1); + }), + ), +); + +const captureWarnings = () => { + const warnings: Array = []; + const logger = Logger.make(({ logLevel, message }) => { + if (logLevel === "Warn") warnings.push(message); + }); + return { warnings, layer: Logger.layer([logger], { mergeWithExisting: false }) }; +}; + +const silently = (effect: Effect.Effect) => + Effect.acquireUseRelease( + Effect.sync(() => { + silentWorkers.on = true; + }), + () => effect, + () => + Effect.sync(() => { + silentWorkers.on = false; + silentWorkers.readyOnTerminate = false; + }), + ); + +it.effect("turns the reader off when its worker never opens the database", () => { + const { warnings, layer: logs } = captureWarnings(); + return withFileClient((sql) => + Effect.gen(function* () { + yield* sql`PRAGMA journal_mode = WAL`; + yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`; + yield* sql`INSERT INTO entries VALUES ('kept')`; + const read = yield* silently( + Effect.gen(function* () { + const fiber = yield* Effect.forkChild( + sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), + ); + yield* Effect.yieldNow; + yield* TestClock.adjust("10 seconds"); + return yield* Fiber.join(fiber); + }), + ); + + assert.equal(read, 1); + assert.equal(yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), 1); + assert.lengthOf(warnings, 1); + }), + ).pipe(Effect.provide(logs)); +}); + +it.effect("lets a read be cancelled while its worker is still starting", () => + withFileClient((sql) => + Effect.gen(function* () { + yield* sql`PRAGMA journal_mode = WAL`; + yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`; + yield* silently( + Effect.gen(function* () { + const fiber = yield* Effect.forkChild( + sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), + ); + yield* Effect.yieldNow; + // Never returns if the wait for the worker were uninterruptible: the + // test clock has not moved, so the startup deadline cannot fire. + yield* Fiber.interrupt(fiber); + }), + ); + + // The client stays usable: the next read waits out the deadline, then uses the writer. + const next = yield* Effect.forkChild( + sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), + ); + yield* Effect.yieldNow; + yield* TestClock.adjust("10 seconds"); + assert.equal(yield* Fiber.join(next), 0); + }), + ), +); + +it.effect("logs one warning when concurrent first reads find the reader unavailable", () => { + const { warnings, layer: logs } = captureWarnings(); + return withFileClient((sql, filename) => + Effect.gen(function* () { + yield* sql`PRAGMA journal_mode = WAL`; + yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`; + yield* (yield* FileSystem.FileSystem).remove(filename); + const startedBefore = startedWorkers.length; + + const counts = yield* Effect.all( + [ + sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), + sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), + ], + { concurrency: "unbounded" }, + ); + + assert.deepEqual(counts, [0, 0]); + assert.equal(startedWorkers.length, startedBefore + 1); + assert.lengthOf(warnings, 1); + }), + ).pipe(Effect.provide(logs)); +}); + +it.effect("does not hand out a worker that reports ready after another read timed it out", () => + withFileClient((sql) => + Effect.gen(function* () { + yield* sql`PRAGMA journal_mode = WAL`; + yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`; + yield* sql`INSERT INTO entries VALUES ('kept')`; + const read = () => sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly); + const counts = yield* silently( + Effect.gen(function* () { + const first = yield* Effect.forkChild(read()); + yield* Effect.yieldNow; + yield* TestClock.adjust("5 seconds"); + const second = yield* Effect.forkChild(read()); + yield* Effect.yieldNow; + // The first read reaches its deadline and retires the worker; the second + // is still waiting when the worker's "ready" arrives, before it exits. + silentWorkers.readyOnTerminate = true; + yield* TestClock.adjust("5 seconds"); + return [yield* Fiber.join(first), yield* Fiber.join(second)]; + }), + ); + + assert.deepEqual(counts, [1, 1]); + }), + ), +); diff --git a/packages/shared/src/nodeSqliteClient.ts b/packages/shared/src/nodeSqliteClient.ts index 6b05d8ac6ca11..3045cac5fd273 100644 --- a/packages/shared/src/nodeSqliteClient.ts +++ b/packages/shared/src/nodeSqliteClient.ts @@ -5,6 +5,7 @@ * @module SqliteClient */ import * as NodeSqlite from "node:sqlite"; +import * as NodeWorkerThreads from "node:worker_threads"; import * as Cache from "effect/Cache"; import * as Duration from "effect/Duration"; @@ -13,6 +14,7 @@ import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import { identity } from "effect/Function"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Predicate from "effect/Predicate"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; @@ -36,8 +38,38 @@ export interface SqliteClientConfig { readonly spanAttributes?: Record | undefined; readonly transformResultNames?: ((str: string) => string) | undefined; readonly transformQueryNames?: ((str: string) => string) | undefined; + // T3-CUSTOM(expbkt3): BEGIN upstream 8659fcf145 (#19226) as cherry-picked onto stage. + // Later upstream commits to this file (#19305, #19324, #19417, #19425, #19559) are + // not on stage, so these lines differ from upstream main. Upstream owns all of it: + // on the next upstream merge take upstream's side and drop the markers. + /** + * Run transactions marked with `readOnly` on a second connection in a worker + * thread, so long reads block neither the event loop nor writes. Ignored for + * in-memory databases, which a second connection cannot share. + */ + readonly readerWorker?: boolean | undefined; + /** How long a reader worker may take to open the database before reads use the writer. Default 10 seconds. */ + readonly readerStartupTimeout?: Duration.Input | undefined; } +/** + * Set by `readOnly`. A top-level transaction started while it is true runs on + * the reader worker when the client has one, and on the writer otherwise. + */ +const ReadOnly = Context.Reference("@t3tools/shared/nodeSqliteClient/ReadOnly", { + defaultValue: () => false, +}); + +/** + * Marks the SQL in `effect` as read-only. Wrap the whole transaction, as in + * `sql.withTransaction(reads).pipe(readOnly)`. Inside an open write + * transaction the reads keep using the writer, and a write attempted on the + * reader fails with "attempt to write a readonly database". + */ +export const readOnly = (effect: Effect.Effect): Effect.Effect => + Effect.provideService(effect, ReadOnly, true); +// T3-CUSTOM(expbkt3): END + export class UnsupportedNodeSqliteVersionError extends Schema.TaggedError()( "UnsupportedNodeSqliteVersionError", { @@ -99,6 +131,308 @@ const classifyError = (cause: unknown, message: string, operation: string) => { return classifySqliteError(cause, { message, operation }); }; +// Evaluated source instead of a worker file, so the reader also starts from +// bundles and single-executable builds that have no module file on disk. +const readerWorkerSource = ` +const { parentPort, workerData } = require("node:worker_threads"); +const { DatabaseSync } = require("node:sqlite"); +let db; +try { + // A read-only open cannot be undone by SQL, unlike PRAGMA query_only. + const next = new DatabaseSync(workerData.filename, { + readOnly: true, + allowExtension: workerData.allowExtension, + }); + next.exec("PRAGMA busy_timeout = 5000;"); + // Without WAL a reader's shared lock blocks the writer's commits, so refuse. + const mode = String(next.prepare("PRAGMA journal_mode").get().journal_mode).toLowerCase(); + if (mode !== "wal") throw new Error("database uses journal_mode " + mode + ", not wal"); + db = next; + parentPort.postMessage({ type: "ready" }); +} catch (error) { + parentPort.postMessage({ type: "unavailable", message: String(error && error.message ? error.message : error) }); +} +const statements = new Map(); +const prepare = (sql, cached) => { + if (!cached) return db.prepare(sql); + let statement = statements.get(sql); + if (statement === undefined) { + statement = db.prepare(sql); + if (statements.size >= workerData.prepareCacheSize) statements.delete(statements.keys().next().value); + } else { + statements.delete(sql); + } + statements.set(sql, statement); + return statement; +}; +parentPort.on("message", ({ id, sql, params, cached, values, raw, safeIntegers }) => { + let operation = "open"; + try { + if (db === undefined) throw new Error("database reader is unavailable"); + operation = "prepare"; + const statement = prepare(sql, cached); + operation = "execute"; + statement.setReadBigInts(safeIntegers); + if (statement.columns().length > 0) { + statement.setReturnArrays(values); + parentPort.postMessage({ id, rows: statement.all(...params) }); + } else { + const result = statement.run(...params); + parentPort.postMessage({ id, rows: raw ? result : [] }); + } + } catch (error) { + parentPort.postMessage({ + id, + error: { + operation, + message: String(error && error.message ? error.message : error), + code: error && error.code, + errcode: error && error.errcode, + errstr: error && error.errstr, + }, + }); + } +}); +`; + +interface ReaderRequest { + readonly sql: string; + readonly params: ReadonlyArray; + readonly cached: boolean; + readonly values: boolean; + readonly raw: boolean; +} + +type ReaderStartup = + | { readonly type: "ready" } + | { readonly type: "unavailable"; readonly message: string }; + +type ReaderReply = + | { readonly id: number; readonly rows: ReadonlyArray } + | { + readonly id: number; + readonly error: { + readonly operation: "open" | "prepare" | "execute"; + readonly message: string; + readonly code?: unknown; + readonly errcode?: unknown; + readonly errstr?: unknown; + }; + }; + +const readerFailureMessages = { + open: "Failed to open database", + prepare: "Failed to prepare statement", + execute: "Failed to execute statement", +} as const; + +const failed = (cause: unknown, message: string, operation: string) => + new SqlError({ reason: classifyError(cause, message, operation) }); + +// COMMIT, ROLLBACK and ROLLBACK TO SAVEPOINT. On a reader they change nothing, +// so they succeed once the worker is gone instead of failing the cleanup. +const endsReadTransaction = (sql: string) => /^\s*ROLLBACK\b|^\s*COMMIT\b/i.test(sql); + +// T3-CUSTOM(expbkt3): BEGIN upstream 8659fcf145 (#19226), see the note on readerWorker. +/** + * Read-only connections served by one worker thread at a time. Each worker + * gets its own `Connection`, so a transaction stays on the worker it began on: + * once that worker exits, its statements fail rather than silently continuing + * outside the transaction on a replacement. The next acquire starts a new one. + * A worker that cannot open the database read-only, or finds it outside WAL + * mode, turns the reader off for the life of the client: reads use the writer. + */ +const makeReader = Effect.fnUntraced(function* (options: SqliteClientConfig) { + let current: + | { + readonly worker: NodeWorkerThreads.Worker; + readonly connection: Connection; + readonly started: Promise; + } + | undefined; + let unavailable = false; + let nextRequestId = 0; + + const start = () => { + const worker = new NodeWorkerThreads.Worker(readerWorkerSource, { + eval: true, + workerData: { + filename: options.filename, + allowExtension: options.allowExtension ?? false, + prepareCacheSize: options.prepareCacheSize ?? 200, + }, + }); + let announce: (startup: ReaderStartup) => void = () => {}; + const started = new Promise((resolve) => { + announce = resolve; + }); + let stoppedMessage: string | undefined; + const pending = new Map< + number, + { readonly endsTransaction: boolean; readonly resume: (reply: ReaderReply) => void } + >(); + const stop = (message: string) => { + stoppedMessage = message; + announce({ type: "unavailable", message }); + if (current?.worker === worker) current = undefined; + for (const [id, request] of pending) { + request.resume( + request.endsTransaction + ? { id, rows: [] } + : { id, error: { operation: "execute", message } }, + ); + } + pending.clear(); + }; + worker.on("message", (reply: ReaderReply | ReaderStartup) => { + if ("type" in reply) { + announce(reply); + // An idle reader must not keep the process alive. + if (pending.size === 0) worker.unref(); + return; + } + const request = pending.get(reply.id); + if (request === undefined) return; + pending.delete(reply.id); + if (pending.size === 0) worker.unref(); + request.resume(reply); + }); + worker.on("error", (cause) => stop(`Database reader failed: ${cause.message}`)); + worker.on("exit", (code) => stop(`Database reader exited with code ${code}`)); + + const request = (input: ReaderRequest) => + Effect.withFiber, SqlError>((fiber) => { + const endsTransaction = endsReadTransaction(input.sql); + if (stoppedMessage !== undefined) { + return endsTransaction + ? Effect.succeed([]) + : Effect.fail( + failed(new Error(stoppedMessage), readerFailureMessages.execute, "execute"), + ); + } + const safeIntegers = Boolean(Context.get(fiber.context, Client.SafeIntegers)); + return Effect.callback, SqlError>((resume) => { + const id = nextRequestId++; + pending.set(id, { + endsTransaction, + resume: (reply) => { + if ("rows" in reply) { + resume(Effect.succeed(reply.rows)); + return; + } + const { operation, message, ...codes } = reply.error; + resume( + Effect.fail( + failed( + Object.assign(new Error(message), codes), + readerFailureMessages[operation], + operation, + ), + ), + ); + }, + }); + if (pending.size === 1) worker.ref(); + // oxlint-disable-next-line unicorn/require-post-message-target-origin -- Node workers do not accept a target origin. + worker.postMessage({ id, safeIntegers, ...input }); + }); + }); + + const connection = identity({ + execute(sql, params, rowTransform) { + const effect = request({ sql, params, cached: true, values: false, raw: false }); + return rowTransform ? Effect.map(effect, rowTransform) : effect; + }, + executeRaw(sql, params) { + return request({ sql, params, cached: true, values: false, raw: true }); + }, + executeValues(sql, params) { + return request({ sql, params, cached: true, values: true, raw: false }); + }, + executeValuesUnprepared(sql, params) { + return request({ sql, params: params ?? [], cached: false, values: true, raw: false }); + }, + executeUnprepared(sql, params, rowTransform) { + // A read-only connection rejects BEGIN IMMEDIATE and BEGIN EXCLUSIVE, + // and a deferred BEGIN is all a read snapshot needs. + const statement = /^\s*BEGIN\b/i.test(sql) ? "BEGIN" : sql; + const effect = request({ + sql: statement, + params: params ?? [], + cached: false, + values: false, + raw: false, + }); + return rowTransform ? Effect.map(effect, rowTransform) : effect; + }, + executeStream(_sql, _params) { + return Stream.die(new UnsupportedNodeSqliteOperationError()); + }, + }); + current = { worker, connection, started }; + return current; + }; + + yield* Effect.addFinalizer(() => + Effect.promise(async () => { + await current?.worker.terminate(); + }), + ); + + const startupTimeout = Duration.fromInputUnsafe( + options.readerStartupTimeout ?? Duration.seconds(10), + ); + + // Runs once: concurrent first acquires share one startup and may all see it fail. + const turnOff = (reader: typeof current, reason: string) => + Effect.suspend(() => { + if (reader !== undefined) { + if (current === reader) current = undefined; + void reader.worker.terminate(); + } + if (unavailable) return Effect.succeed(undefined); + unavailable = true; + return Effect.logWarning("SQLite reader worker unavailable; reads use the main connection", { + reason, + }).pipe(Effect.as(undefined)); + }); + + /** The live worker's connection once it is ready, or undefined to use the writer. */ + return Effect.suspend((): Effect.Effect => { + if (unavailable) return Effect.succeed(undefined); + let reader: NonNullable; + try { + reader = current ?? start(); + } catch (cause) { + return turnOff(undefined, String(cause)); + } + // SqlClient acquires transaction connections uninterruptibly. Waiting for the + // worker happens before any lock is taken, so let a cancelled request leave. + return Effect.flatMap( + Effect.interruptible( + Effect.timeoutOption( + Effect.promise(() => reader.started), + startupTimeout, + ), + ), + (startup): Effect.Effect => + // Another acquire may have timed out and retired this worker while this one + // waited; a late "ready" must not hand out a worker that is being terminated. + unavailable || current !== reader + ? Effect.succeed(undefined) + : Option.isSome(startup) && startup.value.type === "ready" + ? Effect.succeed(reader.connection) + : turnOff( + reader, + Option.isSome(startup) && startup.value.type === "unavailable" + ? startup.value.message + : `worker did not open the database within ${Duration.toMillis(startupTimeout)} ms`, + ), + ); + }); +}); +// T3-CUSTOM(expbkt3): END + const make = Effect.fn("makeWithDatabase")(function* ( options: SqliteClientConfig, ): Effect.fn.Return { @@ -261,16 +595,42 @@ const make = Effect.fn("makeWithDatabase")(function* ( const semaphore = yield* Semaphore.make(1); const connection = yield* makeConnection; + // T3-CUSTOM(expbkt3): BEGIN upstream 8659fcf145 (#19226), see the note on readerWorker. + const acquireReader = + options.readerWorker && options.filename !== ":memory:" && options.filename !== "" + ? yield* makeReader(options) + : undefined; + const readerSemaphore = yield* Semaphore.make(1); - const acquirer = semaphore.withPermits(1)(Effect.succeed(connection)); - const transactionAcquirer = Effect.uninterruptibleMask((restore) => { - const fiber = Fiber.getCurrent()!; - const scope = Context.getUnsafe(fiber.context, Scope.Scope); - return Effect.as( - Effect.tap(restore(semaphore.take(1)), () => Scope.addFinalizer(scope, semaphore.release(1))), - connection, - ); - }); + const writer = { lock: semaphore, connection }; + // Picks the reader only for `readOnly` work, and the writer whenever the + // reader is turned off. + const route = Effect.withFiber((fiber) => + acquireReader !== undefined && Context.get(fiber.context, ReadOnly) + ? Effect.map(acquireReader, (readerConnection) => + readerConnection === undefined + ? writer + : { lock: readerSemaphore, connection: readerConnection }, + ) + : Effect.succeed(writer), + ); + + const acquirer = Effect.flatMap(route, (target) => + target.lock.withPermits(1)(Effect.succeed(target.connection)), + ); + const transactionAcquirer = Effect.flatMap(route, (target) => + Effect.uninterruptibleMask((restore) => { + const fiber = Fiber.getCurrent()!; + const scope = Context.getUnsafe(fiber.context, Scope.Scope); + return Effect.as( + Effect.tap(restore(target.lock.take(1)), () => + Scope.addFinalizer(scope, target.lock.release(1)), + ), + target.connection, + ); + }), + ); + // T3-CUSTOM(expbkt3): END return yield* Client.make({ acquirer,