diff --git a/server/src/__tests__/heartbeat-context-compaction-execution.test.ts b/server/src/__tests__/heartbeat-context-compaction-execution.test.ts new file mode 100644 index 0000000000..8db8a9530a --- /dev/null +++ b/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) { + let identity: NativeRunIdentity; + let sourceInstanceId = ""; + let activeTurnId: string | null = null; + let nextTurn = Promise.withResolvers<{ compact: boolean; turnId: string }>(); + const startTurn = vi.fn(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): 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 = 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(); } + }); +}); diff --git a/server/src/services/agent-conversations.ts b/server/src/services/agent-conversations.ts index f84fc0aa71..ec108aca3e 100644 --- a/server/src/services/agent-conversations.ts +++ b/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 }); diff --git a/server/src/services/heartbeat/queue.test.ts b/server/src/services/heartbeat/queue.test.ts index 9e87edb1c0..9d504c6587 100644 --- a/server/src/services/heartbeat/queue.test.ts +++ b/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: "queue-race@paperclip.test", 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(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 = { issueId: f.issue.id, source: "issue.continuation_recovery" }; diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 4bbbbdb5dc..d5f85b0794 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/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);