diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 9880da3408a77..42ec1118d4bbe 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -116,6 +116,18 @@ const refusedBeforePrompt = (error: unknown): boolean => PROMPT_REFUSAL_TAGS.some((tag) => Predicate.isTagged(error, tag)) || (Predicate.hasProperty(error, "cause") && refusedBeforePrompt(error.cause)); +const startFailureMessage = (error: Error): string => { + const nestedCause = "cause" in error ? error.cause : undefined; + return ( + knownProviderFailureMessage(error) ?? + (nestedCause instanceof Error + ? nestedCause.message + : typeof nestedCause === "string" + ? nestedCause + : error.message) + ); +}; + export interface ProviderTurnStartServiceV2Shape { readonly start: (input: { readonly threadId: ThreadId; @@ -525,7 +537,6 @@ export const layer: Layer.Layer< readonly error: Error; readonly message?: string; }) { - const nestedCause = "cause" in failed.error ? failed.error.cause : undefined; yield* settleRunBeforeStart({ signal: failed.signal, status: "failed", @@ -537,14 +548,7 @@ export const layer: Layer.Layer< title: failed.title, failure: makeProviderFailure({ cause: failed.error, - message: - failed.message ?? - knownProviderFailureMessage(failed.error) ?? - (nestedCause instanceof Error - ? nestedCause.message - : typeof nestedCause === "string" - ? nestedCause - : failed.error.message), + message: failed.message ?? startFailureMessage(failed.error), class: "provider_error", }), }, @@ -1151,6 +1155,7 @@ export const layer: Layer.Layer< const sourceBusy = sourceProjection.runs.some( (candidate) => candidate.ordinal > sourceRun.ordinal && isActiveRun(candidate), ); + let nativeForkFailure: string | undefined; if (!sourceBusy) { // Rolled-back turns are no longer in the source conversation, so they cannot bound a fork. const rolledBackAttemptIds = new Set( @@ -1163,7 +1168,7 @@ export const layer: Layer.Layer< ) .map((attempt) => attempt.id), ); - return yield* loadFromProvider( + const forked = yield* Effect.result( session.forkThread({ sourceProviderThread, sourceProviderTurns: sourceProjection.providerTurns.filter( @@ -1178,7 +1183,24 @@ export const layer: Layer.Layer< : { providerTurnId: sourceProviderTurn.id }), }), ); + if (forked._tag === "Success") return forked.success; + if (input.willRetry === true) return yield* forked.failure; + if (!(yield* isCurrentAttemptInStatus("starting"))) return undefined; + // Retries are spent (e.g. the source's native session is gone): start fresh with portable history. + yield* Effect.logWarning("Starting a native fork in a fresh native session", { + driver: session.driver, + providerThreadId: providerThread.id, + sourceProviderThreadId: sourceProviderThread.id, + runId, + errorTag: forked.failure._tag, + }); + nativeForkFailure = makeProviderFailure({ + cause: forked.failure, + message: startFailureMessage(forked.failure), + class: "provider_error", + }).message; } + const nativeForkFailed = nativeForkFailure !== undefined; const sourceItems = yield* ContextHandoffService.readForkHistoryItems( projectionStore, sourceProjection, @@ -1209,7 +1231,11 @@ export const layer: Layer.Layer< modelSelection: run.modelSelection, runtimePolicy: resolvedRuntimePolicy, providerSessionId, - existingProviderThread: { ...providerThread, forkedFrom: null }, + existingProviderThread: nativeForkFailed + ? { ...providerThread, nativeThreadRef: null, forkedFrom: null } + : { ...providerThread, forkedFrom: null }, + // The failed fork may have left the adapter's session half-switched. + ...(nativeForkFailed ? { forceNewNativeThread: true } : {}), }), ); if (loaded === undefined) return undefined; @@ -1242,6 +1268,45 @@ export const layer: Layer.Layer< updatedAt: createdAt, }, }, + ...(nativeForkFailure === undefined + ? [] + : [ + { + id: yield* idAllocator.allocate.event({ threadId: projection.thread.id }), + type: "turn-item.updated" as const, + threadId: projection.thread.id, + runId: run.id, + providerInstanceId: run.providerInstanceId, + occurredAt: createdAt, + payload: { + id: idAllocator.derive.runSignalTurnItem({ + runId, + signal: "native-fork-fallback", + }), + threadId: projection.thread.id, + runId, + nodeId: rootNode.id, + providerThreadId: providerThread.id, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: + Math.max( + 0, + ...projection.turnItems + .filter((item) => item.runId === runId) + .map((item) => item.ordinal), + ) + 1, + status: "completed" as const, + title: "Native fork failed", + startedAt: createdAt, + completedAt: createdAt, + updatedAt: createdAt, + type: "system_notice" as const, + message: `Could not fork the source conversation natively (${nativeForkFailure}). Started a fresh session with its history.`, + }, + }, + ]), ], }); return loaded; diff --git a/apps/server/src/orchestration-v2/ThreadFork.execution.test.ts b/apps/server/src/orchestration-v2/ThreadFork.execution.test.ts index 41a1120620f86..fa6fcea911388 100644 --- a/apps/server/src/orchestration-v2/ThreadFork.execution.test.ts +++ b/apps/server/src/orchestration-v2/ThreadFork.execution.test.ts @@ -18,8 +18,10 @@ import { TurnItemId, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Statement from "effect/sql/Statement"; +import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; @@ -1311,3 +1313,244 @@ it.effect("merge-back rejects a second fork's context until the first is consume ); }).pipe(Effect.provide(forkCases[0]!.layer)), ); + +const runForkWithFailingNativeSource = (forkFailures: number) => { + const driver = ProviderDriverKind.make("claudeAgent"); + const instanceId = ProviderInstanceId.make("claudeAgent"); + const modelSelection = { instanceId, model: "test-model" }; + const started = Deferred.makeUnsafe(); + let forkCalls = 0; + 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; + return { + instanceId, + driver, + providerSessionId: input.providerSessionId, + providerSession: { + id: input.providerSessionId, + driver, + providerInstanceId: instanceId, + status: "ready", + cwd: "/tmp/fork-missing-native-source", + model: modelSelection.model, + capabilities: ClaudeProviderCapabilitiesV2, + createdAt: now, + updatedAt: now, + lastError: null, + }, + events: Stream.never, + ensureThread: ({ threadId, existingProviderThread }) => + Effect.succeed({ + ...existingProviderThread!, + appThreadId: threadId, + nativeThreadRef: { driver, nativeId: "native-fresh-fork", strength: "strong" }, + status: "idle", + updatedAt: now, + }), + resumeThread: ({ providerThread }) => Effect.succeed(providerThread), + startTurn: (turn) => Deferred.succeed(started, turn).pipe(Effect.asVoid), + steerTurn: () => Effect.void, + interruptTurn: () => Effect.void, + respondToRuntimeRequest: () => Effect.void, + readThreadSnapshot: () => Effect.die("unused"), + rollbackThread: () => Effect.die("unused"), + forkThread: ({ sourceProviderThread, targetThreadId }) => + ++forkCalls <= forkFailures + ? Effect.fail( + new ProviderAdapter.ProviderAdapterForkThreadError({ + driver, + providerThreadId: sourceProviderThread.id, + cause: + "Session not found: native-deleted-source; authorization=BearerSyntheticSecret", + }), + ) + : Effect.succeed({ + ...sourceProviderThread, + appThreadId: targetThreadId, + nativeThreadRef: { driver, nativeId: "native-forked", strength: "strong" }, + updatedAt: now, + }), + } satisfies ProviderAdapter.ProviderAdapterV2SessionRuntime; + }), + }; + return Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const now = yield* DateTime.now; + const sourceThreadId = ThreadId.make("fork-missing-source"); + const targetThreadId = ThreadId.make("fork-missing-target"); + const providerThreadId = ProviderThreadId.make("fork-missing-native-thread"); + const sourceRunId = RunId.make("fork-missing-source-run"); + const userMessageId = MessageId.make("fork-missing-source-message"); + + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("fork-missing-create-source"), + threadId: sourceThreadId, + projectId: ProjectId.make("fork-missing-project"), + title: "Fork missing source", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdBy: "user", + creationSource: "web", + }); + yield* eventSink.write({ + events: [ + { + id: EventId.make("fork-missing-provider-thread"), + type: "provider-thread.updated", + threadId: sourceThreadId, + occurredAt: now, + payload: { + id: providerThreadId, + driver, + providerInstanceId: instanceId, + providerSessionId: null, + appThreadId: sourceThreadId, + ownerNodeId: null, + nativeThreadRef: { driver, nativeId: "native-deleted-source", strength: "strong" }, + nativeConversationHeadRef: null, + status: "idle", + firstRunOrdinal: 1, + lastRunOrdinal: 1, + handoffIds: [], + forkedFrom: null, + createdAt: now, + updatedAt: now, + }, + }, + { + id: EventId.make("fork-missing-source-run"), + type: "run.created", + threadId: sourceThreadId, + runId: sourceRunId, + occurredAt: now, + payload: { + id: sourceRunId, + threadId: sourceThreadId, + ordinal: 1, + providerInstanceId: instanceId, + modelSelection, + providerThreadId, + userMessageId, + rootNodeId: null, + activeAttemptId: null, + status: "completed", + queuePosition: null, + requestedAt: now, + startedAt: now, + completedAt: now, + checkpointId: null, + contextHandoffId: null, + }, + }, + { + id: EventId.make("fork-missing-user-item"), + type: "turn-item.updated", + threadId: sourceThreadId, + runId: sourceRunId, + occurredAt: now, + payload: { + id: TurnItemId.make("fork-missing-user-item"), + threadId: sourceThreadId, + runId: sourceRunId, + nodeId: null, + providerThreadId, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "completed", + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + type: "user_message", + createdBy: "user", + creationSource: "web", + inputIntent: "turn_start", + messageId: userMessageId, + text: "FORKED_SOURCE_MARKER", + attachments: [], + }, + }, + ], + }); + yield* orchestrator.dispatch({ + type: "thread.fork", + commandId: CommandId.make("fork-missing-fork"), + sourceThreadId, + targetThreadId, + sourcePoint: { type: "run", runId: sourceRunId }, + createdBy: "user", + creationSource: "web", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("fork-missing-continue"), + threadId: targetThreadId, + messageId: MessageId.make("fork-missing-continue"), + text: "Continue the fork", + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + // Run every start attempt so the last one runs with willRetry false. + for (let attempt = 0; attempt < 5; attempt++) { + yield* worker.drain(); + yield* TestClock.adjust("1 second"); + } + + const projection = yield* orchestrator.getThreadProjection(targetThreadId); + assert.deepEqual( + projection.turnItems.flatMap((item) => (item.type === "error" ? [item.title] : [])), + [], + ); + assert.equal(projection.runs[0]?.status, "running"); + return { projection, turn: yield* Deferred.await(started), forkCalls }; + }).pipe( + Effect.provide( + ProviderReplayHarness.layerWithRegistry( + { name: "fork-missing-native-source" }, + ProviderAdapterRegistry.layerFromAdapters([adapter]), + ), + ), + ); +}; + +it.effect("a native fork whose source session is gone starts fresh with the source history", () => + Effect.gen(function* () { + const { projection, turn } = yield* runForkWithFailingNativeSource(Number.POSITIVE_INFINITY); + // Resolving the fork transfer stops later sends from retrying the native fork. + assert.equal(projection.contextTransfers[0]?.status, "resolved_portable"); + assert.isNull(projection.providerThreads[0]?.forkedFrom); + assert.include(JSON.stringify(turn), "FORKED_SOURCE_MARKER"); + const notices = projection.turnItems.flatMap((item) => + item.type === "system_notice" ? [item.message] : [], + ); + assert.lengthOf(notices, 1); + assert.include(notices[0], "Session not found: native-deleted-source"); + assert.notInclude(notices[0], "BearerSyntheticSecret"); + }), +); + +it.effect("a native fork that fails before its last attempt still forks natively", () => + Effect.gen(function* () { + const { projection, forkCalls } = yield* runForkWithFailingNativeSource(1); + assert.equal(forkCalls, 2); + assert.equal(projection.contextTransfers[0]?.status, "consumed"); + assert.equal(projection.contextTransfers[0]?.resolution?.strategy, "native_fork"); + }), +);