Fix Cursor ACP session replay on resume

Author: tarik02

FORK.md

@@ -29,6 +29,7 @@ This repository is a fork of `pingdotgg/t3code`. Keep this file focused on fork
 ### Provider Launch Environment
 
 - Provider sessions use a shared launch environment pipeline instead of ad hoc environment assembly.
+- Cursor ACP load replay is reconciled against already-projected assistant text so restarts do not duplicate seen output while still allowing unseen external assistant output to appear.
 
 ### Base Path And Remote URLs
 

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

@@ -2396,6 +2396,152 @@ describe("ProviderRuntimeIngestion", () => {
     ).toBe(false);
   });
 
+  it("reconciles cursor load replay against already projected assistant text before the next turn", async () => {
+    const harness = await createHarness({ serverSettings: { enableAssistantStreaming: true } });
+    const now = "2026-01-01T00:00:00.000Z";
+    const replayedItemId = asItemId("cursor-replay-before-turn");
+
+    harness.emit({
+      type: "turn.started",
+      eventId: asEventId("evt-cursor-load-reconcile-turn-1-started"),
+      provider: ProviderDriverKind.make("cursor"),
+      createdAt: now,
+      threadId: asThreadId("thread-1"),
+      turnId: asTurnId("turn-cursor-load-reconcile-1"),
+    });
+    harness.emit({
+      type: "content.delta",
+      eventId: asEventId("evt-cursor-load-reconcile-turn-1-delta"),
+      provider: ProviderDriverKind.make("cursor"),
+      createdAt: now,
+      threadId: asThreadId("thread-1"),
+      turnId: asTurnId("turn-cursor-load-reconcile-1"),
+      itemId: replayedItemId,
+      payload: {
+        streamKind: "assistant_text",
+        delta: "already seen",
+      },
+    });
+    harness.emit({
+      type: "turn.completed",
+      eventId: asEventId("evt-cursor-load-reconcile-turn-1-completed"),
+      provider: ProviderDriverKind.make("cursor"),
+      createdAt: now,
+      threadId: asThreadId("thread-1"),
+      turnId: asTurnId("turn-cursor-load-reconcile-1"),
+      status: "completed",
+    });
+
+    await waitForThread(
+      harness.readModel,
+      (thread) =>
+        thread.session?.status === "ready" &&
+        thread.messages.some(
+          (message: ProviderRuntimeTestMessage) => message.text === "already seen",
+        ),
+    );
+
+    harness.emit({
+      type: "content.delta",
+      eventId: asEventId("evt-cursor-load-reconcile-replayed-delta"),
+      provider: ProviderDriverKind.make("cursor"),
+      createdAt: now,
+      threadId: asThreadId("thread-1"),
+      itemId: replayedItemId,
+      payload: {
+        streamKind: "assistant_text",
+        delta: "already seen",
+      },
+    });
+    await harness.drain();
+
+    const snapshot = await harness.readModel();
+    const thread = snapshot.threads.find((entry) => entry.id === asThreadId("thread-1"));
+    expect(thread?.messages.map((message: ProviderRuntimeTestMessage) => message.text)).toEqual([
+      "already seen",
+    ]);
+  });
+
+  it("projects unseen cursor load replay assistant text before the next turn", async () => {
+    const harness = await createHarness({ serverSettings: { enableAssistantStreaming: true } });
+    const now = "2026-01-01T00:00:00.000Z";
+    const replayedItemId = asItemId("cursor-replay-with-new-text");
+
+    harness.emit({
+      type: "turn.started",
+      eventId: asEventId("evt-cursor-load-reconcile-new-turn-1-started"),
+      provider: ProviderDriverKind.make("cursor"),
+      createdAt: now,
+      threadId: asThreadId("thread-1"),
+      turnId: asTurnId("turn-cursor-load-reconcile-new-1"),
+    });
+    harness.emit({
+      type: "content.delta",
+      eventId: asEventId("evt-cursor-load-reconcile-new-turn-1-delta"),
+      provider: ProviderDriverKind.make("cursor"),
+      createdAt: now,
+      threadId: asThreadId("thread-1"),
+      turnId: asTurnId("turn-cursor-load-reconcile-new-1"),
+      itemId: replayedItemId,
+      payload: {
+        streamKind: "assistant_text",
+        delta: "already seen",
+      },
+    });
+    harness.emit({
+      type: "turn.completed",
+      eventId: asEventId("evt-cursor-load-reconcile-new-turn-1-completed"),
+      provider: ProviderDriverKind.make("cursor"),
+      createdAt: now,
+      threadId: asThreadId("thread-1"),
+      turnId: asTurnId("turn-cursor-load-reconcile-new-1"),
+      status: "completed",
+    });
+
+    await waitForThread(
+      harness.readModel,
+      (thread) =>
+        thread.session?.status === "ready" &&
+        thread.messages.some(
+          (message: ProviderRuntimeTestMessage) => message.text === "already seen",
+        ),
+    );
+
+    harness.emit({
+      type: "content.delta",
+      eventId: asEventId("evt-cursor-load-reconcile-new-replayed-delta"),
+      provider: ProviderDriverKind.make("cursor"),
+      createdAt: now,
+      threadId: asThreadId("thread-1"),
+      itemId: replayedItemId,
+      payload: {
+        streamKind: "assistant_text",
+        delta: "already seen",
+      },
+    });
+    harness.emit({
+      type: "content.delta",
+      eventId: asEventId("evt-cursor-load-reconcile-new-unseen-delta"),
+      provider: ProviderDriverKind.make("cursor"),
+      createdAt: now,
+      threadId: asThreadId("thread-1"),
+      itemId: replayedItemId,
+      payload: {
+        streamKind: "assistant_text",
+        delta: "outside change",
+      },
+    });
+
+    const thread = await waitForThread(harness.readModel, (entry) =>
+      entry.messages.some(
+        (message: ProviderRuntimeTestMessage) => message.text === "outside change",
+      ),
+    );
+    expect(
+      thread.messages.map((message: ProviderRuntimeTestMessage) => message.text).sort(),
+    ).toEqual(["already seen", "outside change"]);
+  });
+
   it("streams assistant deltas when thread.turn.start requests streaming mode", async () => {
     const harness = await createHarness({ serverSettings: { enableAssistantStreaming: true } });
     const now = "2026-01-01T00:00:00.000Z";

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

@@ -55,6 +55,8 @@ const TURN_MESSAGE_IDS_BY_TURN_CACHE_CAPACITY = 10_000;
 const TURN_MESSAGE_IDS_BY_TURN_TTL = Duration.minutes(120);
 const STALE_REPLAY_ITEM_IDS_BY_TURN_CACHE_CAPACITY = 10_000;
 const STALE_REPLAY_ITEM_IDS_BY_TURN_TTL = Duration.minutes(120);
+const CURSOR_REPLAY_REMAINING_ASSISTANT_TEXT_BY_THREAD_CACHE_CAPACITY = 10_000;
+const CURSOR_REPLAY_REMAINING_ASSISTANT_TEXT_BY_THREAD_TTL = Duration.minutes(120);
 const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_CACHE_CAPACITY = 20_000;
 const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_TTL = Duration.minutes(120);
 const BUFFERED_PROPOSED_PLAN_BY_ID_CACHE_CAPACITY = 10_000;
@@ -228,6 +230,26 @@ function hasRenderableAssistantText(text: string | undefined): boolean {
   return (text?.trim().length ?? 0) > 0;
 }
 
+function joinedAssistantMessageText(messages: ReadonlyArray<OrchestrationMessage>): string {
+  let text = "";
+  for (let index = 0; index < messages.length; index += 1) {
+    const message = messages[index];
+    if (message?.role === "assistant" && message.text.length > 0) {
+      text += message.text;
+    }
+  }
+  return text;
+}
+
+function isCursorTurnlessAssistantDelta(event: ProviderRuntimeEvent): boolean {
+  return (
+    String(event.provider) === "cursor" &&
+    event.type === "content.delta" &&
+    event.turnId === undefined &&
+    event.payload.streamKind === "assistant_text"
+  );
+}
+
 function proposedPlanIdForTurn(threadId: ThreadId, turnId: TurnId): string {
   return `plan:${threadId}:turn:${turnId}`;
 }
@@ -797,6 +819,12 @@ const make = Effect.gen(function* () {
     lookup: () => Effect.succeed(new Set<string>()),
   });
 
+  const cursorReplayRemainingAssistantTextByThreadId = yield* Cache.make<ThreadId, string>({
+    capacity: CURSOR_REPLAY_REMAINING_ASSISTANT_TEXT_BY_THREAD_CACHE_CAPACITY,
+    timeToLive: CURSOR_REPLAY_REMAINING_ASSISTANT_TEXT_BY_THREAD_TTL,
+    lookup: () => Effect.succeed(""),
+  });
+
   const bufferedProposedPlanById = yield* Cache.make<string, { text: string; createdAt: string }>({
     capacity: BUFFERED_PROPOSED_PLAN_BY_ID_CACHE_CAPACITY,
     timeToLive: BUFFERED_PROPOSED_PLAN_BY_ID_TTL,
@@ -909,6 +937,39 @@ const make = Effect.gen(function* () {
   const clearStaleReplayItemsForTurn = (threadId: ThreadId, turnId: TurnId) =>
     Cache.invalidate(staleReplayItemIdsByTurnKey, providerTurnKey(threadId, turnId));
 
+  const reconcileCursorTurnlessAssistantDelta = (input: {
+    threadId: ThreadId;
+    messages: ReadonlyArray<OrchestrationMessage>;
+    delta: string;
+  }) =>
+    Cache.getOption(cursorReplayRemainingAssistantTextByThreadId, input.threadId).pipe(
+      Effect.flatMap((cachedRemaining) =>
+        Effect.gen(function* () {
+          const remainingAlreadySeenText = Option.getOrElse(cachedRemaining, () =>
+            joinedAssistantMessageText(input.messages),
+          );
+          if (remainingAlreadySeenText.length === 0) {
+            yield* Cache.set(cursorReplayRemainingAssistantTextByThreadId, input.threadId, "");
+            return input.delta;
+          }
+          if (remainingAlreadySeenText.startsWith(input.delta)) {
+            yield* Cache.set(
+              cursorReplayRemainingAssistantTextByThreadId,
+              input.threadId,
+              remainingAlreadySeenText.slice(input.delta.length),
+            );
+            return "";
+          }
+          if (input.delta.startsWith(remainingAlreadySeenText)) {
+            yield* Cache.set(cursorReplayRemainingAssistantTextByThreadId, input.threadId, "");
+            return input.delta.slice(remainingAlreadySeenText.length);
+          }
+          yield* Cache.set(cursorReplayRemainingAssistantTextByThreadId, input.threadId, "");
+          return input.delta;
+        }),
+      ),
+    );
+
   const getActiveAssistantMessageIdForTurn = (threadId: ThreadId, turnId: TurnId) =>
     getAssistantSegmentStateForTurn(threadId, turnId).pipe(
       Effect.map((state) =>
@@ -1565,6 +1626,16 @@ const make = Effect.gen(function* () {
 
       if (assistantDelta && assistantDelta.length > 0) {
         const turnId = toTurnId(event.turnId);
+        const reconciledAssistantDelta = isCursorTurnlessAssistantDelta(event)
+          ? yield* reconcileCursorTurnlessAssistantDelta({
+              threadId: thread.id,
+              messages: (yield* getLoadedThreadDetail())?.messages ?? [],
+              delta: assistantDelta,
+            })
+          : assistantDelta;
+        if (reconciledAssistantDelta.length === 0) {
+          return;
+        }
         const assistantMessageId = yield* getOrCreateAssistantMessageId({
           threadId: thread.id,
           event,
@@ -1579,7 +1650,10 @@ const make = Effect.gen(function* () {
           (settings) => (settings.enableAssistantStreaming ? "streaming" : "buffered"),
         );
         if (assistantDeliveryMode === "buffered") {
-          const spillChunk = yield* appendBufferedAssistantText(assistantMessageId, assistantDelta);
+          const spillChunk = yield* appendBufferedAssistantText(
+            assistantMessageId,
+            reconciledAssistantDelta,
+          );
           if (spillChunk.length > 0) {
             yield* orchestrationEngine.dispatch({
               type: "thread.message.assistant.delta",
@@ -1597,7 +1671,7 @@ const make = Effect.gen(function* () {
             commandId: yield* providerCommandId(event, "assistant-delta"),
             threadId: thread.id,
             messageId: assistantMessageId,
-            delta: assistantDelta,
+            delta: reconciledAssistantDelta,
             ...(turnId ? { turnId } : {}),
             createdAt: now,
           });
@@ -1760,6 +1834,7 @@ const make = Effect.gen(function* () {
           yield* clearAssistantMessageIdsForTurn(thread.id, turnId);
           yield* clearAssistantSegmentStateForTurn(thread.id, turnId);
           yield* clearStaleReplayItemsForTurn(thread.id, turnId);
+          yield* Cache.invalidate(cursorReplayRemainingAssistantTextByThreadId, thread.id);
 
           yield* finalizeBufferedProposedPlan({
             event,
@@ -1774,6 +1849,7 @@ const make = Effect.gen(function* () {
 
       if (event.type === "session.exited") {
         yield* clearTurnStateForSession(thread.id);
+        yield* Cache.invalidate(cursorReplayRemainingAssistantTextByThreadId, thread.id);
       }
 
       if (event.type === "runtime.error") {

apps/server/src/provider/acp/AcpSessionRuntime.test.ts

@@ -0,0 +1,81 @@
+import { assert, it } from "@effect/vitest";
+import * as Effect from "effect/Effect";
+import * as Queue from "effect/Queue";
+import * as Ref from "effect/Ref";
+import type * as EffectAcpSchema from "effect-acp/schema";
+
+import { handleSessionUpdateForTest } from "./AcpSessionRuntime.ts";
+import type {
+  AcpParsedSessionEvent,
+  AcpSessionModeState,
+  AcpToolCallState,
+} from "./AcpRuntimeModel.ts";
+
+it.effect("emits loaded-session updates for downstream reconciliation", () =>
+  Effect.gen(function* () {
+    const queue = yield* Queue.unbounded<AcpParsedSessionEvent>();
+    const modeStateRef = yield* Ref.make<AcpSessionModeState | undefined>({
+      currentModeId: "ask",
+      availableModes: [
+        { id: "ask", name: "Ask" },
+        { id: "code", name: "Code" },
+      ],
+    });
+    const toolCallsRef = yield* Ref.make(new Map<string, AcpToolCallState>());
+    const assistantSegmentRef = yield* Ref.make({ nextSegmentIndex: 0 });
+
+    const handle = (params: EffectAcpSchema.SessionNotification) =>
+      handleSessionUpdateForTest({
+        queue,
+        modeStateRef,
+        toolCallsRef,
+        assistantSegmentRef,
+        params,
+      });
+
+    yield* handle({
+      sessionId: "cursor-session",
+      update: {
+        sessionUpdate: "current_mode_update",
+        currentModeId: "code",
+      },
+    });
+    yield* handle({
+      sessionId: "cursor-session",
+      update: {
+        sessionUpdate: "plan",
+        entries: [{ content: "Old replayed plan", priority: "high", status: "completed" }],
+      },
+    });
+    yield* handle({
+      sessionId: "cursor-session",
+      update: {
+        sessionUpdate: "user_message_chunk",
+        content: { type: "text", text: "old replayed user prompt" },
+      },
+    });
+    yield* handle({
+      sessionId: "cursor-session",
+      update: {
+        sessionUpdate: "agent_message_chunk",
+        content: { type: "text", text: "old replayed assistant text" },
+      },
+    });
+
+    assert.equal(yield* Queue.size(queue), 4);
+    assert.equal((yield* Ref.get(modeStateRef))?.currentModeId, "code");
+
+    const mode = yield* Queue.take(queue);
+    const plan = yield* Queue.take(queue);
+    const started = yield* Queue.take(queue);
+    const delta = yield* Queue.take(queue);
+    assert.equal(mode._tag, "ModeChanged");
+    assert.equal(plan._tag, "PlanUpdated");
+    assert.equal(started._tag, "AssistantItemStarted");
+    assert.equal(delta._tag, "ContentDelta");
+    if (delta._tag === "ContentDelta") {
+      assert.equal(delta.text, "old replayed assistant text");
+    }
+    assert.equal(yield* Queue.size(queue), 0);
+  }),
+);

apps/server/src/provider/acp/AcpSessionRuntime.ts

@@ -672,6 +672,8 @@ const handleSessionUpdate = ({
     }
   });
 
+export const handleSessionUpdateForTest = handleSessionUpdate;
+
 function updateModeState(modeState: AcpSessionModeState, nextModeId: string): AcpSessionModeState {
   const normalized = nextModeId.trim();
   if (!normalized) {