diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index 556e03d128712..936deff745752 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -396,12 +396,14 @@ function makeLocalCommandHarness(input: { readonly failReadsAfterRunning?: boolean; readonly recoveryReadFailure?: string; readonly previousTurn?: { - readonly runStatus: "failed" | "completed"; + readonly runStatus: "failed" | "completed" | "interrupted"; readonly turnStatus: "running" | "completed" | "interrupted"; readonly differentThread?: boolean; readonly liveSession?: boolean; readonly interruptFailure?: boolean; readonly interruptCurrentRun?: boolean; + /** Puts the running turn on an attempt a steering restart superseded. */ + readonly supersededTurn?: boolean; }; /** * A side chat whose context targets this run, an earlier run that failed or was cancelled before @@ -861,19 +863,57 @@ function makeLocalCommandHarness(input: { if (input.previousTurn !== undefined) { const previousRunId = RunId.make("previous-failed-run"); const previousAttemptId = RunAttemptId.make("previous-failed-attempt"); + const steeredAttemptId = RunAttemptId.make("previous-steered-attempt"); const previousProviderThreadId = input.previousTurn.differentThread ? oldProviderThreadId : providerThreadId; projection = { ...projection, + attempts: [ + { + id: previousAttemptId, + runId: previousRunId, + attemptOrdinal: 1, + rootNodeId: NodeId.make("previous-root"), + providerInstanceId: newInstanceId, + providerThreadId: previousProviderThreadId, + providerTurnId: null, + reason: "initial", + status: input.previousTurn.supersededTurn === true ? "superseded" : "interrupted", + startedAt: now, + completedAt: now, + }, + ...(input.previousTurn.supersededTurn === true + ? [ + { + id: steeredAttemptId, + runId: previousRunId, + attemptOrdinal: 2, + rootNodeId: NodeId.make("previous-steered-root"), + providerInstanceId: newInstanceId, + providerThreadId: previousProviderThreadId, + providerTurnId: null, + reason: "steering_restart" as const, + status: "interrupted" as const, + startedAt: null, + completedAt: now, + }, + ] + : []), + ...projection.attempts, + ], runs: [ { ...run, id: previousRunId, ordinal: 1, status: input.previousTurn.runStatus, - rootNodeId: NodeId.make("previous-root"), - activeAttemptId: previousAttemptId, + rootNodeId: + input.previousTurn.supersededTurn === true + ? NodeId.make("previous-steered-root") + : NodeId.make("previous-root"), + activeAttemptId: + input.previousTurn.supersededTurn === true ? steeredAttemptId : previousAttemptId, providerThreadId: previousProviderThreadId, completedAt: now, }, @@ -890,6 +930,23 @@ function makeLocalCommandHarness(input: { }, ], providerTurns: [ + // The steered attempt reached the provider once; only the earlier attempt's + // turn is still running when the run was interrupted. + ...(input.previousTurn.supersededTurn === true + ? [ + { + id: ProviderTurnId.make("previous-steered-turn"), + providerThreadId: previousProviderThreadId, + nodeId: NodeId.make("previous-steered-root"), + runAttemptId: steeredAttemptId, + ordinal: 0, + nativeTurnRef: null, + status: "interrupted" as const, + startedAt: now, + completedAt: now, + }, + ] + : []), { id: previousTurnId, providerThreadId: previousProviderThreadId, @@ -905,7 +962,8 @@ function makeLocalCommandHarness(input: { }; } let providerBusy = - input.previousTurn?.runStatus === "failed" && + (input.previousTurn?.runStatus === "failed" || + input.previousTurn?.runStatus === "interrupted") && input.previousTurn.turnStatus === "running" && !input.previousTurn.differentThread && input.previousTurn.liveSession !== false; @@ -1529,6 +1587,32 @@ effectIt.effect("recovers an abandoned failed turn before continuing its native }), ); +effectIt.effect( + "recovers a running turn its interrupted run left behind on a superseded attempt", + () => + Effect.gen(function* () { + const harness = makeLocalCommandHarness({ + text: "Continue", + previousTurn: { runStatus: "interrupted", turnStatus: "running", supersededTurn: true }, + }); + + yield* harness.start; + + expect(harness.interruptTurn).toHaveBeenCalledOnce(); + // The steered attempt's turn was already interrupted in the fixture; the + // repair this asserts is the running turn left on the superseded attempt. + const abandoned = harness + .projection() + .providerTurns.find((turn) => turn.id === ProviderTurnId.make("previous-provider-turn")); + expect(abandoned).toMatchObject({ + status: "interrupted", + completedAt: expect.anything(), + }); + expect(harness.projection().runs.at(-1)?.status).toBe("running"); + expect(harness.startRootRun).toHaveBeenCalledOnce(); + }), +); + effectIt.effect("sends new steering text when restarting a manual continuation", () => Effect.gen(function* () { const harness = makeLocalCommandHarness({ diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 15792dacecfa2..f6ad1471de56b 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -78,6 +78,14 @@ export class ProviderTurnStartError extends Schema.TaggedError = new Set([ + "failed", + "interrupted", + "cancelled", + "rolled_back", +]); + const isProviderTurnStartError = Schema.is(ProviderTurnStartError); const isProviderAdapterProtocolError = Schema.is(ProviderAdapter.ProviderAdapterProtocolError); const isProviderAdapterOpenSessionError = Schema.is( @@ -652,22 +660,31 @@ export const layer: Layer.Layer< const existingSessionProjection = projection.providerSessions.find( (candidate) => candidate.id === providerSessionId, ); - const failedRuns = new Map( - projection.runs - .filter( - (previous) => - previous.ordinal < run.ordinal && - previous.providerThreadId === providerThread.id && - previous.status === "failed", - ) - .map((previous) => [previous.activeAttemptId, previous] as const), - ); + // A run that ended without completing can leave its provider turn projected as + // running when the agent ignored Stop or an interrupt timed out. Every attempt + // of such a run owns a possible abandoned turn, including attempts a steering + // restart superseded, and the session cannot be reused until those turns stop. + const abandonedRuns = new Map(); + for (const previous of projection.runs) { + if ( + previous.ordinal >= run.ordinal || + previous.providerThreadId !== providerThread.id || + !NON_COMPLETING_TERMINAL_RUN_STATUSES.has(previous.status) + ) { + continue; + } + if (previous.activeAttemptId !== null) { + abandonedRuns.set(previous.activeAttemptId, previous); + } + for (const attempt of projection.attempts) { + if (attempt.runId === previous.id) abandonedRuns.set(attempt.id, previous); + } + } const abandonedTurns = projection.providerTurns.flatMap((turn) => { - const previous = failedRuns.get(turn.runAttemptId); + const previous = abandonedRuns.get(turn.runAttemptId); return turn.providerThreadId === providerThread.id && turn.status === "running" && - previous !== undefined && - turn.nodeId === previous.rootNodeId + previous !== undefined ? [{ turn, run: previous }] : []; }); @@ -675,8 +692,9 @@ export const layer: Layer.Layer< if (abandonedTurn !== undefined) { const liveSession = yield* providerSessions.get(providerSessionId); if (Option.isSome(liveSession)) { - // Older versions could fail the run without stopping its provider. - // Recover that exact turn before reusing its native conversation. + // The run already ended without stopping its provider — a failed start, an + // ignored Stop, or an interrupt that never confirmed. Recover that exact + // turn before reusing its native conversation. const recovered = yield* Effect.result( liveSession.value.interruptTurn({ providerThread, @@ -700,15 +718,15 @@ export const layer: Layer.Layer< runId: run.id, activeAttemptId: attempt.id, expectedStatus: "starting", - events: yield* Effect.forEach(abandonedTurns, ({ turn, run: failedRun }) => + events: yield* Effect.forEach(abandonedTurns, ({ turn, run: abandonedRun }) => Effect.gen(function* () { return { id: yield* idAllocator.allocate.event({ threadId: projection.thread.id }), type: "provider-turn.updated" as const, threadId: projection.thread.id, - runId: failedRun.id, + runId: abandonedRun.id, nodeId: turn.nodeId, - providerInstanceId: failedRun.providerInstanceId, + providerInstanceId: abandonedRun.providerInstanceId, occurredAt: completedAt, payload: { ...turn, status: "interrupted" as const, completedAt }, };