Allow context compaction while Agent Chat waits (#19801)

## Thinking Path

> - Paperclip manages agents and their work.
> - Agent Chat can wait for a user between useful turns.
> - The task Properties panel already lets that user request context
compaction.
> - The queue treated that maintenance request as an ordinary wake and
refused it while the chat waited.
> - Compaction must maintain that session without starting a new
conversation turn.
> - This change admits the authenticated request, preserves the waiting
state during execution, and prevents competing requests from saving
extra maintenance work.
> - Ordinary and unauthorized wakes remain blocked.

## Linked Issues or Issue Description

Refs #19378, which introduced the existing Compact now action. Refs
#19377, whose review identified this independent queue admission gap.
Searches for `context_compaction` with `waiting` and `Compact now` with
`waiting` found no existing fix for this admission bug.

**What happened?**

An owner clicks Compact now for a waiting Agent Chat conversation. The
real compaction service reaches the heartbeat queue, receives no run,
and reports that compaction did not start.

**Expected behavior**

The authenticated owner can request one separate compaction run while
the chat waits. The request must keep its existing actor, task, agent,
budget, and execution checks. Ordinary waiting-chat wakes must stay
blocked.

**Steps to reproduce**

1. Use an open Agent Chat conversation whose agent supports compaction.
2. Wait until the conversation is waiting for its owner.
3. Request Compact now from the task Properties panel.
4. Observe the refusal before this fix. The new database regression
calls the real compaction service and queue to reproduce it.

## What Changed

- Admit only the existing complete manual compaction envelope. Require a
user actor, no coalescing, the compaction intent, and matching nonempty
request IDs.
- Preserve the waiting conversation state, session generation, reply
boundary and unanswered questions in the existing conversation
preparation owner.
- Recheck active and deferred maintenance under the existing issue
admission lock. Competing clicks cannot save a second maintenance
request.
- Add execution coverage with and without a pending question. Add
concurrent admission and deferred-request controls, including admission
after the pending receipt is cancelled. Keep the existing authorization
and ordinary conversation controls.

## Verification

- The original service-to-queue request failed before the admission fix.
- The execution regression then caught the review finding: compaction
succeeded, but both waiting-state controls failed. The same complete
controls passed after the conversation preparation fix.
- A real concurrent request test recorded two durable maintenance
receipts before the atomic admission fix. It counts deferred authority
as well as admitted runs.
- Current integrated candidate: normal current Runner core and surface
TypeScript exports passed. All seven normal App typecheck programs
passed. The whole queue/execution suites passed 86/86, the whole
context-compaction suite passed 5/5, and 11 selected ordinary
conversation controls passed. There were 102 actual passes; the 29
unselected conversation cases are not counted as passes.
- Database controls use fresh dedicated PostgreSQL 17 databases, Node
24, one test worker and pools capped at four. Private fixture adapters
use the existing dedicated database and normal migrations. They keep
test bodies, assertions and time limits intact.
- Earlier fixture and setup failures remain recorded separately from
product failures. These include an omitted external URL, an invalid
closed-task fixture, a missing crypto import, an incorrect
result-envelope matcher, and an unavailable test-library helper. Each
was corrected at its owning boundary. A test-output text matcher also
missed Vitest's abbreviated failure; the raw assertion stack and failure
were retained.
- Local session tests exercise the actual admission, native Runner
lifecycle, accepted result and persisted conversation state through the
supported test backend. They do not prove live provider compaction, UI
rendering or deployed suspension.
- Exact-head remote CI passed: 59 successful checks, three intentional
skips and one neutral check. Greptile and Ponytail both reviewed this
head at 5/5 with no actionable findings. Ponytail explicitly marked it
Ship.
- Master advanced during review. A local merge evaluation preserves
every unrelated master change and all four PR files. The current App
programs and all three execution controls passed again on that projected
squash tree to check the workspace-finalization seam.

## Risks

The change affects admission and preparation for one maintenance
request. A broad exception could start ordinary work or reopen a waiting
chat. The request therefore keeps the complete existing manual envelope
and authorization checks. Conversation preparation still checks the
session generation. The existing issue lock owns duplicate admission.
There is no new timer, database migration or provider protocol.

## Model Used

OpenAI Codex based on GPT-6 managed the change and reviewed the shared
queue fix. OpenAI `gpt-6.1-sol` authored the focused regression
coverage. Both used reasoning, repository tools and local code
execution. The runtime did not expose the context-window size.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes (the
existing Compact now documentation remains accurate; no new
configuration is added)
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <[email protected]>

server/src/__tests__/heartbeat-context-compaction-execution.test.ts

@@ -0,0 +1,175 @@
+import { randomUUID } from "node:crypto";
+import { and, eq } from "drizzle-orm";
+import { agents, agentTaskSessions, authUsers, companyMemberships, createDb, heartbeatRuns,
+  issueComments, issueThreadInteractions, issues, nativeRunResults } from "@paperclipai/db";
+import type { NativeRunIdentity, NativeSession, NativeSessionBackend, PrpEvent } from "@paperclipai/paperclip-runner";
+import { describe, expect, it, vi } from "vitest";
+import { createRunDispatch } from "../modules/run-dispatch/index.js";
+import { heartbeatService, type HeartbeatServiceOptions } from "../services/heartbeat.js";
+import { contextCompactionService } from "../services/context-compaction.js";
+import { instanceSettingsService } from "../services/instance-settings.js";
+import { closeIdleWarmNativeSessionsForRestart } from "../services/native-runtime/native-session-executor.js";
+import { nativeRestartFixture, waitForCondition } from "./helpers/native-restart-fixture.js";
+import { getRecoveryPostgresSupport } from "./helpers/recovery-postgres-fixture.js";
+
+// Only this owner opts into its dedicated database. Importing the fixture core
+// avoids the default support facade's embedded-platform probe in that mode.
+const dedicatedDatabaseUrl = process.env.PAPERCLIP_CONTEXT_COMPACTION_EXECUTION_TEST_DATABASE_URL;
+const databaseOptions = dedicatedDatabaseUrl === undefined ? {} : { dedicatedDatabaseUrl };
+const support = await getRecoveryPostgresSupport(databaseOptions);
+
+describe.skipIf(!support.supported)("heartbeat context compaction execution", () => {
+  const fixture = nativeRestartFixture(databaseOptions);
+
+  async function conversation() {
+    const { companyId, agentId, temporary } = fixture.current;
+    const db = createDb(temporary.connectionString, { maxConnections: 4 });
+    const userId = randomUUID();
+    await db.insert(authUsers).values({ id: userId, name: "Conversation owner", email: `${userId}@example.invalid`,
+      createdAt: new Date(), updatedAt: new Date() });
+    await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, membershipRole: "owner" });
+    await db.update(agents).set({ status: "idle", adapterConfig: { provider: "codex", lifecycleMode: "warm", idleTimeoutMs: 60_000 } })
+      .where(eq(agents.id, agentId));
+    await instanceSettingsService(db).updateExperimental({ enableAgentChat: true, enableNativeRunner: true });
+    const [issue] = await db.insert(issues).values({ companyId, title: "Keep the conversation parked", status: "in_progress",
+      assigneeAgentId: agentId, responsibleUserId: userId, conversationAgentId: agentId, conversationUserId: userId,
+      conversationState: "active" }).returning();
+    return { db, companyId, agentId, userId, issue };
+  }
+
+  // The first ordinary conversation turn produces a real idle checkpoint and
+  // task-session pointer through the normal owner. Its final text is a provider
+  // event, not a seeded result. The second turn uses the existing Runner
+  // compaction protocol; result() deliberately never supplies an outcome.
+  function conversationBackend(db: ReturnType<typeof createDb>) {
+    let identity: NativeRunIdentity;
+    let sourceInstanceId = "";
+    let activeTurnId: string | null = null;
+    let nextTurn = Promise.withResolvers<{ compact: boolean; turnId: string }>();
+    const startTurn = vi.fn<NativeSession["startTurn"]>(async input => {
+      const turnId = randomUUID();
+      activeTurnId = turnId;
+      nextTurn.resolve({ compact: input.compact === true, turnId });
+      return { turnId };
+    });
+    const capabilities = { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true };
+    const session: NativeSession = {
+      identity: () => identity,
+      capabilities: async () => capabilities,
+      attachRun: async input => { identity = input.identity; },
+      startTurn,
+      async *events() {
+        const turn = await nextTurn.promise;
+        const event = (sourceSeq: number, eventType: PrpEvent["eventType"], payload: Record<string, unknown>): PrpEvent => ({
+          schema: "paperclip.prp.event.v1", sourceKind: "runner", sourceInstanceId,
+          sourceEventId: `${sourceInstanceId}:${identity.runId}:${sourceSeq}`, sourceSeq,
+          runId: identity.runId, normalizedSessionId: identity.sessionId, turnId: turn.turnId,
+          schemaVersion: 1, priority: 0, emittedAt: new Date().toISOString(), eventType, payload,
+        });
+        yield event(1, "turn.started", {});
+        yield turn.compact
+          ? event(2, "context.compacted", { schema: "paperclip.context.compacted.v1", compactionId: randomUUID(),
+              reason: "provider", preTokens: null, postTokens: null, sameSession: true })
+          : event(2, "item.completed", { kind: "agentMessage", channel: "final", text: "Ready for your next message." });
+        activeTurnId = null;
+        nextTurn = Promise.withResolvers<{ compact: boolean; turnId: string }>();
+        yield event(3, "turn.completed", { status: "completed" });
+      },
+      result: async () => null,
+      snapshot: async () => ({ backendKind: "mock", sessionId: "compaction-session", identity,
+        providerSessionId: "compaction-provider-session", cursor: null, activeTurnId, pendingRuntimeRequests: [], lineage: [] }),
+      close: vi.fn(async () => {}),
+    };
+    const factory: NonNullable<HeartbeatServiceOptions["nativeSessionBackendFactory"]> = execution => {
+      const backend: NativeSessionBackend = {
+        descriptor: async () => {
+          const [run] = await db.select({ runnerInstanceId: heartbeatRuns.runnerInstanceId }).from(heartbeatRuns)
+            .where(eq(heartbeatRuns.id, execution.binding.runId));
+          if (!run?.runnerInstanceId) throw new Error("Normal native admission did not persist a runner identity");
+          sourceInstanceId = run.runnerInstanceId;
+          return { kind: "mock", name: "conversation-compaction", version: "1", capabilities,
+            runtimeContextCapabilities: { instructions: "native", skills: "native", mcp: "native" } };
+        },
+        openSession: async input => { identity = input.identity; return session; },
+        recoverSession: async snapshot => { identity = snapshot.identity; return { recovered: true, session }; },
+      };
+      return backend;
+    };
+    return { factory, startTurn, session };
+  }
+
+  it.each([false, true])("keeps a successfully compacted conversation waiting (pending question: %s)", async pendingQuestion => {
+    const f = await conversation();
+    const backend = conversationBackend(f.db);
+    const heartbeat = heartbeatService(f.db, { nativeSessionBackendFactory: backend.factory });
+    try {
+      const [message] = await f.db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issue.id,
+        authorType: "user", authorUserId: f.userId, body: "Start this conversation." }).returning();
+      const first = await heartbeat.wakeup(f.agentId, { source: "on_demand", triggerDetail: "manual", reason: "issue_commented",
+        requestedByActorType: "user", requestedByActorId: f.userId, manualUserWake: true,
+        payload: { issueId: f.issue.id, commentId: message.id },
+        contextSnapshot: { issueId: f.issue.id, taskKey: f.issue.id, wakeCommentId: message.id } });
+      expect(first).not.toBeNull();
+      await waitForCondition("normal conversation execution and wakes to settle", () => heartbeat.getTaskDrainStatus().quiescent);
+      expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, first!.id)))[0])
+        .toMatchObject({ status: "succeeded" });
+      const [idle] = await f.db.select().from(agentTaskSessions).where(and(eq(agentTaskSessions.companyId, f.companyId),
+        eq(agentTaskSessions.agentId, f.agentId), eq(agentTaskSessions.taskKey, f.issue.id)));
+      expect(idle).toMatchObject({ lastRunId: first!.id });
+      expect(idle.sessionParamsJson).toBeTruthy();
+      const [before] = await f.db.select().from(issues).where(eq(issues.id, f.issue.id));
+      expect(before).toMatchObject({ status: "in_review", conversationState: "waiting" });
+      const [question] = pendingQuestion ? await f.db.insert(issueThreadInteractions).values({
+        companyId: f.companyId, issueId: f.issue.id, createdByAgentId: f.agentId, sourceRunId: first!.id,
+        kind: "ask_user_questions", status: "pending", addresseeUserId: f.userId,
+        payload: { version: 1, questions: [{ id: "q", prompt: "Which option?", selectionMode: "single",
+          options: [{ id: "a", label: "Option A" }] }] },
+      }).returning() : [];
+      const commentsBefore = await f.db.select().from(issueComments).where(eq(issueComments.issueId, f.issue.id));
+      const interactionsBefore = await f.db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.issueId, f.issue.id));
+      const compact = await contextCompactionService(f.db, { wakeup: heartbeat.wakeup }).request({
+        companyId: f.companyId, issue: before, userId: f.userId,
+      });
+      await waitForCondition("normal compaction execution and wakes to settle", () => heartbeat.getTaskDrainStatus().quiescent);
+      const [run] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, compact.runId));
+      expect(run).toMatchObject({ status: "succeeded", contextSnapshot: { contextCompactionRequestId: compact.requestId } });
+      expect(backend.startTurn).toHaveBeenCalledTimes(2);
+      expect(backend.startTurn).toHaveBeenLastCalledWith(expect.objectContaining({ compact: true, continuation: true }));
+      expect(await f.db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, compact.runId)))
+        .toEqual([expect.objectContaining({ schemaStatus: "accepted", resultJson: expect.objectContaining({
+          result: expect.objectContaining({ summary: "Context compacted on request.", reportedWorkDisposition: "yielded" }),
+        }) })]);
+      expect((await f.db.select().from(issues).where(eq(issues.id, f.issue.id)))[0]).toMatchObject({
+        status: "in_review", conversationState: "waiting", conversationSessionGeneration: before.conversationSessionGeneration,
+        conversationBoundaryCommentId: before.conversationBoundaryCommentId,
+      });
+      expect(await f.db.select().from(issueComments).where(eq(issueComments.issueId, f.issue.id))).toEqual(commentsBefore);
+      expect(await f.db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.issueId, f.issue.id))).toEqual(interactionsBefore);
+      if (question) expect(question.status).toBe("pending");
+      expect(await heartbeat.wakeup(f.agentId, { source: "automation", triggerDetail: "system", reason: "issue.continuation_recovery",
+        requestedByActorType: "system", payload: { issueId: f.issue.id }, contextSnapshot: { issueId: f.issue.id } })).toBeNull();
+      expect(backend.startTurn).toHaveBeenCalledTimes(2);
+    } finally {
+      await heartbeat.drainActiveRunExecutions();
+      await closeIdleWarmNativeSessionsForRestart();
+      await f.db.$client.end();
+    }
+  }, 60_000);
+
+  it("still cancels an ordinary continuation while its question awaits an authorized answer", async () => {
+    const f = await conversation();
+    try {
+      await f.db.update(issues).set({ status: "in_review", conversationState: "waiting" }).where(eq(issues.id, f.issue.id));
+      await f.db.insert(issueThreadInteractions).values({ companyId: f.companyId, issueId: f.issue.id,
+        createdByAgentId: f.agentId, addresseeUserId: f.userId, kind: "ask_user_questions", status: "pending",
+        payload: { version: 1, questions: [{ id: "q", prompt: "Continue?", selectionMode: "single", options: [{ id: "a", label: "Yes" }] }] } });
+      const [run] = await f.db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId,
+        status: "queued", runtimeMode: "legacy", invocationSource: "automation", triggerDetail: "system", responsibleUserId: f.userId,
+        resultJson: { conversationContinuation: "continue_conversation_v1" }, contextSnapshot: { issueId: f.issue.id } }).returning();
+      const result = await createRunDispatch(f.db).cancelStaleQueuedRun({ runId: run.id, companyId: f.companyId, expectedStatus: "queued" });
+      expect(result).toMatchObject({ outcome: "cancelled", errorCode: "issue_waiting_for_response" });
+      expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0])
+        .toMatchObject({ status: "cancelled", errorCode: "issue_waiting_for_response" });
+    } finally { await f.db.$client.end(); }
+  });
+});

