Sync coder-only with upstream main and complete Coder adaptations

apps/server/src/checkpointing/CheckpointDiffQuery.test.ts

@@ -77,6 +77,7 @@ describe("CheckpointDiffQuery.layer", () => {
           Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
             getCommandReadModel: () =>
               Effect.die("CheckpointDiffQuery should not request the command read model"),
+            getUserInputActivity: () => Effect.die("unused"),
             getSnapshot: () =>
               Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
             getShellSnapshot: () =>
@@ -186,6 +187,7 @@ describe("CheckpointDiffQuery.layer", () => {
           Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
             getCommandReadModel: () =>
               Effect.die("CheckpointDiffQuery should not request the command read model"),
+            getUserInputActivity: () => Effect.die("unused"),
             getSnapshot: () =>
               Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
             getShellSnapshot: () =>
@@ -270,6 +272,7 @@ describe("CheckpointDiffQuery.layer", () => {
           Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
             getCommandReadModel: () =>
               Effect.die("CheckpointDiffQuery should not request the command read model"),
+            getUserInputActivity: () => Effect.die("unused"),
             getSnapshot: () =>
               Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
             getShellSnapshot: () =>
@@ -339,6 +342,7 @@ describe("CheckpointDiffQuery.layer", () => {
           Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
             getCommandReadModel: () =>
               Effect.die("CheckpointDiffQuery should not request the command read model"),
+            getUserInputActivity: () => Effect.die("unused"),
             getSnapshot: () =>
               Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
             getShellSnapshot: () =>
@@ -393,6 +397,7 @@ describe("CheckpointDiffQuery.layer", () => {
           Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
             getCommandReadModel: () =>
               Effect.die("CheckpointDiffQuery should not request the command read model"),
+            getUserInputActivity: () => Effect.die("unused"),
             getSnapshot: () =>
               Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
             getShellSnapshot: () =>

apps/server/src/checkpointing/CheckpointStore.test.ts

@@ -147,6 +147,37 @@ it.layer(TestLayer)("CheckpointStore.layer", (it) => {
       }),
     );
 
+    it.effect("keeps a/ and b/ patch prefixes when the repository disables them", () =>
+      Effect.gen(function* () {
+        const tmp = yield* makeTmpDir();
+        yield* initRepoWithCommit(tmp);
+        yield* git(tmp, ["config", "diff.noprefix", "true"]);
+        const checkpointStore = yield* CheckpointStore.CheckpointStore;
+        const threadId = ThreadId.make("thread-checkpoint-store-noprefix");
+        const fromCheckpointRef = checkpointRefForThreadTurn(threadId, 0);
+        const toCheckpointRef = checkpointRefForThreadTurn(threadId, 1);
+
+        yield* checkpointStore.captureCheckpoint({
+          cwd: tmp,
+          checkpointRef: fromCheckpointRef,
+        });
+        yield* writeTextFile(NodePath.join(tmp, "README.md"), "# changed\n");
+        yield* checkpointStore.captureCheckpoint({
+          cwd: tmp,
+          checkpointRef: toCheckpointRef,
+        });
+
+        const diff = yield* checkpointStore.diffCheckpoints({
+          cwd: tmp,
+          fromCheckpointRef,
+          toCheckpointRef,
+          ignoreWhitespace: false,
+        });
+
+        expect(diff).toContain("diff --git a/README.md b/README.md");
+      }),
+    );
+
     it.effect("can hide indentation churn when changes wrap existing lines", () =>
       Effect.gen(function* () {
         const tmp = yield* makeTmpDir();

apps/server/src/coderWs.ts

@@ -1019,6 +1019,7 @@ export const layer = CoderWsRpcGroup.toLayer(
         pullRequests.setThreadResolution(input),
       [WS_METHODS.pullRequestsSetReaction]: (input) => pullRequests.setReaction(input),
       [WS_METHODS.pullRequestsInvalidate]: (input) => pullRequests.invalidate(input),
+      [WS_METHODS.pullRequestsSubscribeRefreshes]: () => pullRequests.subscribeRefreshes,
       [WS_METHODS.pullRequestsReviewerCandidates]: (input) =>
         pullRequests.reviewerCandidates(input),
       [WS_METHODS.pullRequestsRequestReviewers]: (input) => pullRequests.requestReviewers(input),

apps/server/src/git/GitWorkflowService.ts

@@ -194,7 +194,9 @@ export const layer = Layer.effect(
     const STATUS_CACHE_TTL = Duration.seconds(1);
     const STATUS_CACHE_CAPACITY = 2_048;
     const FETCH_CACHE_TTL = Duration.seconds(1);
-    const PR_CACHE_TTL = Duration.minutes(2);
+    // Match the automatic settlement sweep cadence so an external merge is
+    // observed on the next sweep instead of waiting on an older cache entry.
+    const PR_CACHE_TTL = Duration.seconds(60);
     const FAILURE_BASE_TTL = Duration.seconds(20);
     const FAILURE_MAX_TTL = Duration.minutes(15);
     const failureTtl = (failures: number) =>

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

@@ -59,6 +59,7 @@ import { checkpointRefForThreadTurn } from "../../checkpointing/Utils.ts";
 import { ServerConfig } from "../../config.ts";
 import * as WorkspaceEntries from "../../workspace/WorkspaceEntries.ts";
 import * as WorkspacePaths from "../../workspace/WorkspacePaths.ts";
+import * as PullRequestService from "../../pullRequest/PullRequestService.ts";
 
 const asProjectId = (value: string): ProjectId => ProjectId.make(value);
 const asTurnId = (value: string): TurnId => TurnId.make(value);
@@ -107,6 +108,7 @@ function createProviderServiceHarness(
   const service: ProviderServiceShape = {
     startSession: () => unsupported(),
     sendTurn: () => unsupported(),
+    compactThread: () => unsupported(),
     interruptTurn: () => unsupported(),
     respondToRequest: () => unsupported(),
     respondToUserInput: () => unsupported(),
@@ -314,6 +316,8 @@ describe("CheckpointReactor", () => {
     const ServerConfigLayer = ServerConfig.layerTest(process.cwd(), {
       prefix: "t3-checkpoint-reactor-test-",
     });
+    const pullRequestRefreshes: number[] = [];
+    const refreshAfterTurn = Effect.sync(() => void pullRequestRefreshes.push(1));
     const vcsStatusLayer = Layer.succeed(CoderVcsStatus, {
       refresh: (cwd: string) =>
         Effect.sync(() => {
@@ -343,6 +347,7 @@ describe("CheckpointReactor", () => {
       Layer.provideMerge(projectionSnapshotLayer),
       Layer.provideMerge(RuntimeReceiptBusLive),
       Layer.provideMerge(Layer.succeed(ProviderService, provider.service)),
+      Layer.provideMerge(Layer.mock(PullRequestService.PullRequestService)({ refreshAfterTurn })),
       Layer.provideMerge(vcsStatusLayer),
       Layer.provideMerge(CheckpointStore.layer.pipe(Layer.provide(VcsDriverRegistry.layer))),
       Layer.provideMerge(
@@ -454,6 +459,7 @@ describe("CheckpointReactor", () => {
       provider,
       cwd,
       drain,
+      pullRequestRefreshes,
     };
   }
 
@@ -773,6 +779,7 @@ describe("CheckpointReactor", () => {
     const midReadModel = await harness.readModel();
     const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.make("thread-1"));
     expect(midThread?.checkpoints).toHaveLength(0);
+    expect(harness.pullRequestRefreshes).toEqual([]);
 
     harness.provider.emit({
       type: "turn.completed",
@@ -790,6 +797,8 @@ describe("CheckpointReactor", () => {
       (entry) => entry.latestTurn?.turnId === "turn-main" && entry.checkpoints.length === 1,
     );
     expect(thread.checkpoints[0]?.checkpointTurnCount).toBe(1);
+    await harness.drain();
+    expect(harness.pullRequestRefreshes).toEqual([1]);
   });
 
   it("captures pre-turn and completion checkpoints for claude runtime events", async () => {

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

@@ -38,6 +38,7 @@ import type { OrchestrationDispatchError } from "../Errors.ts";
 import { isGitRepository } from "../../git/Utils.ts";
 import { CoderVcsStatus } from "../../coderVcsStatus.ts";
 import * as WorkspaceEntries from "../../workspace/WorkspaceEntries.ts";
+import * as PullRequestService from "../../pullRequest/PullRequestService.ts";
 
 const nowIso = Effect.map(DateTime.now, DateTime.formatIso);
 
@@ -96,6 +97,9 @@ const make = Effect.gen(function* () {
   const receiptBus = yield* RuntimeReceiptBus;
   const workspaceEntries = yield* WorkspaceEntries.WorkspaceEntries;
   const vcsStatus = yield* CoderVcsStatus;
+  const pullRequests = yield* PullRequestService.PullRequestService;
+  const startedTurns = new Map<ThreadId, TurnId>();
+  const pendingTurns = new Set<ThreadId>();
 
   const appendRevertFailureActivity = (input: {
     readonly threadId: ThreadId;
@@ -881,6 +885,9 @@ const make = Effect.gen(function* () {
 
   const processDomainEvent = Effect.fn("processDomainEvent")(function* (event: OrchestrationEvent) {
     if (event.type === "thread.turn-start-requested" || event.type === "thread.message-sent") {
+      if (event.type === "thread.turn-start-requested") {
+        pendingTurns.add(event.payload.threadId);
+      }
       yield* ensurePreTurnBaselineFromDomainTurnStart(event);
       return;
     }
@@ -925,7 +932,22 @@ const make = Effect.gen(function* () {
   const processRuntimeEvent = Effect.fn("processRuntimeEvent")(function* (
     event: ProviderRuntimeEvent,
   ) {
+    if (event.type === "session.exited") {
+      startedTurns.delete(event.threadId);
+      pendingTurns.delete(event.threadId);
+      return;
+    }
+
     if (event.type === "turn.started") {
+      const turnId = toTurnId(event.turnId);
+      const activeTurnId = (yield* providerService.listSessions()).find((session) =>
+        sameId(session.threadId, event.threadId),
+      )?.activeTurnId;
+      const mayReplace = pendingTurns.has(event.threadId) && sameId(activeTurnId, turnId);
+      if (turnId !== null && (!startedTurns.has(event.threadId) || mayReplace)) {
+        startedTurns.set(event.threadId, turnId);
+        pendingTurns.delete(event.threadId);
+      }
       yield* ensurePreTurnBaselineFromTurnStart(event);
       return;
     }
@@ -947,9 +969,30 @@ const make = Effect.gen(function* () {
       return;
     }
 
-    if (event.type === "turn.completed") {
+    if (event.type === "turn.completed" || event.type === "turn.aborted") {
       const turnId = toTurnId(event.turnId);
-      yield* refreshLocalGitStatusAfterAgentCommand(event);
+      const thread = yield* resolveThreadDetail(event.threadId);
+      const startedTurnId = startedTurns.get(event.threadId);
+      const isTrackedTurn = sameId(startedTurnId, turnId);
+      if (isTrackedTurn) {
+        startedTurns.delete(event.threadId);
+      }
+      if (event.type === "turn.completed") {
+        yield* refreshLocalGitStatusAfterAgentCommand(event);
+      }
+      if (
+        turnId !== null &&
+        thread !== undefined &&
+        (isTrackedTurn ||
+          sameId(thread.session?.activeTurnId, turnId) ||
+          (startedTurnId === undefined && !thread.session?.activeTurnId))
+      ) {
+        pendingTurns.delete(event.threadId);
+        yield* pullRequests.refreshAfterTurn;
+      }
+      if (event.type === "turn.aborted") {
+        return;
+      }
       yield* captureCheckpointFromTurnCompletion(event).pipe(
         Effect.catch((error) =>
           Effect.flatMap(nowIso, (createdAt) =>
@@ -1012,6 +1055,8 @@ const make = Effect.gen(function* () {
           event.type !== "turn.started" &&
           event.type !== "session.configured" &&
           event.type !== "turn.completed" &&
+          event.type !== "turn.aborted" &&
+          event.type !== "session.exited" &&
           !(event.type === "item.completed" && event.payload.itemType === "command_execution")
         ) {
           return Effect.void;

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

@@ -173,6 +173,7 @@ describe("OrchestrationEngine", () => {
       Layer.provide(
         Layer.succeed(ProjectionSnapshotQuery, {
           getCommandReadModel: () => Effect.succeed(commandReadModel),
+          getUserInputActivity: () => Effect.die("unused"),
           getSnapshot: () =>
             Effect.sync(() => {
               fullSnapshotReadCount += 1;

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

@@ -180,9 +180,19 @@ const makeOrchestrationEngine = Effect.gen(function* () {
           });
         }
 
+        // Command snapshots intentionally omit activity bodies. Read this
+        // request's durable state before deciding whether an async answer is
+        // a provider response or a new user turn.
+        const userInputActivity =
+          envelope.command.type === "thread.user-input.respond"
+            ? yield* projectionSnapshotQuery.getUserInputActivity(envelope.command)
+            : Option.none();
         const eventBase = yield* decideOrchestrationCommand({
           command: envelope.command,
           readModel: commandReadModel,
+          ...(Option.isSome(userInputActivity)
+            ? { userInputActivity: userInputActivity.value }
+            : {}),
         }).pipe(
           Effect.provideService(Crypto.Crypto, crypto),
           Effect.mapError((cause) =>

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

@@ -435,26 +435,13 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
         return;
       }
 
-      const [messages, proposedPlans, activities, pendingApprovals] = yield* Effect.all([
-        projectionThreadMessageRepository.listByThreadId({ threadId }),
-        projectionThreadProposedPlanRepository.listByThreadId({ threadId }),
-        projectionThreadActivityRepository.listByThreadId({ threadId }),
-        projectionPendingApprovalRepository.listByThreadId({ threadId }),
-      ]);
-
-      let latestUserMessageAt: string | null = null;
-      for (const message of messages) {
-        if (
-          message.role === "user" &&
-          (latestUserMessageAt === null || message.createdAt > latestUserMessageAt)
-        ) {
-          latestUserMessageAt = message.createdAt;
-        }
-      }
-
-      const pendingApprovalCount = pendingApprovals.filter(
-        (approval) => approval.status === "pending",
-      ).length;
+      const [latestUserMessageAt, proposedPlans, activities, pendingApprovalCount] =
+        yield* Effect.all([
+          projectionThreadMessageRepository.getLatestUserMessageAt({ threadId }),
+          projectionThreadProposedPlanRepository.listByThreadId({ threadId }),
+          projectionThreadActivityRepository.listByThreadId({ threadId }),
+          projectionPendingApprovalRepository.countPendingByThreadId({ threadId }),
+        ]);
       const pendingUserInputCount = derivePendingUserInputCountFromActivities(activities);
       const hasActionableProposedPlan = deriveHasActionableProposedPlan({
         latestTurnId: existingRow.value.latestTurnId,
@@ -1059,6 +1046,21 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
           return;
 
         case "thread.turn-start-requested": {
+          const pendingTurnStart = yield* projectionTurnRepository.getPendingTurnStartByThreadId({
+            threadId: event.payload.threadId,
+          });
+          if (Option.isSome(pendingTurnStart)) {
+            const pendingMessage = yield* projectionThreadMessageRepository.getByMessageId({
+              messageId: pendingTurnStart.value.messageId,
+            });
+            if (
+              Option.isSome(pendingMessage) &&
+              pendingMessage.value.role === "user" &&
+              pendingMessage.value.text.trim().toLowerCase() === "/compact"
+            ) {
+              return;
+            }
+          }
           yield* projectionTurnRepository.replacePendingTurnStart({
             threadId: event.payload.threadId,
             messageId: event.payload.messageId,
@@ -1069,10 +1071,33 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
           return;
         }
 
+        case "thread.activity-appended": {
+          if (
+            event.payload.activity.kind !== "context-compaction" &&
+            event.payload.activity.kind !== "provider.turn.start.failed"
+          ) {
+            return;
+          }
+          const pendingTurnStart = yield* projectionTurnRepository.getPendingTurnStartByThreadId(
+            event.payload,
+          );
+          if (
+            Option.isNone(pendingTurnStart) ||
+            String(pendingTurnStart.value.messageId) !==
+              extractActivityRequestId(event.payload.activity.payload)
+          ) {
+            return;
+          }
+          yield* projectionTurnRepository.deletePendingTurnStartByThreadId(event.payload);
+          return;
+        }
+
         case "thread.session-set": {
           const turnId = event.payload.session.activeTurnId;
           if (turnId === null || event.payload.session.status !== "running") {
             if (
+              (event.payload.session.status === "ready" &&
+                event.commandId?.startsWith("server:provider-session-set:") === true) ||
               event.payload.session.status === "error" ||
               event.payload.session.status === "stopped" ||
               event.payload.session.status === "interrupted"