🤖 fix: admit fresh automatic input after settled Stop

src/common/orpc/schemas/message.ts

@@ -198,6 +198,7 @@ export const MuxMessageSchema = z.object({
       // Durable boundary marker for compaction summaries.
       compactionBoundary: z.boolean().optional(),
       compactionPublicationId: z.string().min(1).optional().catch(undefined),
+      compactionPublicationGeneration: z.string().min(1).nullable().optional().catch(undefined),
       contextBoundaryKind: z.literal(CONTEXT_BOUNDARY_KINDS.RESET).optional(),
       toolPolicy: z.any().optional(),
       disableWorkspaceAgents: z.boolean().optional(),

src/common/types/message.ts

@@ -1050,6 +1050,8 @@ export interface MuxMetadata {
   compactionBoundary?: boolean;
   /** Exact composed publication occurrence, shared with the pending file writeId. */
   compactionPublicationId?: string;
+  /** Captured generation of this publication; legacy rewrites must not inherit it. */
+  compactionPublicationGeneration?: string | null;
   /** Durable provider-context boundary kind. Existing compaction rows are also boundaries via compactionBoundary. */
   contextBoundaryKind?: PersistedContextBoundaryKind;
   toolPolicy?: ToolPolicy; // Tool policy active when this message was sent (user messages only)

src/node/services/agentSession.admissionGates.test.ts

@@ -67,7 +67,7 @@ describe("AgentSession.sendMessage (admission gates)", () => {
   it("refuses at the pre-persist gate before any row lands when the epoch is stale", async () => {
     const workspaceId = "ws-epoch-prepersist";
     const { session, historyService, streamMessage } = await createSessionHarness(workspaceId);
-    const appendMany = spyOn(historyService, "appendManyToHistory");
+    const publication = spyOn(historyService, "acceptCompactionReplacement");
     let acceptedCalls = 0;
 
     const result = await session.sendMessage(
@@ -95,7 +95,7 @@ describe("AgentSession.sendMessage (admission gates)", () => {
     });
     // Pre-acceptance refusal: nothing persisted, nothing accepted, no stream.
     expect(acceptedCalls).toBe(0);
-    expect(appendMany).not.toHaveBeenCalled();
+    expect(publication).not.toHaveBeenCalled();
     expect(streamMessage).not.toHaveBeenCalled();
     const history = await historyService.getHistoryFromLatestBoundary(workspaceId);
     expect(history.success ? history.data : ["unexpected"]).toHaveLength(0);
@@ -104,7 +104,18 @@ describe("AgentSession.sendMessage (admission gates)", () => {
   it("invokes the cancellation hook when the caller probe goes stale before acceptance", async () => {
     const workspaceId = "ws-caller-stale-cancel";
     const { session, historyService, streamMessage } = await createSessionHarness(workspaceId);
-    const appendMany = spyOn(historyService, "appendManyToHistory");
+    let published = false;
+    const publish = historyService.acceptCompactionReplacement.bind(historyService);
+    spyOn(historyService, "acceptCompactionReplacement").mockImplementationOnce(
+      (id, capture, operation, observer) =>
+        publish(id, capture, operation, {
+          ...observer,
+          onCommitted: (receipt) => {
+            observer.onCommitted(receipt);
+            published = true;
+          },
+        })
+    );
     const canceled: string[] = [];
 
     const result = await session.sendMessage(
@@ -123,7 +134,7 @@ describe("AgentSession.sendMessage (admission gates)", () => {
         // which must roll the rows back AND surface the refusal through the cancellation hook —
         // a queued peer send's caller already returned success and this hook carries its budget
         // refund; without it the reservation would leak.
-        admissionStale: () => appendMany.mock.calls.length > 0,
+        admissionStale: () => published,
         onCanceled: (reason: string) => {
           canceled.push(reason);
         },
@@ -140,7 +151,18 @@ describe("AgentSession.sendMessage (admission gates)", () => {
   it("keeps the charge when a stale send's rollback did not commit", async () => {
     const workspaceId = "ws-caller-stale-rollback-failed";
     const { session, historyService, streamMessage } = await createSessionHarness(workspaceId);
-    const appendMany = spyOn(historyService, "appendManyToHistory");
+    let published = false;
+    const publish = historyService.acceptCompactionReplacement.bind(historyService);
+    spyOn(historyService, "acceptCompactionReplacement").mockImplementationOnce(
+      (id, capture, operation, observer) =>
+        publish(id, capture, operation, {
+          ...observer,
+          onCommitted: (receipt) => {
+            observer.onCommitted(receipt);
+            published = true;
+          },
+        })
+    );
     // Rollback deletion fails and the rows verifiably REMAIN: the cancellation hook must not
     // fire — a refunded reservation with durable rows would let the payload enter provider
     // context after a resume while no longer counting against the sender's budget.
@@ -162,7 +184,7 @@ describe("AgentSession.sendMessage (admission gates)", () => {
             synthetic: true,
           }),
         ],
-        admissionStale: () => appendMany.mock.calls.length > 0,
+        admissionStale: () => published,
         onCanceled: (reason: string) => {
           canceled.push(reason);
         },

src/node/services/agentSession.autoCompaction.test.ts

@@ -1443,117 +1443,91 @@ describe("AgentSession on-send auto-compaction for synthetic guidance sends", ()
     return predicate();
   }
 
-  test("unresolved Stop refuses automatic guidance before creating a compaction continuation", async () => {
-    const workspaceId = "ws-unresolved-stop-guidance";
-    const fixture = await createGuidanceHarness({ workspaceId });
-    try {
-      expect(await fixture.session.interruptStream()).toEqual(Ok(undefined));
-      const accepted = mock(() => undefined);
+  test.each([false, true])(
+    "applies compaction for fresh automatic guidance after settled Stop (%s)",
+    async (stopped) => {
+      const fixture = await createGuidanceHarness({
+        workspaceId: "ws-auto-compaction-synthetic-guidance",
+      });
+
+      const monitor = (fixture.session as unknown as { compactionMonitor: CompactionMonitor })
+        .compactionMonitor;
+      let firstCheck = true;
+      spyOn(monitor, "checkBeforeSend").mockImplementation(() => {
+        const high = firstCheck;
+        firstCheck = false;
+        return {
+          shouldShowWarning: high,
+          shouldForceCompact: high,
+          usagePercentage: high ? 95 : 1,
+          thresholdPercentage: 70,
+          contextTokens: high ? 95_000 : 1_000,
+          maxTokens: 100_000,
+        };
+      });
+      if (stopped) expect(await fixture.session.interruptStream()).toEqual(Ok(undefined));
+
       const result = await fixture.session.sendMessage(
-        "Guidance that would otherwise force compaction",
+        "Updated guidance from parent: focus on the failing tests.",
         { model: "openai:gpt-4o", agentId: "exec" },
         {
           acceptanceOrigin: "automatic",
           synthetic: true,
           agentInitiated: true,
           startStreamInBackground: true,
-          onAccepted: accepted,
         }
       );
-      expect(result.success).toBe(false);
-      expect(accepted).not.toHaveBeenCalled();
-      expect(fixture.streamHistories).toEqual([]);
-      expect(await fixture.historyService.getHistoryFromLatestBoundary(workspaceId)).toEqual(
-        Ok([])
-      );
-      expect(fixture.session.isBusy()).toBe(false);
-      expect(fixture.session.queuedMessageEntryCount()).toBe(0);
-    } finally {
-      await fixture.session.dispose();
-    }
-  });
-
-  test("applies compaction for fresh automatic guidance without Stop", async () => {
-    const fixture = await createGuidanceHarness({
-      workspaceId: "ws-auto-compaction-synthetic-guidance",
-    });
-
-    const monitor = (fixture.session as unknown as { compactionMonitor: CompactionMonitor })
-      .compactionMonitor;
-    let firstCheck = true;
-    spyOn(monitor, "checkBeforeSend").mockImplementation(() => {
-      const high = firstCheck;
-      firstCheck = false;
-      return {
-        shouldShowWarning: high,
-        shouldForceCompact: high,
-        usagePercentage: high ? 95 : 1,
-        thresholdPercentage: 70,
-        contextTokens: high ? 95_000 : 1_000,
-        maxTokens: 100_000,
-      };
-    });
-
-    const result = await fixture.session.sendMessage(
-      "Updated guidance from parent: focus on the failing tests.",
-      { model: "openai:gpt-4o", agentId: "exec" },
-      {
-        acceptanceOrigin: "automatic",
-        synthetic: true,
-        agentInitiated: true,
-        startStreamInBackground: true,
-      }
-    );
-    expect(result.success).toBe(true);
-
-    // First stream must carry the persisted compaction request.
-    await waitFor(() => fixture.streamHistories.length >= 1);
-    expect(fixture.streamHistories.length).toBeGreaterThanOrEqual(1);
-    const firstRequestHasCompactionRequest = fixture.streamHistories[0].some(
-      (message) => message.metadata?.muxMetadata?.type === "compaction-request"
-    );
-    expect(firstRequestHasCompactionRequest).toBe(true);
+      expect(result.success).toBe(true);
 
-    // Compaction must complete: a boundary summary lands in durable history.
-    const boundaryLanded = await waitFor(async () => {
-      const historyResult = await fixture.historyService.getHistoryFromLatestBoundary(
-        "ws-auto-compaction-synthetic-guidance"
+      // First stream must carry the persisted compaction request.
+      await waitFor(() => fixture.streamHistories.length >= 1);
+      expect(fixture.streamHistories.length).toBeGreaterThanOrEqual(1);
+      const firstRequestHasCompactionRequest = fixture.streamHistories[0].some(
+        (message) => message.metadata?.muxMetadata?.type === "compaction-request"
       );
-      return (
-        historyResult.success &&
-        historyResult.data.some((message) => message.metadata?.compactionBoundary === true)
+      expect(firstRequestHasCompactionRequest).toBe(true);
+
+      // Compaction must complete: a boundary summary lands in durable history.
+      const boundaryLanded = await waitFor(async () => {
+        const historyResult = await fixture.historyService.getHistoryFromLatestBoundary(
+          "ws-auto-compaction-synthetic-guidance"
+        );
+        return (
+          historyResult.success &&
+          historyResult.data.some((message) => message.metadata?.compactionBoundary === true)
+        );
+      });
+      expect(boundaryLanded).toBe(true);
+
+      // The original guidance text is re-dispatched as the post-compaction follow-up.
+      const followUpDispatched = await waitFor(() =>
+        fixture.streamHistories.some((history) =>
+          history.some(
+            (message) =>
+              message.role === "user" &&
+              message.metadata?.muxMetadata?.type !== "compaction-request" &&
+              message.parts.some(
+                (part) => part.type === "text" && part.text.includes("focus on the failing tests")
+              )
+          )
+        )
       );
-    });
-    expect(boundaryLanded).toBe(true);
+      expect(followUpDispatched).toBe(true);
 
-    // The original guidance text is re-dispatched as the post-compaction follow-up.
-    const followUpDispatched = await waitFor(() =>
-      fixture.streamHistories.some((history) =>
-        history.some(
-          (message) =>
-            message.role === "user" &&
-            message.metadata?.muxMetadata?.type !== "compaction-request" &&
-            message.parts.some(
-              (part) => part.type === "text" && part.text.includes("focus on the failing tests")
-            )
+      expect(
+        fixture.events.some(
+          (event) => (event as { type?: string }).type === "auto-compaction-triggered"
         )
-      )
-    );
-    expect(followUpDispatched).toBe(true);
-
-    expect(
-      fixture.events.some(
-        (event) => (event as { type?: string }).type === "auto-compaction-triggered"
-      )
-    ).toBe(true);
-    expect(
-      fixture.events.some(
-        (event) => (event as { type?: string }).type === "auto-compaction-completed"
-      )
-    ).toBe(true);
+      ).toBe(true);
+      expect(
+        fixture.events.some(
+          (event) => (event as { type?: string }).type === "auto-compaction-completed"
+        )
+      ).toBe(true);
 
-    await fixture.session.dispose();
-  });
+      await fixture.session.dispose();
+    }
+  );
 
   // Characterization: sends carrying preTurnMessages (family-message payloads)
   // intentionally skip on-send compaction. The trigger row references its