fix: send continueMessage after compaction completes

ammar-agent · ammar-agent · commit c18578f8f4b5 · 2025-12-21T15:13:58.000-06:00
diff --git a/src/node/services/agentSession.ts b/src/node/services/agentSession.ts
@@ -1099,6 +1099,16 @@ export class AgentSession {
       return;
     }
 
+    // If a compaction operation is active, do NOT auto-send queued messages.
+    //
+    // Why: if an earlier stream (the one we interrupted to start compaction) emits a late
+    // stream-end/tool-call-end event, it can trigger sendQueuedMessages() while compaction is
+    // still streaming. That can cause the queued continue message to be sent too early and
+    // then wiped by the compaction history replacement.
+    if (this.activeCompactionOperation) {
+      return;
+    }
+
     // Clear the queued message flag (even if queue is empty, to handle race conditions)
     this.backgroundProcessManager.setMessageQueued(this.workspaceId, false);
 
diff --git a/tests/ipc/compactHistory.test.ts b/tests/ipc/compactHistory.test.ts
@@ -1,22 +1,24 @@
 /**
  * compactHistory integration tests.
  *
- * Ensures compaction is a control-plane operation (not a slash-command string), and that
- * history is replaced only on successful compaction completion.
+ * Ensures compaction is a control-plane operation (not a slash-command string), and that:
+ * - History is replaced only on successful compaction completion
+ * - continueMessage is auto-sent after compaction completes
  *
  * Requirements:
- * - Uses the Haiku model for both normal messages and compaction
- * - Builds history by sending messages (replicates user behavior)
+ * - Uses the Haiku model for both compaction and the follow-up continue message
+ * - Seeds history via HistoryService (test-only) to avoid extra API calls
  */
 
+import type { WorkspaceChatMessage } from "@/common/orpc/types";
 import { shouldRunIntegrationTests, validateApiKeys } from "./setup";
 import {
   createSharedRepo,
   cleanupSharedRepo,
   withSharedWorkspace,
   configureTestRetries,
 } from "./sendMessageTestHelpers";
-import { assertStreamSuccess, modelString, sendMessageWithModel } from "./helpers";
+import { modelString, seedHistoryMessages } from "./helpers";
 import { KNOWN_MODELS } from "../../src/common/constants/knownModels";
 
 // Skip all tests if TEST_INTEGRATION is not set