server/src/services/agent-conversations.ts

@@ -1,3 +1,4 @@
+import { isContextCompactionRun } from "./context-compaction.js";
 import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
 import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
@@ -150,6 +151,23 @@ export async function prepareConversationTurn(
       .for("update");
     if (!isConversation(issue))
       return { context, reset: false, conversation: false };
+    let generation = issue.conversationSessionGeneration;
+    if (
+      typeof context.conversationSessionGeneration === "number" &&
+      context.conversationSessionGeneration !== generation
+    ) {
+      throw new Error(
+        "Conversation session changed; this older turn cannot resume",
+      );
+    }
+    // Compaction maintains this session; it is not a new conversation turn.
+    // Keep the parked state and unanswered interactions while carrying the
+    // same session-generation context into the normal native runtime.
+    if (isContextCompactionRun(context)) {
+      const next = { ...context, conversationSessionGeneration: generation, conversationMode: true };
+      await tx.update(heartbeatRuns).set({ contextSnapshot: next }).where(eq(heartbeatRuns.id, run.id));
+      return { context: next, reset: false, conversation: true };
+    }
     const commentId =
       typeof context.wakeCommentId === "string"
         ? context.wakeCommentId
@@ -171,15 +189,6 @@ export async function prepareConversationTurn(
     const reset = Boolean(
       comment && comment.authorUserId && isConversationReset(comment.body),
     );
-    let generation = issue.conversationSessionGeneration;
-    if (
-      typeof context.conversationSessionGeneration === "number" &&
-      context.conversationSessionGeneration !== generation
-    ) {
-      throw new Error(
-        "Conversation session changed; this older turn cannot resume",
-      );
-    }
     // The boundary lives on the command comment. A crash/retry reuses it instead of resetting twice.
     if (reset && comment && comment.conversationSessionGeneration == null) {
       await notifyChatRunMilestones(tx, { companyId: issue.companyId, issueId: issue.id });

server/src/services/heartbeat/queue.test.ts

@@ -1,3 +1,5 @@
+import { randomUUID } from "node:crypto";
+import { contextCompactionService } from "../context-compaction.js";
 import { claimNativeSessionResumptions } from "../native-runtime/native-finalization-reconciler.js";
 import type { Db } from "@paperclipai/db";
 import { authUsers, agents, agentWakeupRequests, caseHandoffs, cases, chatCompletionDeliveries, companies, companyMemberships, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, environmentLeases, heartbeatRuns, issueComments, issueThreadInteractions, issues, nativeRunFinalizations } from "@paperclipai/db";
@@ -5,11 +7,12 @@ import { eq, inArray, sql } from "drizzle-orm";
 import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
 import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
-import { createPostgresWakeQueueAdapter } from "../../modules/wake-queue/adapters/postgres.js";
+import { createPostgresWakeQueueAdapter, createWakeAdmissionWriter } from "../../modules/wake-queue/adapters/postgres.js";
 import { createReleaseIssueExecution } from "../../modules/wake-queue/application/use-cases.js";
-import { createHeartbeatQueue, shouldDeferFollowupWakeForSameIssue, shouldQueueFollowupForRunningIssueWake, type HeartbeatQueueDependencies } from "./queue.js";
+import { createHeartbeatQueue, DEFERRED_WAKE_CONTEXT_KEY, shouldDeferFollowupWakeForSameIssue, shouldQueueFollowupForRunningIssueWake, type HeartbeatQueueDependencies } from "./queue.js";
 import { createWakeQueue } from "../../modules/wake-queue/index.js";
-import { mergeCoalescedContextSnapshot } from "./run-preparation.js";
+import { TransactionScope } from "../../modules/wake-queue/application/ports.js";
+import { enrichWakeContextSnapshot, mergeCoalescedContextSnapshot } from "./run-preparation.js";
 import { withQueuedCommentIdsInWakePayload, withQueuedCommentIdsInRunContext } from "../issue-queued-comment-queue.js";
 import { settleConversationTurn } from "../agent-conversations.js";
 import { chatCompletionDeliveryService, recordChatCompletion, recordChatHandoff } from "../chat-completion-delivery.js";
@@ -117,7 +120,8 @@ describe.skipIf(!support.supported)("heartbeat queue database wiring", () => {
     expect(deps.budgets.getInvocationBlock).not.toHaveBeenCalled();
   });
 
-  it("does not admit recovery after the conversation settles between precheck and admission", async () => {
+  it.each(["ordinary recovery", "mismatched compaction"] as const)(
+    "does not admit %s after the conversation settles between precheck and admission", async cause => {
     const [user] = await db.insert(authUsers).values({
       id: "queue-conversation-race-user", name: "Conversation owner", email: "[email protected]",
       createdAt: new Date(), updatedAt: new Date(),
@@ -157,10 +161,16 @@ describe.skipIf(!support.supported)("heartbeat queue database wiring", () => {
       await admissionMayContinue;
       return null;
     });
+    const requestId = randomUUID();
+    const compaction = cause === "mismatched compaction";
     const pending = createHeartbeatQueue(db, deps).enqueueWakeup(agent.id, {
-      source: "automation", triggerDetail: "system", reason: "issue.continuation_recovery",
-      requestedByActorType: "system", payload: { issueId: issue.id },
-      contextSnapshot: { issueId: issue.id, retryOfRunId: completed.id, source: "issue.continuation_recovery" },
+      source: compaction ? "on_demand" : "automation", triggerDetail: compaction ? "manual" : "system",
+      reason: compaction ? "context_compaction" : "issue.continuation_recovery",
+      requestedByActorType: compaction ? "user" : "system", requestedByActorId: compaction ? user.id : undefined,
+      ...(compaction ? { manualUserWake: true, allowRunCoalescing: false } : {}),
+      payload: { issueId: issue.id, ...(compaction ? { intent: "context_compaction", requestId: `foreign-${requestId}` } : {}) },
+      contextSnapshot: { issueId: issue.id, retryOfRunId: completed.id, source: "issue.continuation_recovery",
+        ...(compaction ? { contextCompactionRequestId: requestId } : {}) },
     });
     try {
       await atBarrier;
@@ -196,6 +206,192 @@ describe.skipIf(!support.supported)("heartbeat queue database wiring", () => {
     return { user, company, agent, issue, deps, queue: createHeartbeatQueue(db, deps) };
   }
 
+  async function waitingCompactionFixture() {
+    const f = await waitingConversationFixture();
+    const [agent] = await db.update(agents).set({ adapterType: "paperclip_runner", adapterConfig: { provider: "codex" } })
+      .where(eq(agents.id, f.agent.id)).returning();
+    if (!agent) throw new Error("Missing compaction agent");
+    f.deps.getAgent.mockResolvedValue(agent);
+    const service = contextCompactionService(db, { wakeup: f.queue.enqueueWakeup });
+    return { ...f, agent, service };
+  }
+
+  it("admits the owner's context compaction maintenance request while its conversation waits", async () => {
+    const f = await waitingCompactionFixture();
+    const accepted = await f.service.request({ companyId: f.company.id, issue: f.issue, userId: f.user.id });
+    const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, f.agent.id));
+    expect(runs).toHaveLength(1);
+    expect(runs[0]).toMatchObject({ id: accepted.runId, status: "queued", contextSnapshot: {
+      issueId: f.issue.id, contextCompactionRequestId: accepted.requestId, skipIssueComment: true,
+    } });
+    expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, f.agent.id)))
+      .toHaveLength(1);
+    expect((await db.select().from(issues).where(eq(issues.id, f.issue.id)))[0])
+      .toMatchObject({ status: "in_review", conversationState: "waiting", assigneeAgentId: f.agent.id });
+    expect(await db.select().from(issueComments).where(eq(issueComments.issueId, f.issue.id))).toHaveLength(0);
+    expect(f.deps.executeRun).not.toHaveBeenCalled();
+  });
+
+  it("admits only one of two simultaneous waiting-conversation compaction requests", async () => {
+    const f = await waitingCompactionFixture();
+    // Both service requests must finish their real availability read before
+    // either enters queue admission. Dispatch remains suppressed, not mocked.
+    // task_drain explicitly retains admission (queue.ts's admission guard
+    // excludes it), while startNextQueuedRunForAgent stops before any claim.
+    // Do not tie two admissions to a call-count-sensitive suppression sequence.
+    f.deps.getSchedulingSuppression.mockReset().mockResolvedValue({ suppressed: true, reason: "task_drain" });
+    const admission = createWakeQueue(db, {
+      resolveResponsibleUserId: async () => f.user.id,
+      getRoutineEnv: async () => ({ routineId: null, env: null, responsibleUserId: null }),
+      resolveSessionBeforeForWakeup: async () => null,
+      wakeAdmissionHelpers: { filterZombieCoalesceTarget: f.deps.filterZombieCoalesceTarget,
+        mergeCoalescedContextSnapshot, shouldDeferFollowupWakeForSameIssue, shouldQueueFollowupForRunningIssueWake },
+      recovery: { escalateStrandedAssignedIssue: async () => undefined, escalateStrandedRecoveryIssueInPlace: async () => undefined },
+    });
+    f.deps.wakeQueue.createAdmissionTransactionScope.mockImplementation(admission.createAdmissionTransactionScope);
+    f.deps.wakeQueue.admitWakeBehindIssueExecution.mockImplementation(admission.admitWakeBehindIssueExecution);
+    let releaseAvailability!: () => void;
+    const bothAvailable = new Promise<void>(resolve => { releaseAvailability = resolve; });
+    let arrivals = 0;
+    const service = contextCompactionService(db, { wakeup: async (...args) => {
+      if (++arrivals === 2) releaseAvailability();
+      await bothAvailable;
+      return f.queue.enqueueWakeup(...args);
+    } });
+    const results = await Promise.allSettled([0, 1].map(() => service.request({
+      companyId: f.company.id, issue: f.issue, userId: f.user.id,
+    })));
+    // Inspect both forms of durable execution authority before asserting.
+    // A saved deferred maintenance request can replay after the first run;
+    // counting only currently admitted runs would miss that extra work.
+    const maintenanceRuns = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, f.agent.id));
+    const receipts = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, f.agent.id));
+    const maintenanceReceipts = receipts.filter(row => row.status !== "skipped"
+      && row.payload?.intent === "context_compaction" && row.payload.issueId === f.issue.id);
+    expect(arrivals).toBe(2);
+    expect(maintenanceReceipts.map(row => ({ id: row.id, status: row.status, runId: row.runId })),
+      "one durable compaction authority including deferred").toHaveLength(1);
+    expect(maintenanceRuns.map(run => run.id)).toHaveLength(1);
+    expect(maintenanceRuns[0]).toMatchObject({ companyId: f.company.id, contextSnapshot: { issueId: f.issue.id } });
+    expect(results.filter(result => result.status === "fulfilled")).toHaveLength(1);
+    const rejected = results.filter(result => result.status === "rejected");
+    expect(rejected).toHaveLength(1);
+    if (rejected[0]?.status === "rejected") expect(rejected[0].reason).toMatchObject({ code: "already_pending" });
+    expect(receipts.filter(row => row.status === "deferred_issue_execution")).toEqual([]);
+    expect(receipts.filter(row => row.reason === "context_compaction" && row.status !== "skipped")).toHaveLength(1);
+    expect(f.deps.executeRun).not.toHaveBeenCalled();
+  });
+
+  it("preserves an existing deferred compaction authority until cancellation allows a new request", async () => {
+    const f = await waitingCompactionFixture();
+    f.deps.getSchedulingSuppression.mockReset().mockResolvedValue({ suppressed: true, reason: "task_drain" });
+    // Reproduce a pre-existing receipt through the actual normal deferred
+    // writer. In particular, its informational reason is NOT context_compaction.
+    // The service builds the real request; no provider/result rows are seeded.
+    const historical = contextCompactionService(db, { wakeup: async (agentId, options) => {
+      const source = options.source ?? "on_demand";
+      const triggerDetail = options.triggerDetail ?? null;
+      const context = enrichWakeContextSnapshot({ contextSnapshot: { ...options.contextSnapshot },
+        reason: options.reason ?? null, source, triggerDetail, payload: options.payload ?? null });
+      await db.transaction(async tx => {
+        await createWakeAdmissionWriter().insertNewDeferredWake(TransactionScope.create(f.company.id, tx), {
+          companyId: f.company.id, agentId, source, triggerDetail,
+          payload: { ...options.payload, manualUserWake: true, [DEFERRED_WAKE_CONTEXT_KEY]: context.contextSnapshot },
+          requestedByActorType: options.requestedByActorType ?? null,
+          requestedByActorId: options.requestedByActorId ?? null, idempotencyKey: options.idempotencyKey ?? null,
+        });
+      });
+      return null;
+    } });
+    await expect(historical.request({ companyId: f.company.id, issue: f.issue, userId: f.user.id }))
+      .rejects.toMatchObject({ code: "wake_skipped" });
+    const [pending] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, f.agent.id));
+    expect(pending).toMatchObject({ reason: "issue_execution_deferred", status: "deferred_issue_execution", runId: null,
+      payload: { issueId: f.issue.id, intent: "context_compaction" } });
+    await expect(f.service.request({ companyId: f.company.id, issue: f.issue, userId: f.user.id }))
+      .rejects.toMatchObject({ code: "already_pending" });
+    expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, f.agent.id))).toEqual([]);
+    expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, f.agent.id))).toEqual([pending]);
+    // This never-dispatched fixture authority is now cancelled, not completed.
+    // Nonpending history must not permanently block a fresh normal request.
+    await db.update(agentWakeupRequests).set({ status: "cancelled", finishedAt: new Date() })
+      .where(eq(agentWakeupRequests.id, pending.id));
+    const accepted = await f.service.request({ companyId: f.company.id, issue: f.issue, userId: f.user.id });
+    expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, f.agent.id)))
+      .toEqual([expect.objectContaining({ id: accepted.runId, status: "queued" })]);
+    const receipts = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, f.agent.id));
+    expect(receipts).toHaveLength(2);
+    expect(receipts.find(row => row.id === pending.id)).toMatchObject({ status: "cancelled", runId: null });
+    expect(receipts.find(row => row.runId === accepted.runId)).toMatchObject({ status: "queued" });
+    expect(f.deps.executeRun).not.toHaveBeenCalled();
+  });
+
+  it.each([
+    { actor: "missing actor", userId: "", message: "Manual wake requires an authenticated user" },
+    { actor: "foreign owner", userId: "another-conversation-owner", message: "Only the conversation owner can start a chat run" },
+  ])("refuses waiting-conversation compaction from a $actor", async ({ userId, message }) => {
+    const f = await waitingCompactionFixture();
+    await expect(f.service.request({ companyId: f.company.id, issue: f.issue, userId })).rejects.toThrow(message);
+    expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, f.agent.id))).toHaveLength(0);
+    expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, f.agent.id))).toHaveLength(0);
+    expect(f.deps.executeRun).not.toHaveBeenCalled();
+  });
+
+  it.each(["missing request", "malformed request", "mismatched request"] as const)(
+    "rejects a waiting-conversation compaction envelope with a %s", async kind => {
+      const f = await waitingCompactionFixture();
+      const service = contextCompactionService(db, { wakeup: (agentId, options) => {
+        const contextSnapshot = { ...options.contextSnapshot };
+        const payload = { ...options.payload };
+        if (kind === "missing request") delete contextSnapshot.contextCompactionRequestId;
+        if (kind === "malformed request") payload.requestId = { unexpected: true };
+        if (kind === "mismatched request") payload.requestId = randomUUID();
+        return f.queue.enqueueWakeup(agentId, { ...options, contextSnapshot, payload });
+      } });
+      await expect(service.request({ companyId: f.company.id, issue: f.issue, userId: f.user.id }))
+        .rejects.toMatchObject({ code: "wake_skipped" });
+      expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, f.agent.id))).toHaveLength(0);
+      expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, f.agent.id))).toHaveLength(0);
+      expect((await db.select().from(issues).where(eq(issues.id, f.issue.id)))[0])
+        .toMatchObject({ status: "in_review", conversationState: "waiting" });
+      expect(f.deps.executeRun).not.toHaveBeenCalled();
+    });
+
+  it("does not turn an ordinary waiting-conversation wake into compaction authority", async () => {
+    const f = await waitingCompactionFixture();
+    expect(await f.queue.enqueueWakeup(f.agent.id, { source: "on_demand", triggerDetail: "manual",
+      reason: "issue.continuation_recovery", requestedByActorType: "user", requestedByActorId: f.user.id,
+      manualUserWake: true, allowRunCoalescing: false, payload: { issueId: f.issue.id },
+      contextSnapshot: { issueId: f.issue.id } })).toBeNull();
+    expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, f.agent.id))).toHaveLength(0);
+    expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, f.agent.id))).toHaveLength(0);
+    expect(f.deps.executeRun).not.toHaveBeenCalled();
+  });
+
+  it("keeps closed tasks ineligible for a compaction request", async () => {
+    const f = await waitingCompactionFixture();
+    const [closed] = await db.insert(issues).values({ companyId: f.company.id, title: "Closed task",
+      status: "done", assigneeAgentId: f.agent.id }).returning();
+    if (!closed) throw new Error("Missing closed task");
+    await expect(f.service.request({ companyId: f.company.id, issue: closed, userId: f.user.id }))
+      .rejects.toMatchObject({ code: "task_closed" });
+    expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, f.agent.id))).toHaveLength(0);
+    expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, f.agent.id))).toHaveLength(0);
+  });
+
+  it("keeps a busy conversation ineligible for a second compaction run", async () => {
+    const f = await waitingCompactionFixture();
+    const [running] = await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: f.agent.id,
+      issueId: f.issue.id, scopeKind: "issue", status: "running" }).returning();
+    if (!running) throw new Error("Missing active conversation run");
+    await expect(f.service.request({ companyId: f.company.id, issue: f.issue, userId: f.user.id }))
+      .rejects.toMatchObject({ code: "agent_busy" });
+    expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, f.agent.id))).map(run => run.id))
+      .toEqual([running.id]);
+    expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, f.agent.id))).toHaveLength(0);
+    expect(f.deps.executeRun).not.toHaveBeenCalled();
+  });
+
   it.each(["comment", "interaction", "ordinary task"] as const)("preserves admission for a current %s", async cause => {
     const f = await waitingConversationFixture();
     const context: Record<string, unknown> = { issueId: f.issue.id, source: "issue.continuation_recovery" };

server/src/services/heartbeat/queue.ts

@@ -1,3 +1,4 @@
+import { CONTEXT_COMPACTION_WAKE_REASON, ContextCompactionError, isContextCompactionRun, readContextCompactionRequest } from "../context-compaction.js";
 import { createHeartbeatTimerPolicy } from "./timer-policy.js";
 import { notifyExecutionRunWork } from "../execution-control-notifications.js";
 import { PersistedExecutionWaitRefusal, writeDeferredExecutionWait } from "../deferred-execution-wait.js";
@@ -2186,10 +2187,18 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
       triggerDetail,
       payload,
     });
