fix(claude): open a single compaction marker per compaction (#4391)

Suppress repeated Claude compaction heartbeat markers at the provider and re-arm marker creation at compaction and turn boundaries. Verify emitted events through the public session interface, including subsequent compactions after interrupted or abandoned turns.

Closes #4390.

Co-authored-by: Tom Gringauz <17593920+tomgrin10@users.noreply.github.com>

packages/server/src/server/agent/providers/claude/agent.test.ts

@@ -24,6 +24,12 @@ interface TestClaudeSession {
   close(): Promise<void>;
 }
 
+function isLoadingCompactionEvent(event: AgentStreamEvent): boolean {
+  return (
+    event.type === "timeline" && event.item.type === "compaction" && event.item.status === "loading"
+  );
+}
+
 function isPermissionResolvedEvent(
   event: AgentStreamEvent,
 ): event is Extract<AgentStreamEvent, { type: "permission_resolved" }> {
@@ -1981,6 +1987,15 @@ describe("ClaudeAgentSession context window usage", () => {
     };
   }
 
+  function createCompactingStatus(): Record<string, unknown> {
+    return {
+      type: "system",
+      subtype: "status",
+      status: "compacting",
+      session_id: "session-1",
+    };
+  }
+
   test("emits turn_started before the submitted user message", async () => {
     const session = await createSessionForTurns([[]]);
     const events: AgentStreamEvent[] = [];
@@ -2857,6 +2872,110 @@ describe("ClaudeAgentSession context window usage", () => {
     }
   });
 
+  test("repeated compacting statuses open a single compaction marker", async () => {
+    const session = await createSessionForTurns([
+      [
+        createCompactingStatus(),
+        createCompactingStatus(),
+        createCompactingStatus(),
+        createCompactBoundary(),
+        createCompactingStatus(),
+        createCompactingStatus(),
+        createCompactBoundary(),
+        createSuccessResult(),
+      ],
+    ]);
+
+    try {
+      const events = await collectStreamEvents(session, "compact twice");
+      const compactions = events.flatMap((event) =>
+        event.type === "timeline" && event.item.type === "compaction" ? [event.item.status] : [],
+      );
+      expect(compactions).toEqual(["loading", "completed", "loading", "completed"]);
+    } finally {
+      await session.close();
+    }
+  });
+
+  test("a compaction abandoned mid-turn does not suppress the next compaction marker", async () => {
+    // The first turn starts compacting and then ends without ever reaching a
+    // compact_boundary, so the marker it opened is never resolved.
+    const session = await createSessionForTurns([
+      [createCompactingStatus(), createSuccessResult()],
+      [createCompactingStatus(), createSuccessResult()],
+    ]);
+
+    try {
+      const abandonedTurn = await collectStreamEvents(session, "abandoned compaction");
+      expect(abandonedTurn.filter(isLoadingCompactionEvent)).toHaveLength(1);
+
+      const nextTurn = await collectStreamEvents(session, "next compaction");
+      expect(nextTurn.filter(isLoadingCompactionEvent)).toHaveLength(1);
+    } finally {
+      await session.close();
+    }
+  });
+
+  test("an interrupted compaction does not suppress the next compaction marker", async () => {
+    // The first turn starts compacting and is then interrupted, so it never reaches a
+    // compact_boundary and the marker it opened is never resolved.
+    const session = await createSessionForTurns([
+      [createCompactingStatus()],
+      [createCompactingStatus(), createSuccessResult()],
+    ]);
+
+    try {
+      const interruptedTurn: AgentStreamEvent[] = [];
+      const streaming = (async () => {
+        for await (const event of streamSession(session, "interrupted compaction")) {
+          interruptedTurn.push(event);
+        }
+      })();
+
+      await vi.waitFor(() => {
+        expect(interruptedTurn.filter(isLoadingCompactionEvent)).toHaveLength(1);
+      });
+      await session.interrupt();
+      await streaming;
+
+      expect(interruptedTurn).toContainEqual(
+        expect.objectContaining({ type: "turn_canceled", provider: "claude" }),
+      );
+
+      const nextTurn = await collectStreamEvents(session, "next compaction");
+      expect(nextTurn.filter(isLoadingCompactionEvent)).toHaveLength(1);
+    } finally {
+      await session.close();
+    }
+  });
+
+  test("a compaction abandoned in an autonomous turn does not suppress the next marker", async () => {
+    // Trailing output after the foreground result opens an autonomous turn, which starts
+    // compacting and is then ended by the next foreground turn, never reaching a boundary.
+    const session = await createSessionForTurns([
+      [createSuccessResult(), createMessageStartEvent(), createCompactingStatus()],
+      [createCompactingStatus(), createSuccessResult()],
+    ]);
+
+    try {
+      const observed: AgentStreamEvent[] = [];
+      const unsubscribe = session.subscribe((event) => {
+        observed.push(event);
+      });
+
+      await collectStreamEvents(session, "foreground turn");
+      await vi.waitFor(() => {
+        expect(observed.filter(isLoadingCompactionEvent)).toHaveLength(1);
+      });
+      unsubscribe();
+
+      const nextTurn = await collectStreamEvents(session, "next compaction");
+      expect(nextTurn.filter(isLoadingCompactionEvent)).toHaveLength(1);
+    } finally {
+      await session.close();
+    }
+  });
+
   test("result.result is surfaced as an assistant message when no model output was produced", async () => {
     const session = await createSessionForTest();
 

packages/server/src/server/agent/providers/claude/agent.ts

@@ -2095,6 +2095,7 @@ class ClaudeAgentSession implements AgentSession {
   private lastOptionsModel: string | null = null;
   private lastRuntimeModel: string | null = null;
   private compacting = false;
+  private compactionMarkerOpen = false;
   private queryPumpPromise: Promise<void> | null = null;
   private queryRestartNeeded = false;
   private pendingInterruptAbort = false;
@@ -3572,6 +3573,7 @@ class ClaudeAgentSession implements AgentSession {
     this.activeForegroundInput = null;
     this.cancelCurrentTurn = null;
     this.activeTurnHasAssistantText = false;
+    this.compactionMarkerOpen = false;
     this.syncTurnState("foreground turn terminal");
   }
 
@@ -3583,6 +3585,7 @@ class ClaudeAgentSession implements AgentSession {
     }
 
     if (terminalSeen) {
+      this.compactionMarkerOpen = false;
       if (this.activeForegroundTurnId) {
         this.activeForegroundTurnId = null;
         this.activeForegroundQuery = null;
@@ -3624,6 +3627,7 @@ class ClaudeAgentSession implements AgentSession {
     this.activeForegroundQuery = null;
     this.activeForegroundInput = null;
     this.activeTurnHasAssistantText = false;
+    this.compactionMarkerOpen = false;
     this.syncTurnState("autonomous turn completed");
   }
 
@@ -4260,15 +4264,22 @@ class ClaudeAgentSession implements AgentSession {
       const status = toObjectRecord(message)?.status;
       if (status === "compacting") {
         this.compacting = true;
-        events.push({
-          type: "timeline",
-          item: { type: "compaction", status: "loading" },
-          provider: "claude",
-        });
+        // Claude Code repeats this status every 30 seconds until the compaction
+        // finishes. Each repeat used to open its own marker, and the app only ever
+        // resolves one of them, so the rest stayed on "Compacting..." forever.
+        if (!this.compactionMarkerOpen) {
+          this.compactionMarkerOpen = true;
+          events.push({
+            type: "timeline",
+            item: { type: "compaction", status: "loading" },
+            provider: "claude",
+          });
+        }
       }
       return;
     }
     if (message.subtype === "compact_boundary") {
+      this.compactionMarkerOpen = false;
       const compactMetadata = readCompactionMetadata(message);
       events.push({
         type: "timeline",