diff --git a/server/src/__tests__/deferred-execution-receipt-work.test.ts b/server/src/__tests__/deferred-execution-receipt-work.test.ts index 56b0b057af3..61cee11a3fb 100644 --- a/server/src/__tests__/deferred-execution-receipt-work.test.ts +++ b/server/src/__tests__/deferred-execution-receipt-work.test.ts @@ -379,6 +379,99 @@ describe("deferred receipt durable dispatch", () => { expect(resume).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ companyId: f.companyId, requestId: wake.id })); await quiet(); }); + it("parks an accepted conversation Retry while chat is disabled and resumes it on enablement", async () => { + const settings = instanceSettingsService(db), original = await settings.getExperimental(); + await settings.updateExperimental({ enableAgentChat: true }); + try { + const f = await fixture(); + await db.update(issues).set({ status: "in_review", conversationAgentId: f.agent.id, + conversationUserId: f.actorId, conversationState: "waiting" }).where(eq(issues.id, f.issue.id)); + const request = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issue.id, + actorId: f.actorId, observedRunId: f.source.id, idempotencyKey: randomUUID() }); + expect(request.accepted).toBe(true); + await settings.updateExperimental({ enableAgentChat: false }); + await expect(requestTaskRetry({ db, companyId: f.companyId, issueId: f.issue.id, + actorId: f.actorId, observedRunId: f.source.id, idempotencyKey: randomUUID() })) + .rejects.toMatchObject({ status: 404, message: "Agent Chat is disabled" }); + await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready; + await heartbeat.registerRunDispatchWork(coordinator, () => !drain).ready; + await Promise.allSettled([...passes]); + expect(errors).toEqual([]); + const read = async () => (await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!)))[0]; + expect(await read()).toMatchObject({ status: "deferred_issue_execution", runId: null }); + await vi.waitFor(async () => { + await Promise.allSettled([...passes]); + expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 40_000 }); + const idleStatements = statements.length; + await new Promise(resolve => setTimeout(resolve, 31_000)); + expect(statements).toHaveLength(idleStatements); + await settings.updateExperimental({ enableAgentChat: true }); + let runId: string | null = null; + await vi.waitFor(async () => { + const saved = await read(); expect(saved.runId).toBeTruthy(); runId = saved.runId; + }, { timeout: 5_000 }); + await vi.waitFor(async () => { + await heartbeat.drainActiveRunExecutions(); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId!)); + expect(run.status).toBe("succeeded"); expect(run.stdoutExcerpt).toContain("receipt-proof"); + }, { timeout: 15_000 }); + expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${request.requestId}`)).toHaveLength(1); + await heartbeat.drainActiveRunExecutions(); + } finally { vi.useRealTimers(); await settings.updateExperimental({ enableAgentChat: original.enableAgentChat }); } + }); + it("resumes saved conversation input only after chat is re-enabled and its transaction commits", async () => { + const settings = instanceSettingsService(db), original = await settings.getExperimental(); + await settings.updateExperimental({ enableAgentChat: true }); + try { + const f = await fixture(); + await db.update(issues).set({ status: "in_review", conversationAgentId: f.agent.id, + conversationUserId: f.actorId, conversationState: "waiting" }).where(eq(issues.id, f.issue.id)); + const comment = await issueService(db).addComment(f.issue.id, "Continue with this saved input", { userId: f.actorId }); + const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id, + source: "on_demand", status: "deferred_issue_execution", reason: "issue_commented", + requestedByActorType: "user", requestedByActorId: f.actorId, updatedAt: new Date(Date.now() - 31_000), + payload: { issueId: f.issue.id, queuedCommentInterrupt: { actorId: f.actorId }, + _paperclipWakeContext: { issueId: f.issue.id, wakeCommentIds: [comment.id] } } }).returning(); + await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date(Date.now() - 31_000), + resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false }, + queuedCommentInterruptQueueId: wake.id } }).where(eq(heartbeatRuns.id, f.source.id)); + await settings.updateExperimental({ enableAgentChat: false }); + await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready; + await heartbeat.registerRunDispatchWork(coordinator, () => !drain).ready; + await vi.waitFor(async () => { + await Promise.allSettled([...passes]); + expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 5_000 }); + const idleStatements = statements.length; + await new Promise(resolve => setTimeout(resolve, 31_000)); + expect(statements).toHaveLength(idleStatements); + const read = async () => (await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)))[0]; + expect(await read()).toMatchObject({ status: "deferred_issue_execution", runId: null }); + await expect(db.transaction(async tx => { + await instanceSettingsService(tx as unknown as typeof db).updateExperimental({ enableAgentChat: true }); + throw new Error("rolled back chat enablement"); + })).rejects.toThrow("rolled back chat enablement"); + await Promise.allSettled([...passes]); + expect(await read()).toMatchObject({ status: "deferred_issue_execution", runId: null }); + expect((await settings.getExperimental()).enableAgentChat).toBe(false); + await settings.updateExperimental({ enableAgentChat: true }); + let runId: string | null = null; + await vi.waitFor(async () => { + const saved = await read(); expect(saved.runId).toBeTruthy(); runId = saved.runId; + }, { timeout: 5_000 }); + await vi.waitFor(async () => { + await heartbeat.drainActiveRunExecutions(); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId!)); + expect(run.status).toBe("succeeded"); expect(run.stdoutExcerpt).toContain("receipt-proof"); + }, { timeout: 15_000 }); + const runs = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + sql`${heartbeatRuns.id} <> ${f.source.id}`)); + expect(runs).toHaveLength(1); + expect(runs[0].contextSnapshot?.wakeCommentIds).toEqual([comment.id]); + await quiet(); + } finally { vi.useRealTimers(); await settings.updateExperimental({ enableAgentChat: original.enableAgentChat }); } + }); it("settles a processed chat receipt without restaging it until an actual eligibility commit", async () => { const f = await fixture(); await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready; diff --git a/server/src/services/heartbeat/run-control.ts b/server/src/services/heartbeat/run-control.ts index a410d905d6a..06f0976cb4e 100644 --- a/server/src/services/heartbeat/run-control.ts +++ b/server/src/services/heartbeat/run-control.ts @@ -1,3 +1,4 @@ +import { instanceSettingsService } from "../instance-settings.js"; import { writeDeferredExecutionWait } from "../deferred-execution-wait.js"; import type { DeferredReceiptRecoveryOptions } from "../deferred-execution-notifications.js"; import { observeStoppedWorkspaceRecovery } from "../workspace-recovery.js"; @@ -777,7 +778,8 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont async function resumeExecutionWaitComments(scope: DeferredReceiptRecoveryOptions = {}) { if ((await getSchedulingSuppression()).suppressed) return; await reconcileTaskChatRetryReceipts(db, scope); - const retries = await db.select({ id: agentWakeupRequests.id, companyId: agentWakeupRequests.companyId }) + const retries = await db.select({ id: agentWakeupRequests.id, companyId: agentWakeupRequests.companyId, + conversationAgentId: issues.conversationAgentId }) .from(agentWakeupRequests).innerJoin(issues, and(eq(issues.companyId, agentWakeupRequests.companyId), sql`${issues.id}::text = ${agentWakeupRequests.payload}->>'issueId'`, eq(issues.assigneeAgentId, agentWakeupRequests.agentId))) @@ -788,7 +790,12 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont sql`${agentWakeupRequests.payload}->'taskRetry'->>'chatActionId' is null`, notInArray(issues.status, ["done", "cancelled"]))) .orderBy(asc(agentWakeupRequests.updatedAt)).limit(50); + const chatEnabled = !retries.some(retry => retry.conversationAgentId) + || (await instanceSettingsService(db).getExperimental()).enableAgentChat; for (const retry of retries) { + // Accepted chat work waits for the flag's committed eligibility signal. + // Public Retry still rejects new requests while the feature is disabled. + if (retry.conversationAgentId && !chatEnabled) continue; await db.update(agentWakeupRequests).set({ updatedAt: new Date() }).where(eq(agentWakeupRequests.id, retry.id)); await resumeTaskRetry(retry.companyId, retry.id).catch(err => { scope.onError?.(err); logger.warn({ err, requestId: retry.id }, "Task retry dispatch deferred"); diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 6c5f948d303..488e909c353 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -625,8 +625,11 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService } } } + const chatBecameEnabled = !toExperimentalView(current.experimental).enableAgentChat + && toExperimentalView(nextExperimental).enableAgentChat; const now = new Date(); const [updated] = await db.transaction(async tx => { + if (chatBecameEnabled) await notifyDeferredExecutionEligibility(tx); await notifyDeliveryWork(tx, DELIVERY_QUEUES.mcpEvents, "*"); await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, "*"); await notifyDeliveryWork(tx, DELIVERY_QUEUES.grokDispatch);