fix(server): forks whose source session is gone start fresh instead of failing every send

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;

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<ProviderAdapter.ProviderAdapterV2TurnInput>();
+  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");
+  }),
+);