Improve loading skeletons and preserve active turn status

apps/server/src/cli/marketingStudioSeed.ts

@@ -249,6 +249,7 @@ const seedThreads = (input: MarketingStudioSeedInput) =>
               threadId,
               messageId: assistantMessageId,
               turnId,
+              completesTurn: true,
               createdAt: completedAt,
             });
           }
@@ -422,6 +423,7 @@ const seedThreads = (input: MarketingStudioSeedInput) =>
           ],
           ...(scenario.assistantText !== undefined ? { assistantMessageId } : {}),
           checkpointTurnCount: 1,
+          completesTurn: true,
           createdAt: completedAt,
         });
 

apps/server/src/orchestration/Layers/CheckpointReactor.test.ts

@@ -612,6 +612,67 @@ describe("CheckpointReactor", () => {
     ).toBe("v2\n");
   });
 
+  it("captures but does not finalize non-terminal provider diff placeholders", async () => {
+    const harness = await createHarness({ seedFilesystemCheckpoints: false });
+    const threadId = ThreadId.make("thread-1");
+    const turnId = asTurnId("turn-mid-diff-placeholder");
+    const createdAt = "2026-01-01T00:00:05.000Z";
+
+    await Effect.runPromise(
+      harness.engine.dispatch({
+        type: "thread.session.set",
+        commandId: CommandId.make("cmd-session-set-mid-diff-placeholder"),
+        threadId,
+        session: {
+          threadId,
+          status: "running",
+          providerName: "codex",
+          runtimeMode: "approval-required",
+          activeTurnId: turnId,
+          lastError: null,
+          updatedAt: createdAt,
+        },
+        createdAt,
+      }),
+    );
+    fs.writeFileSync(path.join(harness.cwd, "README.md"), "v2\n", "utf8");
+
+    await Effect.runPromise(
+      harness.engine.dispatch({
+        type: "thread.turn.diff.complete",
+        commandId: CommandId.make("cmd-mid-diff-placeholder"),
+        threadId,
+        turnId,
+        completedAt: createdAt,
+        checkpointRef: asCheckpointRef("provider-diff:evt-mid-diff-placeholder"),
+        status: "missing",
+        files: [{ path: "README.md", kind: "modified", additions: 1, deletions: 1 }],
+        assistantMessageId: MessageId.make("assistant:mid-diff-placeholder"),
+        checkpointTurnCount: 1,
+        completesTurn: false,
+        createdAt,
+      }),
+    );
+    await harness.drain();
+
+    await waitForGitRefExists(harness.cwd, checkpointRefForThreadTurn(threadId, 1));
+    const snapshot = await harness.readModel();
+    const thread = snapshot.threads.find((entry) => entry.id === threadId);
+    expect(thread?.latestTurn).toMatchObject({
+      turnId,
+      state: "running",
+      completedAt: null,
+    });
+    expect(thread?.checkpoints).toEqual([
+      expect.objectContaining({
+        turnId,
+        checkpointRef: checkpointRefForThreadTurn(threadId, 1),
+        status: "ready",
+        files: [{ path: "README.md", kind: "modified", additions: 1, deletions: 1 }],
+      }),
+    ]);
+  });
+
   it("skips changed-file summaries from a shared checkout while another session is active", async () => {
     const harness = await createHarness({
       seedFilesystemCheckpoints: false,
@@ -720,6 +781,7 @@ describe("CheckpointReactor", () => {
         status: "ready",
         files: [{ path: "EXTERNAL.md", kind: "modified", additions: 1, deletions: 0 }],
         checkpointTurnCount: 1,
+        completesTurn: true,
         createdAt: "2026-01-01T00:02:00.000Z",
       }),
     );
@@ -795,6 +857,7 @@ describe("CheckpointReactor", () => {
         status: "missing",
         files: providerFiles,
         checkpointTurnCount: 1,
+        completesTurn: false,
         createdAt,
       }),
     );
@@ -804,7 +867,6 @@ describe("CheckpointReactor", () => {
     const thread = await waitForThread(
       harness.readModel,
       (entry) =>
-        entry.latestTurn?.turnId === "turn-shared-provider-summary" &&
         entry.checkpoints.length === 1 &&
         entry.checkpoints[0]?.status === "ready" &&
         entry.activities.some(
@@ -885,16 +947,14 @@ describe("CheckpointReactor", () => {
         status: "missing",
         files: providerFiles,
         checkpointTurnCount: 1,
+        completesTurn: false,
         createdAt,
       }),
     );
 
     await waitForThread(
       harness.readModel,
-      (entry) =>
-        entry.latestTurn?.turnId === "turn-shared-refresh-same-path-summary" &&
-        entry.checkpoints.length === 1 &&
-        entry.checkpoints[0]?.status === "ready",
+      (entry) => entry.checkpoints.length === 1 && entry.checkpoints[0]?.status === "ready",
     );
 
     fs.writeFileSync(path.join(harness.cwd, "README.md"), "final\nmore\n", "utf8");
@@ -960,16 +1020,14 @@ describe("CheckpointReactor", () => {
         status: "missing",
         files: [],
         checkpointTurnCount: 1,
+        completesTurn: false,
         createdAt,
       }),
     );
 
     await waitForThread(
       harness.readModel,
-      (entry) =>
-        entry.latestTurn?.turnId === "turn-empty-provider-summary" &&
-        entry.checkpoints.length === 1 &&
-        entry.checkpoints[0]?.status === "ready",
+      (entry) => entry.checkpoints.length === 1 && entry.checkpoints[0]?.status === "ready",
     );
 
     harness.provider.emit({
@@ -1103,6 +1161,7 @@ describe("CheckpointReactor", () => {
         status: "missing",
         files: [],
         checkpointTurnCount: 1,
+        completesTurn: false,
         createdAt,
       }),
     );
@@ -1173,16 +1232,14 @@ describe("CheckpointReactor", () => {
         status: "missing",
         files: providerFiles,
         checkpointTurnCount: 1,
+        completesTurn: false,
         createdAt,
       }),
     );
 
     await waitForThread(
       harness.readModel,
-      (entry) =>
-        entry.latestTurn?.turnId === "turn-refresh-summary" &&
-        entry.checkpoints.length === 1 &&
-        entry.checkpoints[0]?.status === "ready",
+      (entry) => entry.checkpoints.length === 1 && entry.checkpoints[0]?.status === "ready",
     );
 
     fs.writeFileSync(path.join(harness.cwd, "README.md"), "final\n", "utf8");
@@ -1663,6 +1720,7 @@ describe("CheckpointReactor", () => {
         status: "ready",
         files: [],
         checkpointTurnCount: 1,
+        completesTurn: true,
         createdAt,
       }),
     );
@@ -1677,6 +1735,7 @@ describe("CheckpointReactor", () => {
         status: "ready",
         files: [],
         checkpointTurnCount: 2,
+        completesTurn: true,
         createdAt,
       }),
     );
@@ -1784,6 +1843,7 @@ describe("CheckpointReactor", () => {
         status: "ready",
         files: [],
         checkpointTurnCount: 1,
+        completesTurn: true,
         createdAt,
       }),
     );
@@ -1798,6 +1858,7 @@ describe("CheckpointReactor", () => {
         status: "ready",
         files: [],
         checkpointTurnCount: 2,
+        completesTurn: true,
         createdAt,
       }),
     );
