From 5e79b4b632ebe70d87193252d913f4a21d4d3d45 Mon Sep 17 00:00:00 2001 From: Pyseph <45090858+Pyseph@users.noreply.github.com> Date: Wed, 4 Nov 2026 14:02:47 +0100 Subject: [PATCH 1/7] fix(server): release a finished delegated child's provider session at once A delegate_task child usually runs once and is never settled, yet its provider session (one claude process plus its MCP servers) stays up for the 30 minute idle timeout. When a delegated child's run completes and its result is recorded, queue an effect that releases the child's one-process-per-thread session if nothing can resume it: no run or queued follow-up, no subagent or background task, no pending wake message. A later message to the child opens a new session as usual. The idle timeout for other threads is unchanged. --- .../DelegatedCompletionDelivery.test.ts | 56 +++++- .../src/orchestration-v2/EffectOutbox.ts | 4 + .../src/orchestration-v2/EffectWorker.test.ts | 1 + .../src/orchestration-v2/EffectWorker.ts | 11 ++ .../src/orchestration-v2/Orchestrator.ts | 20 +- .../ProviderSessionManager.test.ts | 174 ++++++++++++++++++ .../ProviderSessionManager.ts | 56 ++++++ .../ProviderTurnControlService.test.ts | 1 + .../ThreadManagementService.test.ts | 1 + 9 files changed, 322 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 3f0f9cf0342ee..67c97609cb524 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -25,6 +25,7 @@ import * as Effect from "effect/Effect"; import * as Deferred from "effect/Deferred"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; +import * as SqlClient from "effect/sql/SqlClient"; import * as Stream from "effect/Stream"; import * as Tracer from "effect/Tracer"; @@ -144,7 +145,7 @@ const layerTest = Layer.mergeAll( ), Layer.provide(McpSessionRegistryTestkit.layer), Layer.provide(McpProviderSessions.layer), - Layer.provide(SqlitePersistence.layerMemory), + Layer.provideMerge(SqlitePersistence.layerMemory), Layer.provide(layerCheckpointStoreTest), Layer.provide(layerServerConfig), Layer.provide(ServerSettings.layerTest()), @@ -2403,6 +2404,59 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => { }), ); + it.effect( + "queues the release of a completed child's provider session, not a cancelled one's", + () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sql = yield* SqlClient.SqlClient; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:release-delegate-parent"); + const projectId = ProjectId.make("project:release-delegate"); + const runId = RunId.make("run:release-delegate-parent"); + const rootNodeId = NodeId.make("node:release-delegate-parent"); + yield* seedParentWithTerminalTask({ + threadId, + projectId, + runId, + rootNodeId, + taskId: NodeId.make("node:release-delegate-seed"), + deliveryState: "acknowledged", + now, + }); + const child = (name: string, runStatus: "cancelled" | "completed") => + seedRestartCancelledChild({ + parentThreadId: threadId, + projectId, + parentRunId: runId, + rootNodeId, + name, + completionWake: "always", + continuationPending: false, + runStatus, + now, + }); + const completed = yield* child("release-delegate-completed", "completed"); + const cancelled = yield* child("release-delegate-cancelled", "cancelled"); + + yield* orchestrator.recoverDelegatedTasks; + + const queued = yield* sql<{ readonly thread_id: string }>` + SELECT thread_id FROM orchestration_v2_effect_outbox + WHERE effect_type = 'provider-session.release-finished-delegate' + AND thread_id LIKE 'thread:release-delegate-%' + `; + assert.deepEqual( + queued.map((row) => row.thread_id), + [completed.childThreadId], + ); + assert.notInclude( + queued.map((row) => row.thread_id), + cancelled.childThreadId, + ); + }), + ); + it.effect("holds a restart-cancelled child for its continuation's result", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/apps/server/src/orchestration-v2/EffectOutbox.ts b/apps/server/src/orchestration-v2/EffectOutbox.ts index 92e1e01ad43f2..e4d35592ca9c6 100644 --- a/apps/server/src/orchestration-v2/EffectOutbox.ts +++ b/apps/server/src/orchestration-v2/EffectOutbox.ts @@ -40,6 +40,9 @@ export const OrchestrationEffectRequestV2 = Schema.Union([ detail: Schema.optional(Schema.String), revokeMcpCredential: Schema.optional(Schema.Boolean), }), + Schema.Struct({ + type: Schema.Literal("provider-session.release-finished-delegate"), + }), Schema.Struct({ type: Schema.Literal("provider-turn.start"), runId: RunId, @@ -136,6 +139,7 @@ export const REPLAY_SAFE_EFFECT_TYPES_AFTER_PROCESS_LOSS = [ ] as const satisfies ReadonlyArray; export const PROCESS_BOUND_EFFECT_TYPES = [ + "provider-session.release-finished-delegate", "provider-turn.start", "provider-turn.interrupt", "provider-turn.steer", diff --git a/apps/server/src/orchestration-v2/EffectWorker.test.ts b/apps/server/src/orchestration-v2/EffectWorker.test.ts index 2329f2aa20cf0..fce17cf2968b0 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.test.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.test.ts @@ -111,6 +111,7 @@ function layerExecutorFor(input: { closeInstance: () => Effect.void, release: () => record("release"), detach: () => record("detach"), + releaseFinishedDelegate: () => Effect.void, otherWorkingThreads: () => Effect.succeed([]), releaseUnlessBusy: () => Effect.succeed(true), }), diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index c5b5ea9b383a9..cdd659fa9cf0f 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -138,6 +138,17 @@ export const layerExecutor: Layer.Layer< }), ), ); + case "provider-session.release-finished-delegate": + return providerSessions.releaseFinishedDelegate({ threadId: effect.threadId }).pipe( + Effect.mapError( + (cause) => + new OrchestrationEffectExecutionError({ + effectId: effect.id, + effectType: effect.request.type, + cause, + }), + ), + ); case "provider-turn.start": return providerTurnStart .start({ threadId: effect.threadId, runId: effect.request.runId, willRetry }) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index b4bdbd705c290..0f65fbba22d8a 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -12722,7 +12722,25 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio payload: resultTransfer, }, ]; - yield* writeSystemEvents(completionEvents, [], { updateDelegatedCompletionOnly: true }); + // A finished delegate is rarely resumed, so its provider process need not idle for the full + // timeout. The session manager releases it only if nothing can resume it by then. + const releaseCommandId = CommandId.make( + `command:system:release-finished-delegate:${childRun.id}`, + ); + yield* writeSystemEvents( + completionEvents, + terminalStatus === "completed" + ? [ + { + id: `effect:${releaseCommandId}:provider-session.release-finished-delegate`, + commandId: releaseCommandId, + threadId: childThreadId, + request: { type: "provider-session.release-finished-delegate" }, + } satisfies PendingOrchestrationEffectV2, + ] + : [], + { updateDelegatedCompletionOnly: true }, + ); if (completionPlan.offer && completionPlan.parentRun !== undefined) { yield* offerDelegatedCompletionDelivery(parentThreadId, completionPlan.parentRun.id); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index d8c046c018651..bff8236fc05bf 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -3950,3 +3950,177 @@ it.effect("ProviderSessionManagerV2 gives a delegated child its project's device assert.isTrue(child?.capabilities?.has("device")); }), ); + +const finishedDelegateRunEvent = (input: { + readonly idAllocator: IdAllocator.IdAllocatorV2["Service"]; + readonly threadId: ThreadId; + readonly ordinal: number; + readonly status: "completed" | "queued" | "running"; + readonly now: DateTime.Utc; +}) => + Effect.gen(function* () { + const runId = input.idAllocator.derive.run({ + threadId: input.threadId, + ordinal: input.ordinal, + }); + return { + id: yield* input.idAllocator.allocate.event({ threadId: input.threadId }), + type: "run.updated" as const, + threadId: input.threadId, + runId, + occurredAt: input.now, + payload: { + id: runId, + threadId: input.threadId, + ordinal: input.ordinal, + providerInstanceId: modelSelection.instanceId, + modelSelection, + providerThreadId: null, + userMessageId: yield* input.idAllocator.allocate.message({ + threadId: input.threadId, + ordinal: input.ordinal, + }), + rootNodeId: null, + activeAttemptId: null, + status: input.status, + requestedAt: input.now, + startedAt: input.status === "queued" ? null : input.now, + completedAt: input.status === "completed" ? input.now : null, + checkpointId: null, + contextHandoffId: null, + }, + }; + }); + +// Opens a one-process-per-thread session on a thread with the given runs, then asks to release +// it as a finished delegate with no clock advance, far inside the idle timeout. +const runFinishedDelegateScenario = (input: { + readonly delegated: boolean; + readonly runs: ReadonlyArray<"completed" | "queued" | "running">; + readonly hasPendingBackgroundWork?: boolean; +}) => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const effect = Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const projectId = yield* idAllocator.allocate.project({ + fixtureName: "provider-session-manager-finished-delegate", + }); + const threadId = yield* idAllocator.allocate.thread({ + fixtureName: "provider-session-manager-finished-delegate", + projectId, + }); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + yield* eventSink.write({ + events: [ + yield* makeThreadCreatedEvent({ + idAllocator, + threadId, + now, + projectId, + ...(input.delegated + ? { delegatedBy: ThreadId.make("thread-provider-session-manager-parent") } + : {}), + }), + ...(yield* Effect.forEach(input.runs, (status, index) => + finishedDelegateRunEvent({ + idAllocator, + threadId, + ordinal: index + 1, + status, + now, + }), + )), + ], + }); + yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy }); + + yield* manager.releaseFinishedDelegate({ threadId }); + + const released = (yield* Ref.get(state)).closeCount === 1; + const live = yield* manager.isLive(providerSessionId); + const sessionStatus = (yield* projectionStore.getThreadProjection(threadId)) + .providerSessions[0]?.status; + // A message to the child after the release opens a new session as usual. + const reopened = yield* manager.open({ + threadId, + providerSessionId, + modelSelection, + runtimePolicy, + }); + return { + released, + live, + sessionStatus, + openCount: (yield* Ref.get(state)).openCount, + reopenedLive: yield* manager.isLive(providerSessionId), + reopened, + }; + }); + return yield* effect.pipe( + Effect.provide( + layerTest({ + state, + idleTimeoutMs: 60 * 60_000, + capabilities: ExclusiveCapabilities, + ...(input.hasPendingBackgroundWork === undefined + ? {} + : { hasPendingBackgroundWork: Effect.succeed(input.hasPendingBackgroundWork) }), + }), + ), + ); + }); + +it.effect( + "ProviderSessionManagerV2 releases a finished delegated child's session at once and reopens it for a follow-up", + () => + Effect.gen(function* () { + const result = yield* runFinishedDelegateScenario({ delegated: true, runs: ["completed"] }); + assert.isTrue(result.released); + assert.isFalse(result.live); + assert.equal(result.sessionStatus, "stopped"); + assert.equal(result.openCount, 2); + assert.isTrue(result.reopenedLive); + }), +); + +it.effect.each<{ + readonly name: string; + readonly delegated: boolean; + readonly runs: ReadonlyArray<"completed" | "queued" | "running">; + readonly hasPendingBackgroundWork?: boolean; +}>([ + { name: "an ordinary thread", delegated: false, runs: ["completed"] }, + { name: "a delegated child whose run is still running", delegated: true, runs: ["running"] }, + { + name: "a delegated child with a queued follow-up", + delegated: true, + runs: ["completed", "queued"], + }, + { + name: "a delegated child with pending background work", + delegated: true, + runs: ["completed"], + hasPendingBackgroundWork: true, + }, +])("ProviderSessionManagerV2 keeps the session of $name", (scenario) => + Effect.gen(function* () { + const result = yield* runFinishedDelegateScenario({ + delegated: scenario.delegated, + runs: scenario.runs, + ...(scenario.hasPendingBackgroundWork === undefined + ? {} + : { hasPendingBackgroundWork: scenario.hasPendingBackgroundWork }), + }); + assert.isFalse(result.released); + assert.isTrue(result.live); + assert.equal(result.openCount, 1); + }), +); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index f06011676c667..5b934c13104f7 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -53,6 +53,7 @@ import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter" import { makeProviderFailure } from "@t3tools/provider-core/server/failure"; import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; +import { delegatedTaskProgress } from "@t3tools/provider-core/server/subagentProjection"; const DEFAULT_IDLE_TIMEOUT_MS = 30 * 60 * 1000; const DEFAULT_MAX_IDLE_PIN_MS = 4 * 60 * 60 * 1000; @@ -189,6 +190,15 @@ export interface ProviderSessionManagerV2Shape { readonly detail?: string; readonly revokeMcpCredential?: boolean; }) => Effect.Effect; + /** + * Releases the one-process-per-thread session of a delegated child whose result is out and + * which nothing can resume: no run, queued follow-up, subagent, background task or wake + * message. Keeps the session otherwise; a later message to the child opens a new one. + */ + readonly releaseFinishedDelegate: (input: { + readonly threadId: ThreadId; + readonly detail?: string; + }) => Effect.Effect; /** * Releases the session unless, at the moment of removal, a turn of a provider thread outside * `ownProviderThreadIds` is busy on it. Checked atomically with turn starts. Succeeds with @@ -2359,6 +2369,52 @@ export const layerWithOptions = ( ), ), release: releaseEntry, + releaseFinishedDelegate: Effect.fn("ProviderSessionManager.releaseFinishedDelegate")( + function* (input) { + // Shared processes unload per thread instead; only a sole attached thread owns its process. + const owned = [...(yield* Ref.get(sessions)).values()].filter( + (entry) => + !entry.supportsMultipleProviderThreads && + entry.attachedThreadIds.size === 1 && + entry.attachedThreadIds.has(input.threadId), + ); + if (owned.length === 0) return; + // Taken before the probes: any later activity bumps it and keeps the session. + const generations = new Map(owned.map((entry) => [entry, entry.idleGeneration])); + const finished = yield* projectionStore + .getThreadRecords( + input.threadId, + ["runs", "messages", "subagents", "providerThreads"], + { messageRoles: ["user"] }, + ) + .pipe( + Effect.map( + (records) => + records.thread.lineage.relationshipToParent === "subagent" && + delegatedTaskProgress(records).state === "result_available", + ), + // An unreadable thread may still be working; never kill it unseen. + Effect.orElseSucceed(() => false), + ); + if (!finished) return; + for (const entry of owned) { + if (entry.busyTurns.size > 0) continue; + const pending = + entry.runtime.hasPendingBackgroundWork === undefined + ? false + : yield* entry.runtime.hasPendingBackgroundWork.pipe( + Effect.catchCause(() => Effect.succeed(true)), + ); + if (pending) continue; + yield* releaseEntry({ + providerSessionId: entry.runtime.providerSessionId, + reason: "idle_timeout", + detail: input.detail ?? "Delegated task finished.", + onlyIfIdleGeneration: generations.get(entry)!, + }); + } + }, + ), releaseUnlessBusy: Effect.fn("ProviderSessionManager.releaseUnlessBusy")(function* (input) { const key = sessionKey(input.providerSessionId); const before = (yield* Ref.get(sessions)).get(key); diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index ad5341dd99e49..58cc51a89ac95 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -290,6 +290,7 @@ it.effect( closeInstance: () => Effect.void, release: () => Effect.void, detach: () => Effect.void, + releaseFinishedDelegate: () => Effect.void, otherWorkingThreads: () => Effect.succeed([]), releaseUnlessBusy: () => Effect.succeed(true), }), diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index 9285f72d46f56..8c99acff7edf8 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -657,6 +657,7 @@ it.effect.each([ Layer.provide( Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({ isLive: () => Effect.succeed(true), + releaseFinishedDelegate: () => Effect.void, otherWorkingThreads: () => Effect.succeed(others.map((id) => ThreadId.make(id))), releaseUnlessBusy: () => Effect.sync(() => teardowns.push("release")).pipe( From e6a670b9a58cfd250b8c57be53709ff03285bd59 Mon Sep 17 00:00:00 2001 From: Pyseph <45090858+Pyseph@users.noreply.github.com> Date: Thu, 5 Nov 2026 14:03:43 +0100 Subject: [PATCH 2/7] test(server): recovery cancels the new process-bound effect type too A pending provider-session.release-finished-delegate is cancelled after a restart, like the other process-bound effects, so the pinned list includes it. --- .../src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts index 28f1df58a6b1f..7414e2ec27cc9 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts @@ -770,6 +770,7 @@ it.effect( assert.equal(summary.stoppedSessions, 1); assert.equal(summary.retiredEffects, 2); assert.deepEqual(committedInput?.cancelUnsettledEffects?.effectTypes, [ + "provider-session.release-finished-delegate", "provider-turn.start", "provider-turn.interrupt", "provider-turn.steer", From 7a53dc214d266876e1531b716c1aeac670f93773 Mon Sep 17 00:00:00 2001 From: Pyseph <45090858+Pyseph@users.noreply.github.com> Date: Thu, 5 Nov 2026 14:10:18 +0100 Subject: [PATCH 3/7] fix(server): a failed delegate release no longer holds up the child's effects The release runs as an outbox effect on the child thread, and a pending effect blocks later effects on that thread, so a failed release made a follow-up wait for the retry of an optional cleanup. The retry can't help either: the live entry is already gone and retryReleaseRecords owns the record. Log and swallow the failure, like the other release paths. --- apps/server/src/orchestration-v2/ProviderSessionManager.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 5b934c13104f7..f54c139240f34 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -2411,7 +2411,7 @@ export const layerWithOptions = ( reason: "idle_timeout", detail: input.detail ?? "Delegated task finished.", onlyIfIdleGeneration: generations.get(entry)!, - }); + }).pipe(logReleaseFailure(entry.runtime.providerSessionId)); } }, ), From 93af51440032fb96167c9a1d1b4904544df203a6 Mon Sep 17 00:00:00 2001 From: Pyseph <45090858+Pyseph@users.noreply.github.com> Date: Thu, 5 Nov 2026 14:24:11 +0100 Subject: [PATCH 4/7] fix(server): a held follow-up keeps a finished child's session delegatedTaskProgress ignores held queued runs, which is right for its own purpose, but a held follow-up still needs this session when its queue resumes. --- .../ProviderSessionManager.test.ts | 16 +++++++++++----- .../orchestration-v2/ProviderSessionManager.ts | 2 ++ 2 files changed, 13 insertions(+), 5 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index cff139612f891..17232f32078a5 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -3954,7 +3954,7 @@ const finishedDelegateRunEvent = (input: { readonly idAllocator: IdAllocator.IdAllocatorV2["Service"]; readonly threadId: ThreadId; readonly ordinal: number; - readonly status: "completed" | "queued" | "running"; + readonly status: "completed" | "queued" | "held" | "running"; readonly now: DateTime.Utc; }) => Effect.gen(function* () { @@ -3981,9 +3981,10 @@ const finishedDelegateRunEvent = (input: { }), rootNodeId: null, activeAttemptId: null, - status: input.status, + status: input.status === "held" ? ("queued" as const) : input.status, + ...(input.status === "held" ? { queueHeld: true } : {}), requestedAt: input.now, - startedAt: input.status === "queued" ? null : input.now, + startedAt: input.status === "queued" || input.status === "held" ? null : input.now, completedAt: input.status === "completed" ? input.now : null, checkpointId: null, contextHandoffId: null, @@ -3995,7 +3996,7 @@ const finishedDelegateRunEvent = (input: { // it as a finished delegate with no clock advance, far inside the idle timeout. const runFinishedDelegateScenario = (input: { readonly delegated: boolean; - readonly runs: ReadonlyArray<"completed" | "queued" | "running">; + readonly runs: ReadonlyArray<"completed" | "queued" | "held" | "running">; readonly hasPendingBackgroundWork?: boolean; }) => Effect.gen(function* () { @@ -4093,7 +4094,7 @@ it.effect( it.effect.each<{ readonly name: string; readonly delegated: boolean; - readonly runs: ReadonlyArray<"completed" | "queued" | "running">; + readonly runs: ReadonlyArray<"completed" | "queued" | "held" | "running">; readonly hasPendingBackgroundWork?: boolean; }>([ { name: "an ordinary thread", delegated: false, runs: ["completed"] }, @@ -4103,6 +4104,11 @@ it.effect.each<{ delegated: true, runs: ["completed", "queued"], }, + { + name: "a delegated child with a held follow-up", + delegated: true, + runs: ["completed", "held"], + }, { name: "a delegated child with pending background work", delegated: true, diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index f54c139240f34..5fb44244175da 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -2391,6 +2391,8 @@ export const layerWithOptions = ( Effect.map( (records) => records.thread.lineage.relationshipToParent === "subagent" && + // Progress ignores held runs, but a held follow-up still needs this session. + !records.runs.some((run) => run.status === "queued") && delegatedTaskProgress(records).state === "result_available", ), // An unreadable thread may still be working; never kill it unseen. From 407ddf0465b2742b657b153427762efcbd4bb660 Mon Sep 17 00:00:00 2001 From: Pyseph <45090858+Pyseph@users.noreply.github.com> Date: Thu, 5 Nov 2026 14:34:27 +0100 Subject: [PATCH 5/7] fix(server): a finished child's release leaves a replacement session alone The checks yield, and a session opened meanwhile takes the same id and restarts the idle generation, so the generation alone can match a session the checks never looked at. Match the runtime as well. --- .../ProviderSessionManager.test.ts | 75 +++++++++++++++++++ .../ProviderSessionManager.ts | 7 ++ 2 files changed, 82 insertions(+) diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 17232f32078a5..998816506d723 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -4129,3 +4129,78 @@ it.effect.each<{ assert.equal(result.openCount, 1); }), ); + +it.effect( + "ProviderSessionManagerV2 a finished child's release leaves a replacement session alone", + () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const probing = yield* Deferred.make(); + const replaced = yield* Deferred.make(); + const effect = Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const now = yield* DateTime.now; + const projectId = yield* idAllocator.allocate.project({ + fixtureName: "provider-session-manager-delegate-replacement", + }); + const threadId = yield* idAllocator.allocate.thread({ + fixtureName: "provider-session-manager-delegate-replacement", + projectId, + }); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + yield* eventSink.write({ + events: [ + yield* makeThreadCreatedEvent({ + idAllocator, + threadId, + now, + projectId, + delegatedBy: ThreadId.make("thread-provider-session-manager-parent"), + }), + yield* finishedDelegateRunEvent({ + idAllocator, + threadId, + ordinal: 1, + status: "completed", + now, + }), + ], + }); + yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy }); + + const release = yield* manager + .releaseFinishedDelegate({ threadId }) + .pipe(Effect.forkScoped); + yield* Deferred.await(probing); + // The replacement takes the same id and reaches the same idle generation. + yield* manager.close(providerSessionId); + yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy }); + yield* Deferred.succeed(replaced, undefined); + yield* Fiber.join(release); + + assert.isTrue(yield* manager.isLive(providerSessionId)); + const { openCount, closeCount } = yield* Ref.get(state); + assert.equal(openCount, 2); + assert.equal(closeCount, 1); + }); + + yield* effect.pipe( + Effect.provide( + layerTest({ + state, + idleTimeoutMs: 60 * 60_000, + capabilities: ExclusiveCapabilities, + hasPendingBackgroundWork: Deferred.succeed(probing, undefined).pipe( + Effect.andThen(Deferred.await(replaced)), + Effect.as(false), + ), + }), + ), + ); + }), +); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 5fb44244175da..479a3f923c8d7 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -932,6 +932,7 @@ export const layerWithOptions = ( function* (input: { readonly providerSessionId: ProviderSessionId; readonly onlyIfIdleGeneration?: number; + readonly onlyIfRuntime?: LiveSessionEntry["runtime"]; readonly keepIfBusyOtherThan?: ReadonlyArray; }): Effect.fn.Return, DateTime.Utc]> { const key = sessionKey(input.providerSessionId); @@ -945,6 +946,9 @@ export const layerWithOptions = ( if (existing !== candidate) { return [existing === undefined ? "gone" : "changed", current] as const; } + if (input.onlyIfRuntime !== undefined && existing.runtime !== input.onlyIfRuntime) { + return ["kept", current] as const; + } if ( input.onlyIfIdleGeneration !== undefined && (existing.busyTurns.size > 0 || @@ -1019,6 +1023,7 @@ export const layerWithOptions = ( readonly detachedThreadId?: ThreadId; readonly cancelIdleFiber?: boolean; readonly onlyIfIdleGeneration?: number; + readonly onlyIfRuntime?: LiveSessionEntry["runtime"]; readonly keepIfBusyOtherThan?: ReadonlyArray; readonly gracefulSubscribers?: boolean; }) => @@ -2413,6 +2418,8 @@ export const layerWithOptions = ( reason: "idle_timeout", detail: input.detail ?? "Delegated task finished.", onlyIfIdleGeneration: generations.get(entry)!, + // A replacement reuses the session id and restarts the generation count. + onlyIfRuntime: entry.runtime, }).pipe(logReleaseFailure(entry.runtime.providerSessionId)); } }, From 162c3659a51dbcda8207d2c7376e890dda8a4cc0 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 03:38:51 +0000 Subject: [PATCH 6/7] test(server): drop redundant finished-delegate assertions AI-assisted; verified with the tests and tools listed above. --- .../orchestration-v2/DelegatedCompletionDelivery.test.ts | 6 +----- .../src/orchestration-v2/ProviderSessionManager.test.ts | 3 +-- 2 files changed, 2 insertions(+), 7 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 747df1e5d329f..289837db90ea1 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -3202,7 +3202,7 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => { now, }); const completed = yield* child("release-delegate-completed", "completed"); - const cancelled = yield* child("release-delegate-cancelled", "cancelled"); + yield* child("release-delegate-cancelled", "cancelled"); yield* orchestrator.recoverDelegatedTasks; @@ -3215,10 +3215,6 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => { queued.map((row) => row.thread_id), [completed.childThreadId], ); - assert.notInclude( - queued.map((row) => row.thread_id), - cancelled.childThreadId, - ); }), ); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index f23b38d60887c..974442525cf23 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -4634,7 +4634,7 @@ const runFinishedDelegateScenario = (input: { const sessionStatus = (yield* projectionStore.getThreadProjection(threadId)) .providerSessions[0]?.status; // A message to the child after the release opens a new session as usual. - const reopened = yield* manager.open({ + yield* manager.open({ threadId, providerSessionId, modelSelection, @@ -4646,7 +4646,6 @@ const runFinishedDelegateScenario = (input: { sessionStatus, openCount: (yield* Ref.get(state)).openCount, reopenedLive: yield* manager.isLive(providerSessionId), - reopened, }; }); return yield* effect.pipe( From 1fda925c57a5fce7746683aa8b96058ead22d4c9 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 04:03:48 +0000 Subject: [PATCH 7/7] fix(server): keep non-resumable delegate sessions and drop the persisted release effect A runtime that cannot reload its native conversation keeps its session for the idle window, as the linked-environment release already does. The release now runs from the committed completion event instead of a new outbox effect type, which an older server could claim but not decode, leaving the child's later effects blocked. AI-assisted; verified with the tests and tools listed above. --- .../DelegatedCompletionDelivery.test.ts | 52 +------------------ .../src/orchestration-v2/EffectOutbox.ts | 4 -- .../src/orchestration-v2/EffectWorker.ts | 11 ---- .../src/orchestration-v2/Orchestrator.ts | 46 +++++++++------- .../ProviderRuntimeRecoveryService.test.ts | 1 - .../ProviderSessionManager.test.ts | 27 +++++++--- .../ProviderSessionManager.ts | 4 +- .../ThreadManagementService.test.ts | 1 - 8 files changed, 51 insertions(+), 95 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 289837db90ea1..a5c123157bae1 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -25,7 +25,6 @@ import * as Effect from "effect/Effect"; import * as Deferred from "effect/Deferred"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; -import * as SqlClient from "effect/sql/SqlClient"; import * as Stream from "effect/Stream"; import * as Tracer from "effect/Tracer"; import * as TestClock from "effect/testing/TestClock"; @@ -149,7 +148,7 @@ const layerTest = Layer.mergeAll( ), Layer.provide(McpSessionRegistryTestkit.layer), Layer.provide(McpProviderSessions.layer), - Layer.provideMerge(SqlitePersistence.layerMemory), + Layer.provide(SqlitePersistence.layerMemory), Layer.provide(layerCheckpointStoreTest), Layer.provide(layerServerConfig), Layer.provide(ServerSettings.layerTest()), @@ -3169,55 +3168,6 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => { }), ); - it.effect( - "queues the release of a completed child's provider session, not a cancelled one's", - () => - Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const sql = yield* SqlClient.SqlClient; - const now = yield* DateTime.now; - const threadId = ThreadId.make("thread:release-delegate-parent"); - const projectId = ProjectId.make("project:release-delegate"); - const runId = RunId.make("run:release-delegate-parent"); - const rootNodeId = NodeId.make("node:release-delegate-parent"); - yield* seedParentWithTerminalTask({ - threadId, - projectId, - runId, - rootNodeId, - taskId: NodeId.make("node:release-delegate-seed"), - deliveryState: "acknowledged", - now, - }); - const child = (name: string, runStatus: "cancelled" | "completed") => - seedRestartCancelledChild({ - parentThreadId: threadId, - projectId, - parentRunId: runId, - rootNodeId, - name, - completionWake: "always", - continuationPending: false, - runStatus, - now, - }); - const completed = yield* child("release-delegate-completed", "completed"); - yield* child("release-delegate-cancelled", "cancelled"); - - yield* orchestrator.recoverDelegatedTasks; - - const queued = yield* sql<{ readonly thread_id: string }>` - SELECT thread_id FROM orchestration_v2_effect_outbox - WHERE effect_type = 'provider-session.release-finished-delegate' - AND thread_id LIKE 'thread:release-delegate-%' - `; - assert.deepEqual( - queued.map((row) => row.thread_id), - [completed.childThreadId], - ); - }), - ); - it.effect("holds a restart-cancelled child for its continuation's result", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/apps/server/src/orchestration-v2/EffectOutbox.ts b/apps/server/src/orchestration-v2/EffectOutbox.ts index e4d35592ca9c6..92e1e01ad43f2 100644 --- a/apps/server/src/orchestration-v2/EffectOutbox.ts +++ b/apps/server/src/orchestration-v2/EffectOutbox.ts @@ -40,9 +40,6 @@ export const OrchestrationEffectRequestV2 = Schema.Union([ detail: Schema.optional(Schema.String), revokeMcpCredential: Schema.optional(Schema.Boolean), }), - Schema.Struct({ - type: Schema.Literal("provider-session.release-finished-delegate"), - }), Schema.Struct({ type: Schema.Literal("provider-turn.start"), runId: RunId, @@ -139,7 +136,6 @@ export const REPLAY_SAFE_EFFECT_TYPES_AFTER_PROCESS_LOSS = [ ] as const satisfies ReadonlyArray; export const PROCESS_BOUND_EFFECT_TYPES = [ - "provider-session.release-finished-delegate", "provider-turn.start", "provider-turn.interrupt", "provider-turn.steer", diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index cdd659fa9cf0f..c5b5ea9b383a9 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -138,17 +138,6 @@ export const layerExecutor: Layer.Layer< }), ), ); - case "provider-session.release-finished-delegate": - return providerSessions.releaseFinishedDelegate({ threadId: effect.threadId }).pipe( - Effect.mapError( - (cause) => - new OrchestrationEffectExecutionError({ - effectId: effect.id, - effectType: effect.request.type, - cause, - }), - ), - ); case "provider-turn.start": return providerTurnStart .start({ threadId: effect.threadId, runId: effect.request.runId, willRetry }) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index f2a7e1a1536b4..b2ac5aae01284 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -13166,25 +13166,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio payload: resultTransfer, }, ]; - // A finished delegate is rarely resumed, so its provider process need not idle for the full - // timeout. The session manager releases it only if nothing can resume it by then. - const releaseCommandId = CommandId.make( - `command:system:release-finished-delegate:${childRun.id}`, - ); - yield* writeSystemEvents( - completionEvents, - terminalStatus === "completed" - ? [ - { - id: `effect:${releaseCommandId}:provider-session.release-finished-delegate`, - commandId: releaseCommandId, - threadId: childThreadId, - request: { type: "provider-session.release-finished-delegate" }, - } satisfies PendingOrchestrationEffectV2, - ] - : [], - { updateDelegatedCompletionOnly: true }, - ); + yield* writeSystemEvents(completionEvents, [], { updateDelegatedCompletionOnly: true }); if (completionPlan.offer && completionPlan.parentRun !== undefined) { yield* offerDelegatedCompletionDelivery(parentThreadId, completionPlan.parentRun.id); @@ -14316,6 +14298,32 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio Effect.forkScoped, ); + // A finished delegate is rarely resumed, so its process need not idle for the full timeout. The + // session manager re-checks that nothing can resume it. Not an outbox effect: older servers + // cannot decode new effect types, and a lost process holds no session to release anyway. + yield* eventSink + .stream({ afterSequence: terminalEventsAfterSequence, eventType: "subagent.updated" }) + .pipe( + Stream.filterMap((stored) => + stored.event.type === "subagent.updated" && + stored.event.payload.status === "completed" && + stored.event.payload.childThreadId !== null + ? Result.succeed(stored.event.payload.childThreadId) + : Result.failVoid, + ), + Stream.runForEach((threadId) => + providerSessions.releaseFinishedDelegate({ threadId }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Failed to release a finished delegate's session", { + threadId, + cause, + }), + ), + ), + ), + Effect.forkScoped, + ); + // A child whose run ends while its background tasks or its own subagents still run has no // result yet. When that work settles there is no run update to finalize it, so these events // do it. A background roster counts once it stops holding after it was seen holding. diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts index a83ca4fd1370e..642a642a80df1 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts @@ -771,7 +771,6 @@ it.effect( assert.equal(summary.stoppedSessions, 1); assert.equal(summary.retiredEffects, 2); assert.deepEqual(committedInput?.cancelUnsettledEffects?.effectTypes, [ - "provider-session.release-finished-delegate", "provider-turn.start", "provider-turn.interrupt", "provider-turn.steer", diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 974442525cf23..931f79a68c3ba 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -434,6 +434,7 @@ const makeProviderAdapter = Effect.fnUntraced(function* ( }) => Effect.Effect; readonly hasPendingBackgroundWork?: Effect.Effect; readonly hasPendingBackgroundWorkForThread?: Effect.Effect; + readonly canResumeAfterRestart?: boolean; readonly backgroundTasksEndWithProcess?: boolean; readonly hangSessionScopeClose?: boolean; readonly startTurn?: Effect.Effect; @@ -533,6 +534,9 @@ const makeProviderAdapter = Effect.fnUntraced(function* ( ...(options.hasPendingBackgroundWork === undefined ? {} : { hasPendingBackgroundWork: options.hasPendingBackgroundWork }), + ...(options.canResumeAfterRestart === undefined + ? {} + : { canResumeAfterRestart: options.canResumeAfterRestart }), ...(options.hasPendingBackgroundWorkForThread === undefined ? {} : { @@ -596,6 +600,7 @@ function layerTest(input: { }; readonly hasPendingBackgroundWork?: Effect.Effect; readonly hasPendingBackgroundWorkForThread?: Effect.Effect; + readonly canResumeAfterRestart?: boolean; readonly backgroundTasksEndWithProcess?: boolean; readonly hangSessionScopeClose?: boolean; readonly startTurn?: Effect.Effect; @@ -628,6 +633,9 @@ function layerTest(input: { ...(input.hasPendingBackgroundWork === undefined ? {} : { hasPendingBackgroundWork: input.hasPendingBackgroundWork }), + ...(input.canResumeAfterRestart === undefined + ? {} + : { canResumeAfterRestart: input.canResumeAfterRestart }), ...(input.hasPendingBackgroundWorkForThread === undefined ? {} : { hasPendingBackgroundWorkForThread: input.hasPendingBackgroundWorkForThread }), @@ -4583,6 +4591,7 @@ const runFinishedDelegateScenario = (input: { readonly delegated: boolean; readonly runs: ReadonlyArray<"completed" | "queued" | "held" | "running">; readonly hasPendingBackgroundWork?: boolean; + readonly canResumeAfterRestart?: boolean; }) => Effect.gen(function* () { const state = yield* Ref.make(emptyState); @@ -4657,6 +4666,9 @@ const runFinishedDelegateScenario = (input: { ...(input.hasPendingBackgroundWork === undefined ? {} : { hasPendingBackgroundWork: Effect.succeed(input.hasPendingBackgroundWork) }), + ...(input.canResumeAfterRestart === undefined + ? {} + : { canResumeAfterRestart: input.canResumeAfterRestart }), }), ), ); @@ -4680,6 +4692,7 @@ it.effect.each<{ readonly delegated: boolean; readonly runs: ReadonlyArray<"completed" | "queued" | "held" | "running">; readonly hasPendingBackgroundWork?: boolean; + readonly canResumeAfterRestart?: boolean; }>([ { name: "an ordinary thread", delegated: false, runs: ["completed"] }, { name: "a delegated child whose run is still running", delegated: true, runs: ["running"] }, @@ -4699,15 +4712,15 @@ it.effect.each<{ runs: ["completed"], hasPendingBackgroundWork: true, }, + { + name: "a delegated child whose runtime cannot resume its conversation", + delegated: true, + runs: ["completed"], + canResumeAfterRestart: false, + }, ])("ProviderSessionManagerV2 keeps the session of $name", (scenario) => Effect.gen(function* () { - const result = yield* runFinishedDelegateScenario({ - delegated: scenario.delegated, - runs: scenario.runs, - ...(scenario.hasPendingBackgroundWork === undefined - ? {} - : { hasPendingBackgroundWork: scenario.hasPendingBackgroundWork }), - }); + const result = yield* runFinishedDelegateScenario(scenario); assert.isFalse(result.released); assert.isTrue(result.live); assert.equal(result.openCount, 1); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index e06bf5320476c..870ec95aab1e1 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -2685,7 +2685,9 @@ export const layerWithOptions = ( ); if (!finished) return; for (const entry of owned) { - if (entry.busyTurns.size > 0) continue; + // A runtime that cannot reload its native conversation keeps it for the idle window. + if (entry.busyTurns.size > 0 || entry.runtime.canResumeAfterRestart === false) + continue; const pending = entry.runtime.hasPendingBackgroundWork === undefined ? false diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index a7e197f5b12cb..742616f27bd3e 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -761,7 +761,6 @@ it.effect.each([ Layer.provide( Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({ isLive: () => Effect.succeed(true), - releaseFinishedDelegate: () => Effect.void, otherWorkingThreads: () => Effect.succeed(others.map((id) => ThreadId.make(id))), releaseUnlessBusy: () => Effect.sync(() => teardowns.push("release")).pipe(