diff --git a/apps/server/src/mcp/DelegatedTaskProviderExit.integration.test.ts b/apps/server/src/mcp/DelegatedTaskProviderExit.integration.test.ts new file mode 100644 index 000000000000..2c6b2257781e --- /dev/null +++ b/apps/server/src/mcp/DelegatedTaskProviderExit.integration.test.ts @@ -0,0 +1,398 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import { + CommandId, + EnvironmentId, + MessageId, + NodeId, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + ProviderSessionId, + ProviderThreadId, + ProviderTurnId, + ThreadId, + TurnItemId, + type OrchestrationV2DomainEvent, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Queue from "effect/Queue"; +import * as Stream from "effect/Stream"; + +import { ClaudeProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/ClaudeAdapterV2.ts"; +import * as CommandReceiptStore from "../orchestration-v2/CommandReceiptStore.ts"; +import * as EffectWorker from "../orchestration-v2/EffectWorker.ts"; +import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; +import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts"; +import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts"; +import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; +import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; +import * as ServerSettings from "../serverSettings.ts"; +import * as ProviderReplayHarness from "../orchestration-v2/testkit/ProviderReplayHarness.ts"; +import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace"; +import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import * as ProjectService from "../project/ProjectService.ts"; +import * as ProviderRegistry from "../provider/ProviderRegistry.ts"; +import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts"; +import * as SecretRequests from "../secrets/SecretRequests.ts"; +import type { McpInvocationScope } from "./McpInvocationContext.ts"; +import * as OrchestratorMcpService from "./OrchestratorMcpService.ts"; + +const layerOrchestratorService = OrchestratorMcpService.layer.pipe( + Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})), + Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})), +); + +const driver = ProviderDriverKind.make("claudeAgent"); +const instanceId = ProviderInstanceId.make("claudeAgent"); +const modelSelection = { instanceId, model: "claude-test" }; + +// A delegated child ends its turn while a native subagent it started keeps +// running, then its provider process dies. Nothing will ever report that +// subagent again, so it must fail and the delegated task must settle. + +it.effect("a delegated child's provider exit settles its task after the root turn completed", () => + Effect.scoped( + Effect.gen(function* () { + const cwd = yield* checkpointWorkspace("provider-exit-native-subagent"); + const turns: Array<{ + readonly turn: ProviderAdapter.ProviderAdapterV2TurnInput; + readonly events: Queue.Queue< + ProviderAdapter.ProviderAdapterV2Event, + ProviderAdapter.ProviderAdapterEventStreamError + >; + }> = []; + const adapter: ProviderAdapter.ProviderAdapterV2["Service"] = { + instanceId, + driver, + getCapabilities: () => Effect.succeed(ClaudeProviderCapabilitiesV2), + planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), + openSession: (input) => + Effect.gen(function* () { + const now = yield* DateTime.now; + const events = yield* Queue.unbounded< + ProviderAdapter.ProviderAdapterV2Event, + ProviderAdapter.ProviderAdapterEventStreamError + >(); + return { + instanceId, + driver, + providerSessionId: input.providerSessionId, + providerSession: { + id: input.providerSessionId, + driver, + providerInstanceId: instanceId, + status: "ready", + cwd, + model: modelSelection.model, + capabilities: ClaudeProviderCapabilitiesV2, + createdAt: now, + updatedAt: now, + lastError: null, + }, + events: Stream.fromQueue(events), + ensureThread: ({ threadId }) => + Effect.succeed({ + id: ProviderThreadId.make(`provider-thread:${threadId}`), + driver, + providerInstanceId: instanceId, + providerSessionId: input.providerSessionId, + appThreadId: threadId, + ownerNodeId: null, + nativeThreadRef: { driver, nativeId: `native:${threadId}`, strength: "strong" }, + nativeConversationHeadRef: null, + status: "idle", + firstRunOrdinal: null, + lastRunOrdinal: null, + handoffIds: [], + forkedFrom: null, + createdAt: now, + updatedAt: now, + }), + resumeThread: ({ providerThread }) => Effect.succeed(providerThread), + startTurn: (turn) => + Effect.gen(function* () { + turns.push({ turn, events }); + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { + id: ProviderTurnId.make(`provider-turn:${turn.attemptId}`), + providerThreadId: turn.providerThread.id, + nodeId: turn.rootNodeId, + runAttemptId: turn.attemptId, + nativeTurnRef: null, + ordinal: turn.providerTurnOrdinal, + status: "running", + startedAt: now, + completedAt: null, + }, + }); + }), + steerTurn: () => Effect.die("unused"), + interruptTurn: () => Effect.void, + respondToRuntimeRequest: () => Effect.die("unused"), + readThreadSnapshot: () => Effect.die("unused"), + rollbackThread: () => Effect.die("unused"), + forkThread: () => Effect.die("unused"), + }; + }), + }; + const layerAdapters = ProviderAdapterRegistry.layerSingle(adapter); + const layerOrchestrator = ProviderReplayHarness.layerWithRegistry( + { name: "provider-exit-native-subagent" }, + layerAdapters, + { runEffectWorker: false }, + ); + const layerOrchestration = Layer.merge( + layerOrchestrator, + ThreadManagementService.layer.pipe( + Layer.provide(layerOrchestrator), + Layer.provide(ServerSettings.layerTest()), + Layer.provide( + CommandReceiptStore.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)), + ), + ), + ); + const layerTest = layerOrchestratorService.pipe( + Layer.provideMerge(layerOrchestration), + Layer.provide( + Layer.mergeAll( + layerAdapters, + Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }), + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + ServerSettings.layerTest(), + Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), + NodeServices.layer, + ), + ), + ); + + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + const parentThreadId = ThreadId.make("thread:provider-exit-parent"); + const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) => + orchestrator.streamDomainEvents.pipe( + Stream.filter(predicate), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + const turnRunning = () => + watch( + (event) => event.type === "provider-turn.updated" && event.payload.status === "running", + ); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create-parent"), + threadId: parentThreadId, + projectId: ProjectId.make("project:provider-exit"), + title: "Parent", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + createdBy: "user", + creationSource: "web", + }); + const parentRunning = yield* turnRunning(); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("start-parent"), + threadId: parentThreadId, + messageId: MessageId.make("message:start-parent"), + text: "Delegate a review", + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + yield* Fiber.join(parentRunning); + + const parentRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs[0]!; + const childRunning = yield* turnRunning(); + yield* orchestrator.dispatch({ + type: "delegated_task.request", + commandId: CommandId.make("delegate"), + parentThreadId, + parentRunId: parentRun.id, + parentNodeId: parentRun.rootNodeId!, + task: "Review the change", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + completionWake: "always", + createdBy: "agent", + creationSource: "mcp", + }); + yield* worker.drain(); + yield* Fiber.join(childRunning); + const task = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]!; + const child = turns.find((entry) => entry.turn.threadId === task.childThreadId)!; + const childTurn = (yield* orchestrator.getThreadProjection(task.childThreadId!)) + .providerTurns[0]!; + + const now = yield* DateTime.now; + const subagentId = NodeId.make("node:native-subagent"); + const native = { + threadId: task.childThreadId!, + runId: child.turn.runId, + origin: "provider_native", + driver, + providerInstanceId: instanceId, + providerThreadId: childTurn.providerThreadId, + childThreadId: null, + prompt: "Check the tests", + title: null, + status: "running", + result: null, + startedAt: now, + completedAt: null, + updatedAt: now, + } as const; + const waiting = yield* watch( + (event) => + event.type === "run.updated" && + event.payload.id === child.turn.runId && + event.payload.status === "waiting", + ); + for (const event of [ + { + type: "node.updated", + driver, + node: { + id: subagentId, + threadId: native.threadId, + runId: native.runId, + parentNodeId: child.turn.rootNodeId, + rootNodeId: child.turn.rootNodeId, + kind: "subagent", + status: "running", + countsForRun: false, + providerThreadId: childTurn.providerThreadId, + providerTurnId: childTurn.id, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: now, + completedAt: null, + }, + }, + { + type: "subagent.updated", + driver, + subagent: { + ...native, + id: subagentId, + parentNodeId: child.turn.rootNodeId, + createdBy: "agent", + nativeTaskRef: null, + model: null, + }, + }, + { + type: "turn_item.updated", + driver, + turnItem: { + ...native, + id: TurnItemId.make("turn-item:native-subagent"), + nodeId: subagentId, + providerTurnId: childTurn.id, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + type: "subagent", + subagentId, + }, + }, + { + type: "provider_turn.updated", + driver, + providerTurn: { ...childTurn, status: "completed", completedAt: now }, + }, + { + type: "turn.terminal", + driver, + providerThreadId: childTurn.providerThreadId, + providerTurnId: childTurn.id, + runOrdinal: child.turn.runOrdinal, + status: "completed", + failure: null, + threadDisposition: "reusable", + }, + ] satisfies ReadonlyArray) { + yield* Queue.offer(child.events, event); + } + yield* Fiber.join(waiting); + yield* worker.drain(); + + const scope: McpInvocationScope = { + environmentId: EnvironmentId.make("environment:provider-exit"), + requestNamespace: "provider-session:provider-exit", + thread: { + threadId: parentThreadId, + providerSessionId: "provider-session:provider-exit", + providerInstanceId: instanceId, + }, + client: undefined, + capabilities: new Set(["orchestration"]), + issuedAt: 1, + }; + // The checkpoint above completed the child's run; only its native + // subagent still holds the task open. + const before = yield* service.taskStatus(scope, task.id); + assert.equal(before.workState, "waiting_for_children"); + + // The parent's own task record is what wakes it; task_status alone + // derives "completed" live from the child. + const parentTaskSettled = yield* watch( + (event) => + event.type === "subagent.updated" && + event.threadId === parentThreadId && + event.payload.id === task.id && + event.payload.result !== null, + ); + const subagentFailed = yield* watch( + (event) => + event.type === "subagent.updated" && + event.payload.id === subagentId && + event.payload.status === "failed", + ); + yield* Queue.fail( + child.events, + new ProviderAdapter.ProviderAdapterEventStreamError({ + driver, + providerSessionId: ProviderSessionId.make("provider-session:provider-exit"), + cause: "provider process exited", + }), + ); + yield* Fiber.join(subagentFailed); + yield* Fiber.join(parentTaskSettled); + yield* worker.drain(); + + const childAfter = yield* orchestrator.getThreadProjection(task.childThreadId!); + assert.equal(childAfter.runs[0]?.status, "completed"); + assert.equal( + childAfter.turnItems.find((item) => item.type === "subagent")?.status, + "failed", + ); + const after = yield* service.taskStatus(scope, task.id); + assert.equal(after.status, "completed"); + assert.equal(after.workState, "result_available"); + const parentTask = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( + (entry) => entry.id === task.id, + ); + assert.equal(parentTask?.status, "completed"); + }).pipe(Effect.provide(layerTest)); + }), + ), +); diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 928ae4a7e13a..fd5f3be9bbbd 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1791,7 +1791,32 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( model: input.modelSelection.model, now, }); - const events = yield* Queue.unbounded(); + const events = yield* Queue.unbounded< + ProviderAdapter.ProviderAdapterV2Event, + ProviderAdapter.ProviderAdapterEventStreamError + >(); + // Every thread in this session shares one app-server. Once it is gone, + // fail the session's stream: the session manager then fails its runs + // and releases the session, so the next turn starts a new app-server. + yield* client.awaitTermination.pipe( + Effect.tap((cause) => + Effect.logWarning("Codex app-server exited; failing its provider session.", { + providerSessionId: input.providerSessionId, + cause, + }), + ), + Effect.flatMap((cause) => + Queue.fail( + events, + new ProviderAdapter.ProviderAdapterEventStreamError({ + driver: CODEX_PROVIDER, + providerSessionId: input.providerSessionId, + cause, + }), + ), + ), + Effect.forkIn(scope), + ); const rateLimitSnapshot = yield* Ref.make(undefined); const limitedTurnItems = yield* Ref.make( new Map>(), @@ -6413,7 +6438,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( driver: CODEX_PROVIDER, providerSessionId: input.providerSessionId, providerSession: session, - events: Stream.fromEffectRepeat(Queue.take(events)), + events: Stream.fromQueue(events), canReuseContextUsage: canReuseCodexContextUsage, // Known gap: a subagent that Codex resumes later reads as completed // (not pending) between turns, so idle release can win the race diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 4464d09916e3..b3b45f540419 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -11499,27 +11499,62 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio // below. Replaying the full event table on every server start delays live // queue promotion in proportion to the lifetime size of the database. const terminalEventsAfterSequence = yield* eventSink.latestSequence().pipe(Effect.orDie); + // See PROVIDER_EXIT_SUBAGENT_CASCADE_COMMAND_PREFIX: those subagent rows settle + // without a run update, so the child's delegated task is reconsidered here. + const settleDelegatedTaskAfterProviderExit = (stored: OrchestrationV2StoredEvent) => + Effect.gen(function* () { + const threadId = stored.event.threadId; + const parentThreadId = yield* appOwnedSubagentParentThreadId(threadId); + if (parentThreadId === undefined) return; + yield* threadDispatch.withLock(parentThreadId, finalizeAppOwnedSubagent(threadId)); + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Failed to settle a delegated task after its provider exited", { + threadId: stored.event.threadId, + cause, + }), + ), + ); + // Queue promotion can wait on a provider or a thread lock. Subscribe to run // updates before buffering so that wait never retains unrelated tool bodies. - yield* eventSink - .stream({ afterSequence: terminalEventsAfterSequence, eventType: "run.updated" }) - .pipe( - Stream.filter( - (stored) => - stored.event.type === "run.updated" && - !String(stored.commandId).startsWith("command:runtime-reconcile:") && - ((stored.event.payload.delegatedTaskId !== undefined && - (stored.event.payload.status === "queued" || - stored.event.payload.status === "starting")) || - stored.event.payload.status === "completed" || - stored.event.payload.status === "interrupted" || - stored.event.payload.status === "failed" || - stored.event.payload.status === "cancelled" || - stored.event.payload.status === "rolled_back"), + // One consumer serializes both reactions. + yield* Stream.merge( + eventSink + .stream({ afterSequence: terminalEventsAfterSequence, eventType: "run.updated" }) + .pipe( + Stream.filter( + (stored) => + stored.event.type === "run.updated" && + !String(stored.commandId).startsWith("command:runtime-reconcile:") && + ((stored.event.payload.delegatedTaskId !== undefined && + (stored.event.payload.status === "queued" || + stored.event.payload.status === "starting")) || + stored.event.payload.status === "completed" || + stored.event.payload.status === "interrupted" || + stored.event.payload.status === "failed" || + stored.event.payload.status === "cancelled" || + stored.event.payload.status === "rolled_back"), + ), ), - Stream.runForEach(handleRunUpdate), - Effect.forkDetach, - ); + eventSink + .stream({ afterSequence: terminalEventsAfterSequence, eventType: "subagent.updated" }) + .pipe( + Stream.filter((stored) => + String(stored.commandId).startsWith( + RunExecutionService.PROVIDER_EXIT_SUBAGENT_CASCADE_COMMAND_PREFIX, + ), + ), + Stream.changesWith((left, right) => left.commandId === right.commandId), + ), + ).pipe( + Stream.runForEach((stored) => + stored.event.type === "run.updated" + ? handleRunUpdate(stored) + : settleDelegatedTaskAfterProviderExit(stored), + ), + Effect.forkDetach, + ); const recoverDelegatedTasks = Effect.gen(function* () { yield* projectionStore.getRecoveryThreadIds("subagent-results").pipe( diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 56db053920a2..6cf7e12f8b53 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -2691,6 +2691,55 @@ it.effect("ProviderSessionManagerV2 releases idle sessions without sweeping all }), ); +it.effect( + "ProviderSessionManagerV2 keeps a reopened session's status when the old release finishes later", + () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const scopeCloseReached = yield* Deferred.make(); + yield* 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 threadId = ThreadId.make("thread-provider-session-manager-reopened-during-release"); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + yield* eventSink.write({ + events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], + }); + yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy }); + + // The old release stalls closing its process; a new turn reopens the + // shared session id meanwhile, as a completion wake does. + const release = yield* manager.close(providerSessionId).pipe(Effect.forkChild); + yield* Deferred.await(scopeCloseReached); + yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy }); + yield* TestClock.adjust("30 seconds"); + yield* Fiber.join(release); + + const projection = yield* projectionStore.getThreadProjection(threadId); + const stopping = yield* manager.shutdown.pipe(Effect.forkChild); + yield* TestClock.adjust("30 seconds"); + yield* Fiber.join(stopping); + assert.equal(projection.providerSessions.at(-1)?.status, "ready"); + }).pipe( + Effect.provide( + layerTest({ + state, + idleTimeoutMs: 60_000, + spawnBeforeOpen: true, + hangSessionScopeClose: true, + scopeCloseReached, + }), + ), + ); + }), +); + it.effect("ProviderSessionManagerV2 persists release when session scope close hangs", () => Effect.gen(function* () { const state = yield* Ref.make(emptyState); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index edcaecfa8848..e099367aaabd 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -989,6 +989,8 @@ export const layerWithOptions = ( readonly cancelIdleFiber?: boolean; readonly onlyIfIdleGeneration?: number; readonly gracefulSubscribers?: boolean; + /** open() releases its own failed start while holding sessionOpen. */ + readonly holdsSessionOpenLock?: boolean; }) => Effect.acquireUseRelease( removeLiveEntry(input), @@ -1028,8 +1030,19 @@ export const layerWithOptions = ( releasedBackgroundTasks, releasedAt, }; + // Closing the scope can take a while. A shared session reopened + // meanwhile owns the session row, so leave its status alone; + // sessionOpen keeps a reopen from landing between check and write. + const writeRecords = Effect.gen(function* () { + const replaced = (yield* Ref.get(sessions)).has( + sessionKey(input.providerSessionId), + ); + yield* writeReleaseRecords({ ...records, replaced }); + }); const recorded = yield* Effect.exit( - writeReleaseRecords({ ...records, replaced: false }), + input.holdsSessionOpenLock === true + ? writeRecords + : sessionOpen.withLock(input.providerSessionId, writeRecords), ); if (Exit.isFailure(recorded)) { yield* retryReleaseRecords(records); @@ -2230,11 +2243,13 @@ export const layerWithOptions = ( providerSessionId: input.providerSessionId, reason: "manual_shutdown", detail: "The provider session start was interrupted.", + holdsSessionOpenLock: true, } : { providerSessionId: input.providerSessionId, reason: "runtime_error", detail: "Failed to persist the provider-session attachment.", + holdsSessionOpenLock: true, }, ).pipe(logReleaseFailure(input.providerSessionId)), ), diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 8cf9a4f28a43..8204c4343944 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -4373,6 +4373,8 @@ function runBackgroundItemScenario( makeEvents: (ids: BackgroundScenarioIds) => ReadonlyArray, options?: { readonly keepEventStreamOpen?: boolean; + readonly failEventStream?: ProviderAdapter.ProviderAdapterEventStreamError; + readonly onWrite?: (events: ReadonlyArray) => Effect.Effect; readonly loadInheritedBackgroundTurnItems?: () => Effect.Effect< ReadonlyArray<{ readonly id: TurnItemId; readonly runId: RunId }> >; @@ -4397,6 +4399,7 @@ function runBackgroundItemScenario( write: () => Effect.succeed([]), writeWithEffects: (input) => Effect.gen(function* () { + yield* options?.onWrite?.(input.events) ?? Effect.void; if ( input.events.some( (event) => event.type === "run.updated" && event.runId === ids.runId, @@ -4447,9 +4450,11 @@ function runBackgroundItemScenario( const events = Stream.fromIterable(makeEvents(ids)); return { events: - options?.keepEventStreamOpen === true - ? events.pipe(Stream.concat(Stream.never)) - : events, + options?.failEventStream !== undefined + ? events.pipe(Stream.concat(Stream.fail(options.failEventStream))) + : options?.keepEventStreamOpen === true + ? events.pipe(Stream.concat(Stream.never)) + : events, close: Deferred.succeed(ingestionDone, undefined), }; }), @@ -4556,6 +4561,46 @@ it.effect("keeps a failed root's still-running native children live while the ru }), ); +it.effect("fails a completed root's native subagents when the provider stream dies", () => + Effect.gen(function* () { + const written = yield* Ref.make>([]); + const observed = yield* runBackgroundItemScenario( + "completed-root-orphaned-child", + (ids) => [ + childThreadCreatedEvent(ids), + subagentEvent(ids, "running"), + backgroundTurnItemEvent(ids, "subagent", "running", 1), + rootTerminalEvent(ids, "completed"), + ], + { + failEventStream: new ProviderAdapter.ProviderAdapterEventStreamError({ + driver, + providerSessionId: ProviderSessionId.make("session:completed-root-orphaned-child"), + cause: "Codex App Server process exited with code 137", + }), + onWrite: (events) => Ref.update(written, (current) => [...current, ...events]), + }, + ); + + assert.deepEqual(observed, ["subagent:running", "turn_item:running", "root-finalized"]); + const events = yield* Ref.get(written); + assert.deepEqual( + events.flatMap((event) => (event.type === "run.updated" ? [event.payload.status] : [])), + ["waiting"], + "the root turn already completed; its own result stands", + ); + assert.deepEqual( + events.flatMap((event) => + event.type === "subagent.updated" || event.type === "turn-item.updated" + ? [`${event.type}:${event.payload.status}`] + : [], + ), + ["subagent.updated:failed", "turn-item.updated:failed"], + "a subagent whose provider died must not stay running", + ); + }), +); + it.effect("releases ingestion after idle subagent rows and items settle", () => Effect.gen(function* () { const observed = yield* runBackgroundItemScenario( diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 9abe812c3d12..00d14a54871f 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -167,6 +167,13 @@ function isRunOwnedSubagentTerminalStatus( return status === "interrupted" || status === "failed" || status === "cancelled"; } +/** + * Tags the write that fails a finalized run's native subagents after its + * provider died. No run update follows it, so the orchestrator watches this tag + * to settle the delegated task waiting on those subagents. + */ +export const PROVIDER_EXIT_SUBAGENT_CASCADE_COMMAND_PREFIX = "command:provider-exit-subagents:"; + function childTurnsContinue(terminal: ProviderTerminalEvent): boolean { return terminal.status === "failed" && terminal.childTurnsContinue === true; } @@ -1165,6 +1172,29 @@ export const layer: Layer.Layer< } } }); + // A completed root keeps ingesting while its native subagents run on. + // If the provider dies then, no event will ever settle them. + const failOpenRunOwnedSubagentsAfterFinalization = Effect.gen(function* () { + const open = yield* Ref.getAndSet( + openRunOwnedSubagents, + emptyOpenRunOwnedSubagentProjection(), + ); + const events = yield* cascadeTerminalizeRunOwnedSubagents({ + run: input.run, + open, + status: "failed", + completedAt: yield* DateTime.now, + allocateEventId: () => idAllocator.allocate.event({ threadId: input.run.threadId }), + }); + if (events.length === 0) return; + yield* eventSink.writeWithEffects({ + commandId: CommandId.make( + `${PROVIDER_EXIT_SUBAGENT_CASCADE_COMMAND_PREFIX}${input.run.id}`, + ), + events, + effects: [], + }); + }); const shouldStopProviderEventIngestion = Effect.gen(function* () { if (!(yield* Ref.get(rootTerminalSeen))) { return false; @@ -1339,7 +1369,7 @@ export const layer: Layer.Layer< }).pipe( Effect.andThen( finalized - ? Effect.void + ? failOpenRunOwnedSubagentsAfterFinalization : Ref.get(latestProviderThread).pipe( Effect.flatMap((providerThread) => Ref.get(latestTurnItemOrdinal).pipe( diff --git a/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts b/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts index 4fb521beac0a..0c474ea6526a 100644 --- a/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts @@ -4,6 +4,7 @@ import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; +import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; import * as CodexReplay from "effect-codex-app-server/replay"; import { @@ -359,6 +360,85 @@ describe("orchestrator replay recovery", () => { ), ); + it.effect( + "fails a Codex run when the app-server exits mid-turn, then serves the next turn on a new one", + () => + Effect.scoped( + Effect.gen(function* () { + const workspace = yield* checkpointWorkspace("provider_thread_resume"); + const recorded = yield* readCodexTranscript(workspace); + // Cut the first app-server off mid-turn: it has started the turn and + // its agent message, then exits without `turn/completed`. The second + // app-server in the recording answers the follow-up message. + const firstExit = recorded.entries.findIndex((entry) => entry.type === "runtime_exit"); + const agentMessageStarted = recorded.entries.findIndex( + (entry, index) => + index < firstExit && + entry.type === "emit_inbound" && + (entry.frame as { readonly method?: string }).method === "item/agentMessage/delta", + ); + assert.isAbove(agentMessageStarted, 0); + const transcript = { + ...recorded, + entries: [ + ...recorded.entries.slice(0, agentMessageStarted), + { type: "runtime_exit" as const, status: "success" as const }, + ...recorded.entries.slice(firstExit + 1), + ], + }; + const driver = yield* CodexReplay.makeReplayDriver(transcript); + const materialized = yield* materializeFixtureInput({ + scenario: "provider_thread_resume", + fixtureInput: { + steps: [ + { type: "message", text: PROVIDER_THREAD_RESUME_FIRST_PROMPT }, + { type: "message", text: PROVIDER_THREAD_RESUME_SECOND_PROMPT }, + ], + }, + driver: ProviderDriverKind.make("codex"), + modelSelection: CODEX_MODEL_SELECTION, + }); + + const result = yield* runOrchestratorV2ProviderReplayScenario( + { + name: "provider_thread_resume/codex:app-server-exit", + transcript, + commands: materialized.steps.flatMap((step) => + step.type === "dispatch" ? [step.command] : [], + ), + steps: materialized.steps, + projectionThreadIds: materialized.projectionThreadIds, + runtimePolicyOverride: { cwd: workspace }, + }, + { + ...CodexOrchestratorReplayHarness, + makeProviderAdapterRegistryLayer: () => + CodexAdapterV2Testkit.layer({ transcript, driver }), + }, + ); + + const projection = projectionFor(result, transcript.scenario); + assert.deepEqual( + projection.runs.map((run) => run.status), + ["failed", "completed"], + ); + const failure = projection.turnItems.find((item) => item.type === "error"); + assert.include( + failure?.type === "error" ? failure.failure.message : "", + "provider event stream closed unexpectedly", + ); + assertAssistantTextIncludes(projection, SECOND_FINAL); + assert.deepEqual(yield* Ref.get(driver.state), { + cursor: transcript.entries.length, + failure: null, + }); + }).pipe( + provideDeterministicTestRuntime, + Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer)), + ), + ), + ); + it.effect( "resumes a provider-native Cursor thread after recreating the orchestrator runtime", () => diff --git a/packages/effect-codex-app-server/src/client.test.ts b/packages/effect-codex-app-server/src/client.test.ts index 842f15c0e912..422fea82724e 100644 --- a/packages/effect-codex-app-server/src/client.test.ts +++ b/packages/effect-codex-app-server/src/client.test.ts @@ -10,6 +10,7 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; import * as CodexClient from "./client.ts"; +import * as CodexError from "./errors.ts"; const mockPeerPath = Effect.map(Effect.service(Path.Path), (path) => path.join(import.meta.dirname, "../test/fixtures/codex-app-server-mock-peer.ts"), @@ -124,6 +125,38 @@ it.layer(NodeServices.layer)("effect-codex-app-server client", (it) => { ]); }), ); + it.effect("reports termination when the app-server exits while the client is idle", () => + Effect.gen(function* () { + const handle = yield* makeHandle({ CODEX_APP_SERVER_TEST_EXIT_AFTER_INITIALIZE: "7" }); + const scope = yield* Scope.make(); + const context = yield* Layer.buildWithScope(CodexClient.layerChildProcess(handle), scope); + + const termination = yield* Effect.gen(function* () { + const client = yield* CodexClient.CodexAppServerClient; + yield* client.request("initialize", { + clientInfo: { + name: "effect-codex-app-server-test", + title: "Effect Codex App Server Test", + version: "0.0.0", + }, + capabilities: { + experimentalApi: true, + optOutNotificationMethods: null, + }, + }); + // No request is pending now, so only the termination signal can tell. + return yield* client.awaitTermination; + }).pipe( + Effect.timeout("5 seconds"), + Effect.provide(context), + Effect.ensuring(Scope.close(scope, Exit.void)), + ); + + assert.instanceOf(termination, CodexError.CodexAppServerProcessExitedError); + assert.equal(termination.code, 7); + }), + ); + it.effect("drains child stderr so large diagnostics cannot block protocol responses", () => Effect.gen(function* () { const handle = yield* makeHandle({ diff --git a/packages/effect-codex-app-server/src/client.ts b/packages/effect-codex-app-server/src/client.ts index e7c5f50b5704..8ab11fe66cee 100644 --- a/packages/effect-codex-app-server/src/client.ts +++ b/packages/effect-codex-app-server/src/client.ts @@ -1,4 +1,5 @@ import * as Context from "effect/Context"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; @@ -40,6 +41,8 @@ export class CodexAppServerClient extends Context.Service< CodexAppServerClient, { readonly raw: CodexAppServerClientRaw; + /** Succeeds with the reason once the app-server exits or its output ends. */ + readonly awaitTermination: Effect.Effect; readonly request: ( method: M, payload: CodexRpc.ClientRequestParamsByMethod[M], @@ -195,6 +198,7 @@ export const make = Effect.fn("effect-codex-app-server/CodexAppServerClient.make : Effect.fail(CodexError.CodexAppServerRequestError.methodNotFound(request.method)); }; + const termination = yield* Deferred.make(); const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({ stdio, ...(terminationError ? { terminationError } : {}), @@ -203,6 +207,7 @@ export const make = Effect.fn("effect-codex-app-server/CodexAppServerClient.make ...(options.logger ? { logger: options.logger } : {}), onNotification: dispatchNotification, onRequest: dispatchRequest, + onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid), }); const request = ( @@ -238,6 +243,7 @@ export const make = Effect.fn("effect-codex-app-server/CodexAppServerClient.make respond: transport.respond, respondError: transport.respondError, }, + awaitTermination: Deferred.await(termination), request, notify, handleServerRequest: (method, handler) => diff --git a/packages/effect-codex-app-server/test/fixtures/codex-app-server-mock-peer.ts b/packages/effect-codex-app-server/test/fixtures/codex-app-server-mock-peer.ts index 59e5eb28faf9..fed62cbc1d46 100644 --- a/packages/effect-codex-app-server/test/fixtures/codex-app-server-mock-peer.ts +++ b/packages/effect-codex-app-server/test/fixtures/codex-app-server-mock-peer.ts @@ -56,6 +56,11 @@ const handleMethod = (message: Record) => { platformFamily: platform === "win32" ? "windows" : "unix", platformOs: platform === "darwin" ? "macos" : platform, }); + const exitCode = process.env.CODEX_APP_SERVER_TEST_EXIT_AFTER_INITIALIZE; + if (exitCode !== undefined) { + // Simulate a crash with stdin still open, like an OOM kill mid-turn. + process.stdout.write("", () => process.exit(Number(exitCode))); + } return; } case "initialized": {