@@ -1854,6 +1915,7 @@ describe("CheckpointReactor", () => {
         status: "ready",
         files: [],
         checkpointTurnCount: 1,
+        completesTurn: true,
         createdAt,
       }),
     );
@@ -1868,6 +1930,7 @@ describe("CheckpointReactor", () => {
         status: "ready",
         files: [],
         checkpointTurnCount: 2,
+        completesTurn: true,
         createdAt,
       }),
     );
@@ -1962,6 +2025,7 @@ describe("CheckpointReactor", () => {
         status: "ready",
         files: [{ path: "README.md", kind: "modified", additions: 1, deletions: 1 }],
         checkpointTurnCount: 1,
+        completesTurn: true,
         createdAt,
       }),
     );
@@ -1979,6 +2043,7 @@ describe("CheckpointReactor", () => {
           { path: "created-by-thread.txt", kind: "added", additions: 1, deletions: 0 },
         ],
         checkpointTurnCount: 2,
+        completesTurn: true,
         createdAt,
       }),
     );
@@ -2077,6 +2142,7 @@ describe("CheckpointReactor", () => {
         status: "ready",
         files: [{ path: "README.md", kind: "modified", additions: 1, deletions: 1 }],
         checkpointTurnCount: 1,
+        completesTurn: true,
         createdAt,
       }),
     );
@@ -2094,6 +2160,7 @@ describe("CheckpointReactor", () => {
           { path: "created-by-thread.txt", kind: "added", additions: 1, deletions: 0 },
         ],
         checkpointTurnCount: 2,
+        completesTurn: true,
         createdAt,
       }),
     );
@@ -2190,6 +2257,7 @@ describe("CheckpointReactor", () => {
         status: "ready",
         files: [{ path: "lines.txt", kind: "modified", additions: 1, deletions: 1 }],
         checkpointTurnCount: 1,
+        completesTurn: true,
         createdAt,
       }),
     );
@@ -2279,6 +2347,7 @@ describe("CheckpointReactor", () => {
         status: "ready",
         files: [{ path: "notes.md", kind: "modified", additions: 4, deletions: 0 }],
         checkpointTurnCount: 1,
+        completesTurn: true,
         createdAt,
       }),
     );
@@ -2389,6 +2458,7 @@ describe("CheckpointReactor", () => {
         status: "ready",
         files: [{ path: "README.md", kind: "modified", additions: 1, deletions: 1 }],
         checkpointTurnCount: 1,
+        completesTurn: true,
         createdAt,
       }),
     );
