fix(server): a failed Antigravity checkpoint no longer bricks the thread

apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts

@@ -758,7 +758,10 @@ describe("AcpAdapterV2", () => {
             Effect.sync(() => {
               watchedSessions.push(sessionId);
               const failed = watchedSessions.length === 1;
-              return () => Effect.succeed(failed ? quota : undefined);
+              return () =>
+                Effect.succeed(
+                  failed ? { failure: quota, threadDisposition: "reusable" as const } : undefined,
+                );
             }),
           makeRuntime: makeMockRuntime({
             childProcessSpawner: yield* ChildProcessSpawner.ChildProcessSpawner,
@@ -824,6 +827,97 @@ describe("AcpAdapterV2", () => {
       assert.equal(watchedSessions[0], watchedSessions[1]);
     }).pipe(Effect.provide(layerTest), Effect.scoped),
   );
+  it.live("marks the thread broken when the outcome check reports unusable native history", () =>
+    Effect.gen(function* () {
+      const path = yield* Path.Path;
+      const instanceId = ProviderInstanceId.make("acp-test-prompt-outcome-broken");
+      const threadId = ThreadId.make("thread-acp-prompt-outcome-broken");
+      const failure = {
+        class: "provider_error",
+        message: "Antigravity stopped this turn because of an error.",
+        code: null,
+        retryable: null,
+      } as const;
+      const adapter = yield* makeAcpAdapterV2({
+        instanceId,
+        selfInvocation: yield* resolveSelfInvocation(),
+        flavor: {
+          driver: ACP_TEST_DRIVER,
+          capabilities: AcpProviderCapabilitiesV2,
+          watchPromptOutcome: () =>
+            Effect.succeed(() => Effect.succeed({ failure, threadDisposition: "broken" as const })),
+          makeRuntime: makeMockRuntime({
+            childProcessSpawner: yield* ChildProcessSpawner.ChildProcessSpawner,
+            mockAgentPath: yield* path.fromFileUrl(
+              new URL("../../../scripts/acp-mock-agent.ts", import.meta.url),
+            ),
+            wrapRuntime: (runtime) => ({
+              ...runtime,
+              prompt: () => Effect.succeed({ stopReason: "end_turn" as const }),
+            }),
+          }),
+        },
+      });
+      const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({
+        runtimeMode: "full-access",
+        interactionMode: "default",
+        cwd: process.cwd(),
+      });
+      const modelSelection = { instanceId, model: "default" } as const;
+      const runtime = yield* adapter.openSession({
+        threadId,
+        providerSessionId: ProviderSessionId.make("provider-session-acp-prompt-outcome-broken"),
+        modelSelection,
+        runtimePolicy,
+      });
+      const providerThread = yield* runtime.ensureThread({
+        threadId,
+        modelSelection,
+        runtimePolicy,
+      });
+      yield* runtime.startTurn(
+        makeTurnInput({
+          threadId,
+          providerThread,
+          instanceId,
+          runtimePolicy,
+          now: yield* DateTime.now,
+          ordinal: 1,
+        }),
+      );
+      const events = Array.from(
+        yield* runtime.events.pipe(
+          Stream.takeUntil((event) => event.type === "turn.terminal"),
+          Stream.runCollect,
+        ),
+      );
+
+      const terminal = events.at(-1);
+      if (terminal?.type !== "turn.terminal" || terminal.status !== "failed") {
+        return assert.fail("expected a failed terminal");
+      }
+      assert.equal(terminal.threadDisposition, "broken");
+      assert.deepEqual(terminal.failure, failure);
+      const updated = events.findLast((event) => event.type === "provider_thread.updated");
+      assert.equal(
+        updated?.type === "provider_thread.updated" ? updated.providerThread.status : null,
+        "error",
+      );
+      // The turn-start service recovers in a fresh native session when resume refuses.
+      const resumed = yield* Effect.result(
+        runtime.resumeThread({
+          threadId,
+          providerThread: { ...providerThread, status: "error" },
+          modelSelection,
+          runtimePolicy,
+        }),
+      );
+      assert.equal(
+        resumed._tag === "Failure" ? resumed.failure._tag : null,
+        "ProviderAdapterResumeThreadError",
+      );
+    }).pipe(Effect.provide(layerTest), Effect.scoped),
+  );
   it.live("retains returned usage when Stop finalizes before the prompt callback", () =>
     Effect.gen(function* () {
       const path = yield* Path.Path;

apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts

@@ -1430,6 +1430,92 @@ it.effect.each([false, true])("rejects thread.create for an existing id (deleted
   }).pipe(Effect.provide(layerTest)),
 );
 
+it.effect("keeps a broken native thread in error when a queued message starts", () =>
+  Effect.gen(function* () {
+    const orchestrator = yield* Orchestrator.OrchestratorV2;
+    const sink = yield* EventSink.EventSinkV2;
+    const threadId = ThreadId.make("thread:broken-native-thread");
+    const send = (id: string, dispatchMode: "start_immediately" | "queue_after_active") =>
+      orchestrator.dispatch({
+        type: "message.dispatch",
+        commandId: CommandId.make(`send:broken-native-thread:${id}`),
+        threadId,
+        messageId: MessageId.make(`message:broken-native-thread:${id}`),
+        text: id,
+        attachments: [],
+        dispatchMode: { type: dispatchMode },
+        createdBy: "user",
+        creationSource: "web",
+      });
+    yield* orchestrator.dispatch({
+      type: "thread.create",
+      commandId: CommandId.make("create:broken-native-thread"),
+      threadId,
+      projectId: ProjectId.make("project:broken-native-thread"),
+      title: "Broken",
+      modelSelection,
+      runtimeMode: "full-access",
+      interactionMode: "default",
+      branch: null,
+      worktreePath: null,
+      createdBy: "user",
+      creationSource: "web",
+    });
+    yield* send("first", "start_immediately");
+    yield* send("queued", "queue_after_active");
+    const { runs, providerThreads } = yield* orchestrator.getThreadRecords(threadId, [
+      "runs",
+      "providerThreads",
+    ]);
+    const [failed, queued] = runs;
+    const [providerThread] = providerThreads;
+    const occurredAt = yield* DateTime.now;
+    // The first turn ended with its native history unusable.
+    yield* sink.write({
+      events: [
+        {
+          id: EventId.make("event:broken-native-thread:run"),
+          type: "run.updated",
+          threadId,
+          runId: failed!.id,
+          providerInstanceId: instanceId,
+          occurredAt,
+          payload: { ...failed!, status: "failed", completedAt: occurredAt },
+        },
+        {
+          id: EventId.make("event:broken-native-thread:provider-thread"),
+          type: "provider-thread.updated",
+          threadId,
+          providerInstanceId: instanceId,
+          occurredAt,
+          payload: {
+            ...providerThread!,
+            nativeThreadRef: {
+              driver: providerThread!.driver,
+              nativeId: "poisoned-native-history",
+              strength: "strong",
+            },
+            status: "error",
+          },
+        },
+      ],
+    });
+
+    yield* orchestrator.dispatch({
+      type: "queue.resume",
+      commandId: CommandId.make("resume:broken-native-thread"),
+      threadId,
+    });
+    const next = yield* orchestrator.getThreadRecords(threadId, ["runs", "providerThreads"]);
+    const started = next.runs.find((run) => run.id === queued!.id);
+    assert.equal(started?.status, "starting");
+    assert.equal(
+      next.providerThreads.find((thread) => thread.id === started?.providerThreadId)?.status,
+      "error",
+    );
+  }).pipe(Effect.provide(layerTest)),
+);
+
 it.effect("replays a thread.create receipt without changing the thread", () =>
   Effect.gen(function* () {
     const orchestrator = yield* Orchestrator.OrchestratorV2;

apps/server/src/orchestration-v2/Orchestrator.ts

@@ -1838,6 +1838,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
             nativeThreadRef: activeProviderThread.nativeThreadRef,
             nativeConversationHeadRef: activeProviderThread.nativeConversationHeadRef,
             nativeMetadata: activeProviderThread.nativeMetadata,
+            status: activeProviderThread.status,
           }
         : queuedProviderThread;
       const targetCapabilities = yield* targetAdapter.getCapabilities().pipe(
@@ -2021,7 +2022,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       const providerThread: OrchestrationV2ProviderThread = {
         ...deliveryProviderThread,
         providerSessionId,
-        status: "not_loaded",
+        // A broken native history must reach the turn start, which replaces it.
+        status: deliveryProviderThread.status === "error" ? "error" : "not_loaded",
         firstRunOrdinal: queuedProviderThread.firstRunOrdinal ?? ordinal,
         lastRunOrdinal: ordinal,
         handoffIds: appendContextHandoffId(
@@ -5168,7 +5170,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
         ...providerThread,
         providerInstanceId: input.modelSelection.instanceId,
         providerSessionId: nextProviderSessionId,
-        status: "not_loaded",
+        status: providerThread.status === "error" ? "error" : "not_loaded",
         updatedAt: now,
       };
       restartSessionTransition = {
@@ -7029,7 +7031,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
         : null;
     const providerThread: OrchestrationV2ProviderThread = {
       ...ensuredProviderThread,
-      status: "active",
+      status: ensuredProviderThread.status === "error" ? "error" : "active",
       firstRunOrdinal: ensuredProviderThread.firstRunOrdinal ?? ordinal,
       lastRunOrdinal: ordinal,
       handoffIds: [

apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts

@@ -1747,11 +1747,32 @@ effectIt.effect("does not mistake a failed state read for a superseded run", ()
   }),
 );
 
-effectIt.effect(
-  "signs out the existing native provider before opening the newly selected provider",
-  () =>
+effectIt.effect.each(["not_loaded", "error"] as const)(
+  "signs out the existing native provider before opening the newly selected provider with status %s",
+  (providerThreadStatus) =>
     Effect.gen(function* () {
       const harness = makeLocalCommandHarness({ text: "/logout", previousNativeSession: true });
+      const initial = harness.projection();
+      const target = initial.providerThreads.find((thread) => thread.id === "new-provider-thread")!;
+      harness.setProjection({
+        ...initial,
+        providerThreads: initial.providerThreads.map((thread) =>
+          thread.id === target.id
+            ? {
+                ...thread,
+                status: providerThreadStatus,
+                nativeThreadRef:
+                  providerThreadStatus === "error"
+                    ? {
+                        driver: thread.driver,
+                        nativeId: "poisoned-native-history",
+                        strength: "strong" as const,
+                      }
+                    : thread.nativeThreadRef,
+              }
+            : thread,
+        ),
+      });
 
       yield* harness.start;
       yield* harness.start;
@@ -1777,6 +1798,9 @@ effectIt.effect(
       ]);
       expect(projection.providerTurns).toEqual([]);
       expect(projection.checkpoints).toEqual([]);
+      expect(projection.providerThreads.find((thread) => thread.id === target.id)?.status).toBe(
+        providerThreadStatus,
+      );
     }),
 );
 

apps/server/src/orchestration-v2/ProviderTurnStartService.ts

@@ -627,7 +627,12 @@ export const layer: Layer.Layer<
                   },
             providerThreadUpdate: {
               ...providerThread,
-              status: providerThread.nativeThreadRef === null ? "not_loaded" : "idle",
+              status:
+                providerThread.status === "error"
+                  ? "error"
+                  : providerThread.nativeThreadRef === null
+                    ? "not_loaded"
+                    : "idle",
               updatedAt: now,
             },
           });