+    const isManualContextCompactionWake = () => Boolean(
+      reason === CONTEXT_COMPACTION_WAKE_REASON && source === "on_demand" && triggerDetail === "manual"
+      && opts.manualUserWake === true && opts.requestedByActorType === "user"
+      && Boolean(opts.requestedByActorId) && opts.allowRunCoalescing === false
+      && payload?.intent === CONTEXT_COMPACTION_WAKE_REASON
+      && readNonEmptyString(payload.requestId) !== null
+      && readContextCompactionRequest(enrichedContextSnapshot)?.requestId === payload.requestId,
+    );
     const hasIndependentConversationWakeCause = () => Boolean(
       wakeCommentId || opts.taskRetryRequestId ||
       hasInteractionContinuationWakeContext(enrichedContextSnapshot) ||
-      reason === CHAT_COMPLETION_WAKE_REASON,
+      reason === CHAT_COMPLETION_WAKE_REASON || isManualContextCompactionWake(),
     );
     // Keep each request's own server-derived origin, including coalesced wakes.
     // A run's merged context cannot establish which caller authored one receipt.
@@ -3729,6 +3738,28 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
             }
           }
 
+          // Availability is advisory: competing clicks may both read it before
+          // admission. The existing issue lock serializes the run and deferred
+          // receipt checks with the admission below, without another lock owner.
+          if (isManualContextCompactionWake()) {
+            if (activeExecutionRun) {
+              const compacting = isContextCompactionRun(activeExecutionRun.contextSnapshot);
+              throw new ContextCompactionError(compacting ? "already_pending" : "agent_busy",
+                compacting ? "The agent is already compacting its context." : "Available when the agent finishes its current run.");
+            }
+            const [pending] = await tx.select({ id: agentWakeupRequests.id }).from(agentWakeupRequests).where(and(
+              eq(agentWakeupRequests.companyId, issue.companyId), eq(agentWakeupRequests.agentId, agentId),
+              inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"]), isNull(agentWakeupRequests.runId),
+              executionWaitRequestId ? ne(agentWakeupRequests.id, executionWaitRequestId) : undefined,
+              sql`${agentWakeupRequests.payload}->>'issueId' = ${issue.id}`,
+              sql`${agentWakeupRequests.payload}->>'intent' = ${CONTEXT_COMPACTION_WAKE_REASON}`,
+              sql`jsonb_typeof(${agentWakeupRequests.payload}->'requestId') = 'string'`,
+              sql`nullif(btrim(${agentWakeupRequests.payload}->>'requestId'), '') is not null`,
+              sql`${agentWakeupRequests.payload}->${DEFERRED_WAKE_CONTEXT_KEY}->>'contextCompactionRequestId' = ${agentWakeupRequests.payload}->>'requestId'`,
+            )).limit(1);
+            if (pending) throw new ContextCompactionError("already_pending", "The agent is already compacting its context.");
+          }
+
           const dependencyReadiness = await issuesSvc
             .listDependencyReadiness(issue.companyId, [issue.id], tx)
             .then((rows) => rows.get(issue.id) ?? null);