@@ -2406,6 +2476,7 @@ describe("CheckpointReactor", () => {
           { path: "shared.txt", kind: "modified", additions: 1, deletions: 1 },
         ],
         checkpointTurnCount: 2,
+        completesTurn: true,
         createdAt,
       }),
     );
@@ -2420,6 +2491,7 @@ describe("CheckpointReactor", () => {
         status: "ready",
         files: [{ path: "shared.txt", kind: "modified", additions: 1, deletions: 0 }],
         checkpointTurnCount: 3,
+        completesTurn: true,
         createdAt,
       }),
     );
@@ -2527,6 +2599,7 @@ describe("CheckpointReactor", () => {
           status: "ready",
           files: [...files],
           checkpointTurnCount: turnCount,
+          completesTurn: true,
           createdAt,
         }),
       );
@@ -2571,6 +2644,7 @@ describe("CheckpointReactor", () => {
           status: "ready",
           files: [{ path: "README.md", kind: "modified", additions: 1, deletions: 1 }],
           checkpointTurnCount: turnCount,
+          completesTurn: true,
           createdAt,
         }),
       );

apps/server/src/orchestration/Layers/CheckpointReactor.ts

@@ -445,6 +445,7 @@ const make = Effect.gen(function* () {
     readonly assistantMessageId: MessageId | undefined;
     readonly providerSummaryFiles: ReadonlyArray<OrchestrationCheckpointFile> | undefined;
     readonly refreshSharedCheckoutSummaryFromCheckpoint: boolean;
+    readonly completesTurn: boolean;
     /** When the turn's diff window opened (turn start). Undefined falls back
      * to the instantaneous live-session check alone. */
     readonly turnWindowStartIso: string | undefined;
@@ -601,6 +602,7 @@ const make = Effect.gen(function* () {
       files,
       assistantMessageId,
       checkpointTurnCount: input.turnCount,
+      completesTurn: input.completesTurn,
       createdAt: input.createdAt,
     });
     yield* appendCheckpointFileChangeActivity({
@@ -619,13 +621,15 @@ const make = Effect.gen(function* () {
       status: input.status,
       createdAt: input.createdAt,
     });
-    yield* receiptBus.publish({
-      type: "turn.processing.quiesced",
-      threadId: input.threadId,
-      turnId: input.turnId,
-      checkpointTurnCount: input.turnCount,
-      createdAt: input.createdAt,
-    });
+    if (input.completesTurn) {
+      yield* receiptBus.publish({
+        type: "turn.processing.quiesced",
+        threadId: input.threadId,
+        turnId: input.turnId,
+        checkpointTurnCount: input.turnCount,
+        createdAt: input.createdAt,
+      });
+    }
 
     yield* orchestrationEngine.dispatch({
       type: "thread.activity.append",
@@ -702,6 +706,7 @@ const make = Effect.gen(function* () {
         assistantMessageId: undefined,
         providerSummaryFiles,
         refreshSharedCheckoutSummaryFromCheckpoint: true,
+        completesTurn: true,
         turnWindowStartIso: turnWindowStartIsoForThread(thread, turnId),
         createdAt: event.createdAt,
       });
@@ -734,6 +739,11 @@ const make = Effect.gen(function* () {
       return;
     }
 
+    const activeTurnId = thread.session?.activeTurnId ?? null;
+    const completesTurn =
+      event.payload.completesTurn !== undefined
+        ? event.payload.completesTurn
+        : activeTurnId === null || !sameId(activeTurnId, turnId);
     // If a real checkpoint already exists for this turn, skip.
     if (
       thread.checkpoints.some(
@@ -768,6 +778,7 @@ const make = Effect.gen(function* () {
       assistantMessageId: event.payload.assistantMessageId ?? undefined,
       providerSummaryFiles: event.payload.files,
       refreshSharedCheckoutSummaryFromCheckpoint: false,
+      completesTurn,
       turnWindowStartIso: turnWindowStartIsoForThread(thread, turnId),
       createdAt: event.payload.completedAt,
     });
@@ -1071,11 +1082,9 @@ const make = Effect.gen(function* () {
       return;
     }
 
-    // When ProviderRuntimeIngestion creates a placeholder checkpoint (status "missing")
-    // from a turn.diff.updated runtime event, capture the real git checkpoint to
-    // replace it. The providerService.streamEvents PubSub does not reliably deliver
-    // turn.completed runtime events to this reactor (shared subscription), so
-    // reacting to the domain event is the reliable path.
+    // Provider diff notifications create a non-terminal placeholder. Capture a
+    // real checkpoint for diff/revert fidelity, but propagate completesTurn so
+    // that capture cannot settle the still-running provider turn.
     if (event.type === "thread.turn-diff-completed") {
       yield* captureCheckpointFromPlaceholder(event).pipe(
         Effect.catch((error) =>

apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts

@@ -824,6 +824,7 @@ describe("OrchestrationEngine", () => {
         status: "ready",
         files: [],
         checkpointTurnCount: 1,
+        completesTurn: true,
         createdAt,
       }),
     );