# Files changed · PR \#23854 · pingdotgg/t3code [View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/23854/diffs) Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code) Visibility: public [fix\(server\): release workers when providers stop answering](https://git.cafe/pingdotgg/t3code/pull/23854.md) Version: 8 Head commit: 6349b26ccc5d676b9bcfa77f567845355f6f3485 Base commit: 2c8bdf6177159d8fcd12a897d7c9311f6eb93237 Comparison base: ef92441313da327139df75daa8f723c0f42348f6 Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing. ## apps/server/src/orchestration-v2/EffectWorker.test.ts Status: modified; +233; −13 ```diff @@ -3,6 +3,7 @@ CheckpointId, CheckpointScopeId, CommandId, + MessageId, NodeId, ProviderSessionId, ProviderThreadId, @@ -17,9 +18,11 @@ import * as Deferred from "effect/Deferred"; 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"; +import * as TestClock from "effect/testing/TestClock"; import * as CheckpointRollbackService from "./CheckpointRollbackService.ts"; import * as CheckpointService from "./CheckpointService.ts"; @@ -83,6 +86,7 @@ readonly continueAfterRestart?: boolean; readonly interrupt?: ProviderTurnControlService.ProviderTurnControlServiceV2Shape["interrupt"]; readonly rollback?: CheckpointRollbackService.CheckpointRollbackServiceV2Shape["execute"]; + readonly start?: ProviderTurnStartService.ProviderTurnStartServiceV2Shape["start"]; }) { const record = (event: string) => Ref.update(input.events, (events) => [...events, event]); const layerDependencies = Layer.mergeAll( @@ -115,19 +119,21 @@ Layer.succeed( ProviderTurnStartService.ProviderTurnStartServiceV2, ProviderTurnStartService.ProviderTurnStartServiceV2.of({ + start: + input.start ?? + (() => + Effect.gen(function* () { + yield* record("start"); + if ( + input.failFirstStart !== undefined && + (yield* Ref.getAndSet(input.failFirstStart, false)) + ) { + return yield* new ProviderTurnStartService.ProviderTurnStartError({ + runId, + cause: "simulated first start failure", - start: () => - Effect.gen(function* () { - yield* record("start"); - if ( - input.failFirstStart !== undefined && - (yield* Ref.getAndSet(input.failFirstStart, false)) + }); + } - ) { - return yield* new ProviderTurnStartService.ProviderTurnStartError({ - runId, - cause: "simulated first start failure", - }); - } - }), + })), }), ), Layer.succeed( @@ -586,3 +592,217 @@ ); }), ); + +it.effect("a provider that never answers turn starts does not hold every worker", () => + Effect.gen(function* () { + const now = DateTime.formatIso(yield* DateTime.now); + const startEffect = (name: string): EffectOutbox.OrchestrationEffectV2 => ({ + id: `effect:${name}`, + commandId: CommandId.make(`command:${name}`), + threadId: ThreadId.make(`thread:${name}`), + request: { type: "provider-turn.start", runId: RunId.make(`run:${name}`) }, + status: "running", + attemptCount: 1, + availableAt: now, + leaseOwner: "worker", + leaseExpiresAt: now, + createdAt: now, + updatedAt: now, + completedAt: null, + lastError: null, + }); + const wedged = ["codex-1", "codex-2", "codex-3", "codex-4"].map(startEffect); + const claude = startEffect("claude"); + const pending = yield* Ref.make([...wedged, claude]); + const wedgedStarts = yield* Ref.make(0); + const allWorkersWedged = yield* Deferred.make(); + const claudeStarted = yield* Deferred.make(); + const layerOutbox = Layer.mock(EffectOutbox.EffectOutboxV2)({ + awaitAvailable: Effect.never, + nextClaimableAt: Effect.succeedNone, + claimNext: () => Ref.modify(pending, ([next, ...rest]) => [Option.fromNullishOr(next), rest]), + get: (effectId) => + Effect.succeed(Option.fromNullishOr([...wedged, claude].find((e) => e.id === effectId))), + awaitCancellation: () => Effect.never, + clearCancellation: () => Effect.void, + succeed: () => Effect.succeed(true), + retry: () => Effect.succeed(true), + fail: () => Effect.succeed(true), + }); + const layerExecutor = layerExecutorFor({ + events: yield* Ref.make>([]), + // The shared Codex app-server stays alive but never answers thread/start. + start: (input) => + input.failBeforeStart !== undefined + ? Effect.void + : input.runId === RunId.make("run:claude") + ? Deferred.succeed(claudeStarted, undefined) + : Ref.updateAndGet(wedgedStarts, (count) => count + 1).pipe( + Effect.flatMap((count) => + count === wedged.length + ? Deferred.succeed(allWorkersWedged, undefined) + : Effect.void, + ), + Effect.andThen(Effect.never), + ), + }); + const layerWorker = EffectWorker.layerWithOptions().pipe( + Layer.provide(Layer.merge(layerOutbox, layerExecutor)), + ); + + yield* Effect.gen(function* () { + yield* EffectWorker.runDaemon.pipe(Effect.forkScoped); + yield* Deferred.await(allWorkersWedged); + + // Observed provider-turn.start latency tops out near 33s and leases last + // 30s; fifteen minutes is far past any deadline that could free a slot. + const outcome = yield* Deferred.await(claudeStarted).pipe( + Effect.as("started" as const), + Effect.timeoutOrElse({ duration: "15 minutes", orElse: () => Effect.succeed("stalled") }), + Effect.forkScoped, + ); + yield* TestClock.adjust("15 minutes"); + + assert.equal( + yield* Fiber.join(outcome), + "started", + "four never-resolving Codex provider-turn.start effects still hold all four effect " + + "workers after 15 minutes, so the Claude provider-turn.start never runs", + ); + }).pipe(Effect.scoped, Effect.provide(layerWorker)); + }), +); + +it.effect.each(["timeout", "ordinary"] as const)( + "a steer that failed by %s is retried only when it was not possibly delivered", + (failure) => + Effect.gen(function* () { + const now = DateTime.formatIso(yield* DateTime.now); + const steer: EffectOutbox.OrchestrationEffectV2 = { + id: "effect:steer", + commandId: CommandId.make("command:steer"), + threadId: ThreadId.make("thread:steer"), + request: { + type: "provider-turn.steer", + providerSessionId: ProviderSessionId.make("provider-session:steer"), + providerThreadId: ProviderThreadId.make("provider-thread:steer"), + providerTurnId: ProviderTurnId.make("provider-turn:steer"), + messageId: MessageId.make("message:steer"), + }, + status: "running", + attemptCount: 1, + availableAt: now, + leaseOwner: "worker", + leaseExpiresAt: now, + createdAt: now, + updatedAt: now, + completedAt: null, + lastError: null, + }; + const settled = yield* Ref.make<"retried" | "failed" | null>(null); + const layerOutbox = Layer.mock(EffectOutbox.EffectOutboxV2)({ + claimNext: () => Effect.succeedSome(steer), + get: () => Effect.succeedSome(steer), + awaitCancellation: () => Effect.never, + clearCancellation: () => Effect.void, + retry: () => Ref.set(settled, "retried").pipe(Effect.as(true)), + fail: () => Ref.set(settled, "failed").pipe(Effect.as(true)), + }); + const layerExecutor = Layer.succeed( + EffectWorker.OrchestrationEffectExecutorV2, + EffectWorker.OrchestrationEffectExecutorV2.of({ + execute: (effect) => + Effect.fail( + new EffectWorker.OrchestrationEffectExecutionError({ + effectId: effect.id, + effectType: effect.request.type, + cause: + failure === "timeout" + ? new ProviderTurnControlService.ProviderTurnControlError({ + threadId: effect.threadId, + operation: "steer", + providerTurnId: ProviderTurnId.make("provider-turn:steer"), + cause: new Cause.TimeoutError(), + }) + : new Error("provider rejected the steer"), + }), + ), + }), + ); + + yield* Effect.gen(function* () { + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + yield* worker.runOnce; + }).pipe( + Effect.provide( + EffectWorker.layerWithOptions().pipe( + Layer.provide(Layer.merge(layerOutbox, layerExecutor)), + ), + ), + ); + + assert.equal(yield* Ref.get(settled), failure === "timeout" ? "failed" : "retried"); + }), +); + +it.effect.each([ + { type: "provider-turn.start", final: false }, + { type: "provider-turn.start", final: true }, + { type: "provider-turn.restart", final: false }, + { type: "provider-turn.restart", final: true }, +] as const)( + "a $type that hits the worker backstop settles its run only when final=$final", + ({ type, final }) => + Effect.gen(function* () { + const now = yield* DateTime.now; + const base = restartEffect(now, { type: "detach" }); + const effect: EffectOutbox.OrchestrationEffectV2 = { + ...base, + attemptCount: final ? 5 : 1, + ...(type === "provider-turn.start" + ? { request: { type: "provider-turn.start", runId } } + : {}), + }; + const events = yield* Ref.make>([]); + const outboxOutcome = yield* Ref.make<"retried" | "failed" | null>(null); + const layerOutbox = Layer.mock(EffectOutbox.EffectOutboxV2)({ + claimNext: () => Effect.succeedSome(effect), + get: () => Effect.succeedSome(effect), + awaitCancellation: () => Effect.never, + clearCancellation: () => Effect.void, + retry: () => Ref.set(outboxOutcome, "retried").pipe(Effect.as(true)), + fail: () => Ref.set(outboxOutcome, "failed").pipe(Effect.as(true)), + }); + const layerExecutor = layerExecutorFor({ + events, + // The provider stays alive but never answers the turn start. + start: (input) => + input.failBeforeStart === undefined + ? Effect.never + : Ref.update(events, (current) => [ + ...current, + `fail-before-start:${input.runId}:${Cause.isTimeoutError(input.failBeforeStart)}`, + ]), + }); + + yield* Effect.gen(function* () { + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const fiber = yield* worker.runOnce.pipe(Effect.forkScoped); + yield* TestClock.adjust("12 minutes"); + yield* Fiber.join(fiber); + }).pipe( + Effect.scoped, + Effect.provide( + EffectWorker.layerWithOptions({ maxAttempts: 5 }).pipe( + Layer.provide(Layer.merge(layerOutbox, layerExecutor)), + ), + ), + ); + + assert.equal(yield* Ref.get(outboxOutcome), final ? "failed" : "retried"); + assert.deepEqual( + (yield* Ref.get(events)).filter((event) => event.startsWith("fail-before-start")), + final ? [`fail-before-start:${runId}:true`] : [], + ); + }), +); ``` ## apps/server/src/orchestration-v2/EffectWorker.ts Status: modified; +80; −2 ```diff @@ -68,6 +68,19 @@ OrchestrationEffectExecutorV2Shape >()("t3/orchestration-v2/EffectWorker/OrchestrationEffectExecutorV2") {} +// Backstop for provider turn control effects. Each provider call carries its +// own tighter deadline; this catches unbounded work between them so a provider +// that stopped answering cannot hold a worker forever. +const PROVIDER_TURN_CONTROL_EFFECT_TIMEOUT = Duration.minutes(12); +const PROVIDER_TURN_CONTROL_EFFECT_TYPES = [ + "provider-turn.start", + "provider-turn.interrupt", + "provider-turn.steer", + "provider-turn.restart", +] as const satisfies ReadonlyArray; +const isProviderTurnControl = (effect: EffectOutbox.OrchestrationEffectV2) => + PROVIDER_TURN_CONTROL_EFFECT_TYPES.some((type) => type === effect.request.type); + export const layerExecutor: Layer.Layer< OrchestrationEffectExecutorV2, never, @@ -94,7 +107,7 @@ yield* ThreadTitleRegenerationService.ThreadTitleRegenerationService; const threads = yield* ThreadManagementService.ThreadManagementService; const settings = yield* ServerSettings.ServerSettingsService; + const executor = OrchestrationEffectExecutorV2.of({ - return OrchestrationEffectExecutorV2.of({ execute: (effect, options) => { const willRetry = options?.willRetry ?? false; switch (effect.request.type) { @@ -366,6 +379,19 @@ willRetry, }), ), + // The replacement run waits in "starting" for this effect. + // Once retries are spent, fail it instead of leaving it there. + Effect.tapError((cause) => + willRetry || effect.request.type !== "provider-turn.restart" + ? Effect.void + : providerTurnStart + .start({ + threadId: effect.threadId, + runId: effect.request.runId, + failBeforeStart: cause, + }) + .pipe(Effect.ignore({ log: true })), + ), Effect.mapError( (cause) => new OrchestrationEffectExecutionError({ @@ -540,6 +566,41 @@ } }, }); + return OrchestrationEffectExecutorV2.of({ + execute: (effect, options) => { + if (!isProviderTurnControl(effect)) return executor.execute(effect, options); + const request = effect.request; + return executor.execute(effect, options).pipe( + Effect.timeout(PROVIDER_TURN_CONTROL_EFFECT_TIMEOUT), + Effect.catchTags({ + TimeoutError: (cause) => + // The timeout interrupted the start before it could settle its + // run, so the final attempt fails the run that is still starting. + (options?.willRetry !== true && + (request.type === "provider-turn.start" || request.type === "provider-turn.restart") + ? providerTurnStart + .start({ + threadId: effect.threadId, + runId: request.runId, + failBeforeStart: cause, + }) + .pipe(Effect.ignore({ log: true })) + : Effect.void + ).pipe( + Effect.andThen( + Effect.fail( + new OrchestrationEffectExecutionError({ + effectId: effect.id, + effectType: request.type, + cause, + }), + ), + ), + ), + }), + ); + }, + }); }), ); @@ -570,6 +631,23 @@ OrchestrationEffectWorkerV2Shape >()("t3/orchestration-v2/EffectWorker/OrchestrationEffectWorkerV2") {} +// A steer that timed out may still have been accepted, and providers do not +// deduplicate steering input, so a retry could deliver the message twice. +const uncertainSteer = ( + effect: EffectOutbox.OrchestrationEffectV2, + cause: Cause.Cause, +) => { + if (effect.request.type !== "provider-turn.steer") return false; + const seen = new Set(); + let error: unknown = Cause.squash(cause); + while (typeof error === "object" && error !== null && !seen.has(error)) { + seen.add(error); + if (Cause.isTimeoutError(error)) return true; + error = "cause" in error ? error.cause : undefined; + } + return false; +}; + export interface OrchestrationEffectWorkerOptions { readonly workerId?: string; readonly leaseDurationMs?: number; @@ -769,7 +847,7 @@ ? yield* outbox .succeed({ effectId: effect.id, workerId }) .pipe(Effect.onError((cause) => terminalizeClaim(effect, cause))) - : effect.attemptCount >= maxAttempts + : effect.attemptCount >= maxAttempts || uncertainSteer(effect, exit.cause) ? yield* outbox .fail({ effectId: effect.id, workerId, error }) .pipe(Effect.onError((cause) => terminalizeClaim(effect, cause))) ``` ## apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts Status: modified; +247; −0 ```diff @@ -1,6 +1,7 @@ import { assert, it } from "@effect/vitest"; import { type ModelSelection, + MessageId, NodeId, type OrchestrationV2ProviderThread, type OrchestrationV2ThreadProjection, @@ -14,19 +15,24 @@ RunId, ThreadId, } from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; 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"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; import * as ProviderTurnControlService from "./ProviderTurnControlService.ts"; +import { PROVIDER_CONTROL_TIMEOUT } from "./RunExecutionService.ts"; const driver = ProviderDriverKind.make("codex"); const providerInstanceId = ProviderInstanceId.make("codex"); @@ -329,3 +335,244 @@ assert.equal(interrupted?.nativeThreadRef?.nativeId, "native-thread:restart-session"); }), ); + +it.effect("gives up on an interrupt the provider never answers", () => + Effect.gen(function* () { + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:unanswered-interrupt"); + const providerSessionId = ProviderSessionId.make("provider-session:unanswered-interrupt"); + const providerThreadId = ProviderThreadId.make("provider-thread:unanswered-interrupt"); + const providerTurnId = ProviderTurnId.make("provider-turn:unanswered-interrupt"); + const interruptRequested = yield* Deferred.make(); + const runtime: ProviderAdapter.ProviderAdapterV2SessionRuntime = { + instanceId: providerInstanceId, + driver, + providerSessionId, + providerSession: { + id: providerSessionId, + driver, + providerInstanceId, + status: "running", + cwd: "/workspace", + model: modelSelection.model, + capabilities: CodexProviderCapabilitiesV2, + createdAt: now, + updatedAt: now, + lastError: null, + }, + events: Stream.empty, + ensureThread: () => Effect.die("unused ensureThread"), + resumeThread: () => Effect.die("unused resumeThread"), + startTurn: () => Effect.die("unused startTurn"), + steerTurn: () => Effect.die("unused steerTurn"), + // The process stays alive but never answers turn/interrupt. + interruptTurn: () => + Deferred.succeed(interruptRequested, undefined).pipe(Effect.andThen(Effect.never)), + respondToRuntimeRequest: () => Effect.die("unused respondToRuntimeRequest"), + readThreadSnapshot: () => Effect.die("unused readThreadSnapshot"), + rollbackThread: () => Effect.die("unused rollbackThread"), + forkThread: () => Effect.die("unused forkThread"), + }; + const layerProjection = Layer.mock(ProjectionStore.ProjectionStoreV2)({ + getProviderControlContext: () => + Effect.succeed({ + providerThread: { + id: providerThreadId, + driver, + providerInstanceId, + providerSessionId, + appThreadId: threadId, + ownerNodeId: null, + nativeThreadRef: null, + nativeConversationHeadRef: null, + status: "active", + firstRunOrdinal: 1, + lastRunOrdinal: 1, + handoffIds: [], + forkedFrom: null, + createdAt: now, + updatedAt: now, + }, + providerTurn: { + id: providerTurnId, + providerThreadId, + nodeId: NodeId.make("node:unanswered-interrupt"), + runAttemptId: RunAttemptId.make("run-attempt:unanswered-interrupt"), + nativeTurnRef: null, + ordinal: 1, + status: "running", + startedAt: now, + completedAt: null, + }, + attempt: undefined, + message: undefined, + run: undefined, + }), + }); + const layerSessionManager = Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({ + get: () => Effect.succeedSome(runtime), + }); + + const exit = yield* Effect.gen(function* () { + const control = yield* ProviderTurnControlService.ProviderTurnControlServiceV2; + const interrupt = yield* control + .interrupt({ threadId, providerSessionId, providerThreadId, providerTurnId }) + .pipe(Effect.exit, Effect.forkScoped); + yield* Deferred.await(interruptRequested); + yield* TestClock.adjust(PROVIDER_CONTROL_TIMEOUT); + return yield* Fiber.join(interrupt); + }).pipe( + Effect.scoped, + Effect.provide( + ProviderTurnControlService.layer.pipe( + Layer.provide(Layer.merge(layerProjection, layerSessionManager)), + ), + ), + ); + + assert.isTrue(Exit.isFailure(exit)); + if (Exit.isFailure(exit)) assert.include(Cause.pretty(exit.cause), "TimeoutError"); + }), +); + +it.effect("a steer that times out after its turn completes is not offered as a follow-up", () => + Effect.gen(function* () { + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:unanswered-steer"); + const providerSessionId = ProviderSessionId.make("provider-session:unanswered-steer"); + const providerThreadId = ProviderThreadId.make("provider-thread:unanswered-steer"); + const providerTurnId = ProviderTurnId.make("provider-turn:unanswered-steer"); + const messageId = MessageId.make("message:unanswered-steer"); + const runId = RunId.make("run:unanswered-steer"); + const turnStatus = yield* Ref.make<"running" | "completed">("running"); + const steerRequested = yield* Deferred.make(); + const runtime: ProviderAdapter.ProviderAdapterV2SessionRuntime = { + instanceId: providerInstanceId, + driver, + providerSessionId, + providerSession: { + id: providerSessionId, + driver, + providerInstanceId, + status: "running", + cwd: "/workspace", + model: modelSelection.model, + capabilities: CodexProviderCapabilitiesV2, + createdAt: now, + updatedAt: now, + lastError: null, + }, + events: Stream.empty, + ensureThread: () => Effect.die("unused ensureThread"), + resumeThread: () => Effect.die("unused resumeThread"), + startTurn: () => Effect.die("unused startTurn"), + // The provider accepts the input and finishes the turn, but the + // acknowledgement never arrives. + steerTurn: () => + Ref.set(turnStatus, "completed").pipe( + Effect.andThen(Deferred.succeed(steerRequested, undefined)), + Effect.andThen(Effect.never), + ), + interruptTurn: () => Effect.die("unused interruptTurn"), + respondToRuntimeRequest: () => Effect.die("unused respondToRuntimeRequest"), + readThreadSnapshot: () => Effect.die("unused readThreadSnapshot"), + rollbackThread: () => Effect.die("unused rollbackThread"), + forkThread: () => Effect.die("unused forkThread"), + }; + const layerProjection = Layer.mock(ProjectionStore.ProjectionStoreV2)({ + getProviderControlContext: () => + Effect.gen(function* () { + return { + providerThread: { + id: providerThreadId, + driver, + providerInstanceId, + providerSessionId, + appThreadId: threadId, + ownerNodeId: null, + nativeThreadRef: null, + nativeConversationHeadRef: null, + status: "active" as const, + firstRunOrdinal: 1, + lastRunOrdinal: 1, + handoffIds: [], + forkedFrom: null, + createdAt: now, + updatedAt: now, + }, + providerTurn: { + id: providerTurnId, + providerThreadId, + nodeId: NodeId.make("node:unanswered-steer"), + runAttemptId: RunAttemptId.make("run-attempt:unanswered-steer"), + nativeTurnRef: null, + ordinal: 1, + status: yield* Ref.get(turnStatus), + startedAt: now, + completedAt: null, + }, + attempt: undefined, + message: { + createdBy: "user" as const, + creationSource: "web" as const, + id: messageId, + threadId, + runId, + nodeId: null, + role: "user" as const, + text: "also check the tests", + attachments: [], + streaming: false, + createdAt: now, + updatedAt: now, + }, + run: { + id: runId, + threadId, + ordinal: 1, + providerInstanceId, + modelSelection, + providerThreadId, + userMessageId: MessageId.make("message:unanswered-steer-run"), + rootNodeId: null, + activeAttemptId: null, + status: "running" as const, + requestedAt: now, + startedAt: now, + completedAt: null, + checkpointId: null, + contextHandoffId: null, + }, + }; + }), + }); + const layerSessionManager = Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({ + get: () => Effect.succeedSome(runtime), + }); + + const exit = yield* Effect.gen(function* () { + const control = yield* ProviderTurnControlService.ProviderTurnControlServiceV2; + const steer = yield* control + .steer({ threadId, providerSessionId, providerThreadId, providerTurnId, messageId }) + .pipe(Effect.exit, Effect.forkScoped); + yield* Deferred.await(steerRequested); + yield* TestClock.adjust(PROVIDER_CONTROL_TIMEOUT); + return yield* Fiber.join(steer); + }).pipe( + Effect.scoped, + Effect.provide( + ProviderTurnControlService.layer.pipe( + Layer.provide(Layer.merge(layerProjection, layerSessionManager)), + ), + ), + ); + + assert.isTrue(Exit.isFailure(exit)); + if (Exit.isFailure(exit)) { + const error = Cause.squash(exit.cause); + assert.instanceOf(error, ProviderTurnControlService.ProviderTurnControlError); + assert.isFalse(error.turnCompleted); + assert.isTrue(Cause.isTimeoutError(error.cause)); + } + }), +); ``` ## apps/server/src/orchestration-v2/ProviderTurnControlService.ts Status: modified; +20; −10 ```diff @@ -8,6 +8,7 @@ RunAttemptId, ThreadId, } from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; import * as Clock from "effect/Clock"; import type * as DateTime from "effect/DateTime"; import * as Context from "effect/Context"; @@ -19,6 +20,7 @@ import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; +import { PROVIDER_CONTROL_TIMEOUT } from "./RunExecutionService.ts"; const yieldToRuntime = Effect.yieldNow.pipe( Effect.andThen( @@ -216,11 +218,13 @@ // A settled turn reaches its adapter too: only the adapter knows // whether it still runs work for the thread, and each one either // stops it or reports there is nothing left to stop. + yield* session.value + .interruptTurn({ + providerThread: loaded.providerThread, + providerTurnId: loaded.providerTurn.id, + requestRuntimeRestart: true, + }) - yield* session.value.interruptTurn({ - providerThread: loaded.providerThread, - providerTurnId: loaded.providerTurn.id, + .pipe(Effect.timeout(PROVIDER_CONTROL_TIMEOUT)); - requestRuntimeRestart: true, - }); // The wait is real time: ingestion runs on other fibers and never // advances a test clock, so a test clock would hold Stop forever. yield* Effect.gen(function* () { @@ -274,10 +278,12 @@ return; } - yield* loaded.session.value.interruptTurn({ - providerThread: loaded.providerThread, - providerTurnId: loaded.providerTurn.id, + yield* loaded.session.value + .interruptTurn({ + providerThread: loaded.providerThread, + providerTurnId: loaded.providerTurn.id, + }) + .pipe(Effect.timeout(PROVIDER_CONTROL_TIMEOUT)); - }); for (let remaining = 1_000; remaining > 0; remaining -= 1) { const { providerTurn, attempt } = yield* projections.getProviderControlContext( @@ -377,6 +383,7 @@ }, }) .pipe( + Effect.timeout(PROVIDER_CONTROL_TIMEOUT), Effect.catch((cause) => Effect.gen(function* () { const current = yield* projections.getProviderControlContext( @@ -387,7 +394,10 @@ threadId: input.threadId, operation: "steer", providerTurnId: input.providerTurnId, + // A timed-out steer may have been accepted, so it must not be + // resent as a follow-up even if the turn has since completed. + turnCompleted: + !Cause.isTimeoutError(cause) && current.providerTurn?.status === "completed", - turnCompleted: current.providerTurn?.status === "completed", cause, }); }), ``` ## apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts Status: modified; +20; −0 ```diff @@ -771,9 +771,29 @@ willRetry: true, }); }).pipe(Effect.provide(layer)), + failBeforeStart: (error: Error) => + Effect.gen(function* () { + yield* (yield* ProviderTurnStart.ProviderTurnStartServiceV2).start({ + threadId, + runId, + failBeforeStart: error, + }); + }).pipe(Effect.provide(layer)), }; } +effectIt.effect("fails a starting run whose restart could not stop the previous turn", () => + Effect.gen(function* () { + const harness = makeLocalCommandHarness({ text: "Continue" }); + + yield* harness.failBeforeStart(new Error("The provider did not answer the interrupt.")); + + expect(harness.projection().runs.at(-1)?.status).toBe("failed"); + expect(harness.open).not.toHaveBeenCalled(); + expect(harness.startRootRun).not.toHaveBeenCalled(); + }), +); + effectIt.effect("recovers an abandoned failed turn before continuing its native conversation", () => Effect.gen(function* () { const harness = makeLocalCommandHarness({ ``` ## apps/server/src/orchestration-v2/ProviderTurnStartService.ts Status: modified; +69; −30 ```diff @@ -88,6 +88,8 @@ readonly threadId: ThreadId; readonly runId: RunId; readonly willRetry?: boolean; + /** Fails the still-starting run with this error instead of starting it. */ + readonly failBeforeStart?: Error; }) => Effect.Effect; } @@ -233,6 +235,7 @@ readonly threadId: ThreadId; readonly runId: RunId; readonly willRetry?: boolean; + readonly failBeforeStart?: Error; }) { const { runId } = input; const projection = yield* projectionStore.getTurnStartContext(input.threadId, runId); @@ -386,6 +389,24 @@ }); }, ); + if (input.failBeforeStart !== undefined) { + yield* settleRunBeforeStart({ + signal: "provider-turn-restart-failure", + status: "failed", + now: yield* DateTime.now, + providerInstanceId: run.providerInstanceId, + itemProviderThreadId: providerThread.id, + item: { + type: "error", + title: "Provider restart failed", + failure: makeProviderFailure({ + cause: input.failBeforeStart, + class: "provider_error", + }), + }, + }); + return; + } if (message.attachments.length === 0 && message.text.trimStart().startsWith("/")) { const isEmptyCompaction = message.text.trim().toLowerCase() === "/compact" && !projection.hasConversation; @@ -598,11 +619,13 @@ // Older versions could fail the run without stopping its provider. // Recover that exact turn before reusing its native conversation. const recovered = yield* Effect.result( - liveSession.value.interruptTurn({ - providerThread, - providerTurnId: abandonedTurn.id, - requestRuntimeRestart: true, + liveSession.value + .interruptTurn({ + providerThread, + providerTurnId: abandonedTurn.id, + requestRuntimeRestart: true, + }) + .pipe(Effect.timeout(RunExecutionService.PROVIDER_CONTROL_TIMEOUT)), - }), ); if (recovered._tag === "Failure") { if (input.willRetry === true) return yield* recovered.failure; @@ -638,24 +661,26 @@ if (!repaired.committed) return; } const sessionResult = yield* Effect.result( + providerSessions + .open({ + threadId: projection.thread.id, + providerSessionId, + modelSelection: run.modelSelection, + runtimePolicy: resolvedRuntimePolicy, + ...(existingSessionProjection === undefined + ? {} + : { resumeFromSession: existingSessionProjection }), + ...(providerThread.nativeThreadRef?.nativeId == null + ? {} + : { initialNativeThreadId: providerThread.nativeThreadRef.nativeId }), + ...(providerThread.nativeMetadata?.itemIdentityVersion === undefined + ? {} + : { + initialProviderItemIdentityVersion: + providerThread.nativeMetadata.itemIdentityVersion, - providerSessions.open({ + }), + }) + .pipe(Effect.timeout(RunExecutionService.PROVIDER_CONTROL_TIMEOUT)), - threadId: projection.thread.id, - providerSessionId, - modelSelection: run.modelSelection, - runtimePolicy: resolvedRuntimePolicy, - ...(existingSessionProjection === undefined - ? {} - : { resumeFromSession: existingSessionProjection }), - ...(providerThread.nativeThreadRef?.nativeId == null - ? {} - : { initialNativeThreadId: providerThread.nativeThreadRef.nativeId }), - ...(providerThread.nativeMetadata?.itemIdentityVersion === undefined - ? {} - : { - initialProviderItemIdentityVersion: - providerThread.nativeMetadata.itemIdentityVersion, - }), - }), ); if (sessionResult._tag === "Failure") { if (input.willRetry === true) return yield* sessionResult.failure; @@ -671,7 +696,9 @@ const loadFromProvider = Effect.fn("ProviderTurnStartService.loadFromProvider")(function* ( load: Effect.Effect, ) { - const loaded = yield* Effect.result(load); + const loaded = yield* Effect.result( + Effect.timeout(load, RunExecutionService.PROVIDER_CONTROL_TIMEOUT), + ); if (loaded._tag === "Success") return loaded.success; if (input.willRetry === true) return yield* loaded.failure; yield* settleStartFailure({ @@ -754,12 +781,14 @@ cause: "Uncertain native history injection", }), ) - : session.resumeThread({ - providerThread, - threadId: projection.thread.id, - modelSelection: run.modelSelection, - runtimePolicy: resolvedRuntimePolicy, - }), + : session + .resumeThread({ + providerThread, + threadId: projection.thread.id, + modelSelection: run.modelSelection, + runtimePolicy: resolvedRuntimePolicy, + }) + .pipe(Effect.timeout(RunExecutionService.PROVIDER_CONTROL_TIMEOUT)), ); if (resumed._tag === "Success") { return resumed.success; @@ -1319,6 +1348,16 @@ shouldStartProviderTurn: runControls.shouldStartProviderTurn, shouldFinalizeRun: runControls.shouldFinalizeRun, hasUnpairedRunInterruptRequest: runControls.hasUnpairedRunInterruptRequest, + // Ending the session stops the process, so a turn it accepted without + // answering cannot keep editing the workspace. The next start reopens + // the session and resumes the native conversation. + containUnansweredStart: providerSessions + .release({ + providerSessionId, + reason: "runtime_error", + detail: "The provider did not answer the turn start.", + }) + .pipe(Effect.ignore({ log: true })), message: { messageId: message.id, text: userText, ``` ## apps/server/src/orchestration-v2/RunExecutionService.test.ts Status: modified; +27; −0 ```diff @@ -30,11 +30,13 @@ import * as Exit from "effect/Exit"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; +import * as Fiber from "effect/Fiber"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import * as ServerSettings from "../serverSettings.ts"; import * as CheckpointService from "./CheckpointService.ts"; @@ -2589,6 +2591,27 @@ }), ); +it.effect("contains and fails a run whose provider never answers the turn start", () => + Effect.gen(function* () { + const startRequested = yield* Deferred.make(); + const contained = yield* Ref.make(false); + const termination = yield* captureRootRunTermination({ + key: "unanswered-start", + containUnansweredStart: Ref.set(contained, true), + shouldFinalizeRun: () => Effect.succeed(true), + events: () => Stream.never, + startTurn: () => + Deferred.succeed(startRequested, undefined).pipe(Effect.andThen(Effect.never)), + }).pipe(Effect.forkScoped); + yield* Deferred.await(startRequested); + yield* TestClock.adjust(RunExecutionService.PROVIDER_CONTROL_TIMEOUT); + + const { observed } = yield* Fiber.join(termination); + assert.include(observed, "run:failed"); + assert.isTrue(yield* Ref.get(contained)); + }), +); + it.effect("does not interrupt an ended provider when its terminal write fails", () => Effect.gen(function* () { const interrupts = yield* Ref.make(0); @@ -2718,6 +2741,7 @@ readonly refreshAfterTurn?: Effect.Effect; readonly onSubscriptionClose?: Effect.Effect; readonly shouldStartProviderTurn?: RunExecutionService.RunExecutionServiceV2StartRootRunInput["shouldStartProviderTurn"]; + readonly containUnansweredStart?: Effect.Effect; readonly appContext?: ReturnType; readonly ingestNormalized?: ProviderEventIngestor.ProviderEventIngestorV2["Service"]["ingestNormalized"]; readonly beforeTerminalWrite?: ( @@ -2881,6 +2905,9 @@ attemptId: ids.attemptId, providerTurnOrdinal: 1, shouldFinalizeRun: input.shouldFinalizeRun, + ...(input.containUnansweredStart === undefined + ? {} + : { containUnansweredStart: input.containUnansweredStart }), ...(input.shouldStartProviderTurn === undefined ? {} : { shouldStartProviderTurn: input.shouldStartProviderTurn }), ``` ## apps/server/src/orchestration-v2/RunExecutionService.ts Status: modified; +20; −1 ```diff @@ -545,6 +545,11 @@ } } +// A provider that stays alive but stops answering control requests would +// otherwise hold an effect worker forever. Healthy starts finish within ~33s; +// this bounds each control call, not the agent turn that follows acceptance. +export const PROVIDER_CONTROL_TIMEOUT = Duration.minutes(2); + export const RunExecutionServiceV2Error = Schema.Union([ RunExecutionStartError, RunExecutionIngestError, @@ -573,6 +578,12 @@ readonly shouldStartProviderTurn?: () => Effect.Effect; readonly shouldFinalizeRun?: () => Effect.Effect; readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect; + /** + * Runs when the start call outlives its deadline. The provider may have + * accepted the turn without answering, so its native work must be stopped + * before the run reads as failed and the conversation is reused. + */ + readonly containUnansweredStart?: Effect.Effect; readonly message: ProviderAdapter.ProviderAdapterV2TurnMessage; readonly modelSelection: ModelSelection; readonly runtimePolicy: ProviderAdapter.ProviderAdapterV2RuntimePolicy; @@ -1621,7 +1632,10 @@ }), )) : input.session.startTurn(turnInput); + yield* Effect.andThen( + shouldStart, + Effect.timeout(startTurn, PROVIDER_CONTROL_TIMEOUT), - yield* Effect.andThen(shouldStart, startTurn).pipe( + ).pipe( // Wait for acceptance so abandoned startup cannot leave a refresher // on the already-detached ingestion fiber. Effect.tap(() => @@ -1636,6 +1650,11 @@ cause, }).pipe( Effect.andThen(Fiber.interrupt(providerEventFiber)), + Effect.andThen( + Cause.isTimeoutError(Cause.squash(cause)) + ? (input.containUnansweredStart ?? Effect.void) + : Effect.void, + ), Effect.andThen(Ref.get(latestProviderThread)), Effect.flatMap((providerThread) => Ref.get(latestTurnItemOrdinal).pipe( ``` ## packages/provider-opencode/src/server/adapter.test.ts Status: modified; +61; −0 ```diff @@ -28,6 +28,7 @@ import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import * as ProviderHost from "@t3tools/provider-core/server/ProviderHost"; import * as TestProviderHost from "@t3tools/provider-testing/TestProviderHost"; @@ -1182,6 +1183,66 @@ ), ); + it.effect("returns from compaction start while a long summarize keeps the turn active", () => + Effect.gen(function* () { + const nativeEvents = asyncEventStream(); + const summarizeStarted = promiseGate(); + const summarizeResult = promiseGate<{ data: boolean }>(); + const harness = yield* makeOpenCodeRuntimeHarness( + "long-compact", + "native-opencode-long-compact", + { + event: { + subscribe: async (_input: unknown, options: { signal?: AbortSignal }) => { + options.signal?.addEventListener("abort", () => nativeEvents.close(), { once: true }); + return { stream: nativeEvents.stream }; + }, + }, + session: { + create: async () => ({ + data: { id: "native-opencode-long-compact", time: { created: 1, updated: 1 } }, + }), + summarize: () => { + summarizeStarted.resolve(); + return summarizeResult.promise; + }, + }, + }, + ); + const events = yield* harness.runtime.events.pipe( + Stream.takeUntil((event) => event.type === "turn.terminal"), + Stream.runCollect, + Effect.forkScoped, + ); + // Orchestration bounds turn start as a control call; a summarize that + // outlives that bound must not hold the start open. + const start = yield* harness + .startTurn("/compact") + .pipe(Effect.timeout("2 minutes"), Effect.exit, Effect.forkScoped); + yield* Effect.promise(() => summarizeStarted.promise); + yield* TestClock.adjust("3 minutes"); + assert.isTrue(Exit.isSuccess(yield* Fiber.join(start))); + summarizeResult.resolve({ data: true }); + const received = Array.from(yield* Fiber.join(events)); + const terminals = received.filter((event) => event.type === "turn.terminal"); + assert.deepEqual( + terminals.map((event) => event.status), + ["completed"], + ); + assert.isTrue( + received.some( + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "compaction" && + event.turnItem.status === "completed", + ), + ); + }).pipe( + Effect.provide(Layer.merge(IdAllocator.layer, McpProviderSessions.layer)), + Effect.scoped, + ), + ); + it.effect("ends a turn on the provider thread it started on", () => Effect.gen(function* () { const nativeEvents = asyncEventStream(); ``` ## packages/provider-opencode/src/server/adapter.ts Status: modified; +4; −0 ```diff @@ -3208,6 +3208,8 @@ }); yield* updateProviderSession("running", null); if (isCompaction) { + // summarize resolves only when compaction finishes, so it runs + // past turn start; the turn's terminal event reports its outcome. yield* sdkCall( "session.summarize", { sessionID: sessionId, ...parsedModel, auto: false }, @@ -3231,6 +3233,8 @@ }), ), Effect.ensuring(Deferred.succeed(admissionSettled, undefined)), + Effect.ignore, + Effect.forkIn(scope), ); return; } ```