@@ -29,47 +31,78 @@ if (shouldRunIntegrationTests()) {
 beforeAll(createSharedRepo);
 afterAll(cleanupSharedRepo);
 
+function getTextFromMessageParts(message: {
+  parts?: Array<{ type: string; text?: string }>;
+}): string {
+  return (
+    message.parts
+      ?.filter((part) => part.type === "text")
+      .map((part) => part.text ?? "")
+      .join("") ?? ""
+  );
+}
+
+async function waitForMatchingEvent(
+  collector: { getEvents: () => WorkspaceChatMessage[] },
+  predicate: (event: WorkspaceChatMessage) => boolean,
+  timeoutMs: number
+): Promise<WorkspaceChatMessage | null> {
+  const start = Date.now();
+  while (Date.now() - start < timeoutMs) {
+    const match = collector.getEvents().find(predicate);
+    if (match) {
+      return match;
+    }
+    await new Promise((resolve) => setTimeout(resolve, 25));
+  }
+  return null;
+}
+
 describeIntegration("compactHistory integration tests", () => {
   configureTestRetries(3);
 
   test.concurrent(
-    "should compact history using Haiku for both messages + compaction",
+    "should compact history and then auto-send continueMessage",
     async () => {
       await withSharedWorkspace("anthropic", async ({ env, workspaceId, collector }) => {
         const haiku = modelString("anthropic", KNOWN_MODELS.HAIKU.providerModelId);
 
-        // Build history via normal user interactions.
-        collector.clear();
-
-        const message1 =
-          "You are helping me plan a small refactor. Explain, in a few sentences, what the risks are when refactoring code without tests.";
-        const result1 = await sendMessageWithModel(env, workspaceId, message1, haiku);
-        expect(result1.success).toBe(true);
-        const streamEnd1 = await collector.waitForEvent("stream-end", 20000);
-        expect(streamEnd1).not.toBeNull();
-        expect((streamEnd1 as { metadata: { model?: string } }).metadata.model).toBe(haiku);
-        assertStreamSuccess(collector);
+        // Seed history quickly (test-only) to avoid extra API calls.
+        const seededIds = await seedHistoryMessages(workspaceId, env.config, [
+          {
+            id: "seed-user-0",
+            role: "user",
+            content: "Context: we are discussing a small code refactor.",
+          },
+          {
+            id: "seed-assistant-0",
+            role: "assistant",
+            content: "Acknowledged. I will help.",
+          },
+          {
+            id: "seed-user-1",
+            role: "user",
+            content: "Please keep responses short and practical.",
+          },
+          {
+            id: "seed-assistant-1",
+            role: "assistant",
+            content: "Understood.",
+          },
+        ]);
 
         collector.clear();
 
-        const message2 =
-          "Now list three concrete steps I should take to refactor safely. Include enough detail that it would be useful in a code review.";
-        const result2 = await sendMessageWithModel(env, workspaceId, message2, haiku);
-        expect(result2.success).toBe(true);
-        const streamEnd2 = await collector.waitForEvent("stream-end", 20000);
-        expect(streamEnd2).not.toBeNull();
-        expect((streamEnd2 as { metadata: { model?: string } }).metadata.model).toBe(haiku);
-        assertStreamSuccess(collector);
-
-        collector.clear();
+        const continueText = "Continue: reply with exactly 'OK'.";
 
-        // Trigger compaction explicitly via the control-plane API.
+        // Trigger compaction via the control-plane API.
         const compactResult = await env.orpc.workspace.compactHistory({
           workspaceId,
           model: haiku,
           maxOutputTokens: 800,
           source: "user",
           interrupt: "none",
+          continueMessage: { text: continueText },
           sendMessageOptions: {
             model: haiku,
             thinkingLevel: "off",
@@ -81,41 +114,89 @@ describeIntegration("compactHistory integration tests", () => {
           throw new Error(String(compactResult.error));
         }
 
-        // Ensure this stream is actually the compaction stream.
-        const streamStart = await collector.waitForEvent("stream-start", 20000);
-        expect(streamStart).not.toBeNull();
-        const compactionMessageId = (streamStart as { messageId: string }).messageId;
-
-        const streamEnd = await collector.waitForEvent("stream-end", 30000);
-        expect(streamEnd).not.toBeNull();
-        expect((streamEnd as { messageId: string }).messageId).toBe(compactionMessageId);
-        expect((streamEnd as { metadata: { model?: string } }).metadata.model).toBe(haiku);
-        assertStreamSuccess(collector);
-
-        // The compaction handler emits a single summary message + delete event.
-        const deleteEvent = collector.getEvents().find((e) => e.type === "delete");
-        expect(deleteEvent).toBeDefined();
-
-        const summaryMessage = collector
-          .getEvents()
-          .find((e) => e.type === "message" && e.role === "assistant" && e.metadata?.compacted);
-        expect(summaryMessage).toBeDefined();
-        expect((summaryMessage as { metadata?: { model?: string } }).metadata?.model).toBe(haiku);
-
-        // Verify persisted history was replaced (user behavior: reload workspace).
+        // Wait for compaction stream to start + end.
+        const compactionStreamStart = await collector.waitForEvent("stream-start", 20000);
+        expect(compactionStreamStart).not.toBeNull();
+        const compactionMessageId = (compactionStreamStart as { messageId: string }).messageId;
+
+        const compactionStreamEnd = await waitForMatchingEvent(
+          collector,
+          (e) =>
+            e.type === "stream-end" &&
+            (e as { messageId?: string }).messageId === compactionMessageId,
+          45000
+        );
+        expect(compactionStreamEnd).not.toBeNull();
+        expect((compactionStreamEnd as { metadata: { model?: string } }).metadata.model).toBe(
+          haiku
+        );
+
+        // Compaction should emit delete + summary message.
+        const deleteEvent = await waitForMatchingEvent(
+          collector,
+          (e) => e.type === "delete",
+          10000
+        );
+        expect(deleteEvent).not.toBeNull();
+
+        const summaryMessage = await waitForMatchingEvent(
+          collector,
+          (e) => e.type === "message" && e.role === "assistant" && Boolean(e.metadata?.compacted),
+          10000
+        );
+        expect(summaryMessage).not.toBeNull();
+
+        // Continue message should be persisted as a user message and then streamed.
+        const continueUserMessage = await waitForMatchingEvent(
+          collector,
+          (e) =>
+            e.type === "message" &&
+            e.role === "user" &&
+            getTextFromMessageParts(e) === continueText,
+          20000
+        );
+        expect(continueUserMessage).not.toBeNull();
+
+        const continueStreamStart = await waitForMatchingEvent(
+          collector,
+          (e) =>
+            e.type === "stream-start" &&
+            (e as { messageId?: string }).messageId !== compactionMessageId &&
+            (e as { metadata?: { model?: string } }).metadata?.model === haiku,
+          20000
+        );
+        expect(continueStreamStart).not.toBeNull();
+
+        const continueMessageId = (continueStreamStart as { messageId: string }).messageId;
+        const continueStreamEnd = await waitForMatchingEvent(
+          collector,
+          (e) =>
+            e.type === "stream-end" &&
+            (e as { messageId?: string }).messageId === continueMessageId,
+          45000
+        );
+        expect(continueStreamEnd).not.toBeNull();
+
+        // Verify persisted history:
+        // - seeded messages were removed
+        // - summary exists
+        // - continue message exists
         const replay = await env.orpc.workspace.getFullReplay({ workspaceId });
         const replayMessages = replay.filter((m) => m.type === "message");
 
-        // After compaction we should only have a single assistant summary message.
-        expect(replayMessages).toHaveLength(1);
-        expect(replayMessages[0].role).toBe("assistant");
-        expect(replayMessages[0].metadata?.compacted).toBeDefined();
-        expect(replayMessages[0].metadata?.model).toBe(haiku);
+        for (const id of seededIds) {
+          expect(replayMessages.some((m) => m.id === id)).toBe(false);
+        }
+
+        const summaryIndex = replayMessages.findIndex(
+          (m) => m.role === "assistant" && Boolean(m.metadata?.compacted)
+        );
+        expect(summaryIndex).toBe(0);
 
-        // Sanity check: original user prompt text should not be present after replacement.
-        const replayText = JSON.stringify(replayMessages[0]);
-        expect(replayText).not.toContain("refactoring code without tests");
-        expect(replayText).not.toContain("three concrete steps");
+        const continueIndex = replayMessages.findIndex(
+          (m) => m.role === "user" && getTextFromMessageParts(m) === continueText
+        );
+        expect(continueIndex).toBeGreaterThan(summaryIndex);
       });
     },
     90000
diff --git a/tests/ipc/helpers.ts b/tests/ipc/helpers.ts
@@ -590,6 +590,36 @@ export async function cleanupTempGitRepo(repoPath: string): Promise<void> {
   console.warn(`Failed to cleanup temp git repo after ${maxRetries} attempts:`, lastError);
 }
 
+/**
+ * Seed a workspace history with explicit messages.
+ *
+ * Test-only: uses HistoryService directly to populate chat.jsonl without making API calls.
+ * Real application code should NEVER bypass IPC like this.
+ */
+export async function seedHistoryMessages(
+  workspaceId: string,
+  config: { getSessionDir: (id: string) => string },
+  messages: Array<{ id?: string; role: "user" | "assistant"; content: string }>
+): Promise<string[]> {
+  // HistoryService only needs getSessionDir, so we can cast the partial config.
+  const historyService = new HistoryService(config as any);
+
+  const ids: string[] = [];
+  for (let i = 0; i < messages.length; i++) {
+    const entry = messages[i];
+    const id = entry.id ?? `seed-msg-${i}`;
+    ids.push(id);
+
+    const message = createMuxMessage(id, entry.role, entry.content, {});
+    const result = await historyService.appendToHistory(workspaceId, message);
+    if (!result.success) {
+      throw new Error(`Failed to append history message ${i} (${id}): ${result.error}`);
+    }
+  }
+
+  return ids;
+}
+
 /**
  * Build large conversation history to test context limits
  *
