From 6fae6f2cb84d20ee1c227ae88ebe566f5040d752 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Wed, 14 Oct 2026 18:18:26 -0700 Subject: [PATCH] fix(wake-queue): preserve pending work when a queued run is stopped (#16357) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The wake queue releases task locks and can promote saved follow-up work when a run ends. > - An operator can stop a queued run before dispatch selects its runtime. > - The legacy release path can promote the old follow-up after that Stop. > - For an obsolete routine cycle, this can conflict with the current active cycle after cancellation has already committed. > - This pull request preserves the deferred input without promoting it at that release boundary. > - The benefit is that a queued Stop stays stopped and does not return an error from starting an obsolete follow-up. ## Linked Issues or Issue Description Related: Refs #14300 and Refs #16339 describe other pre-dispatch cancellation paths. Related PR #16238 changes former-owner wake selection. This change addresses expected operator cancellation before legacy runtime selection, without changing those contracts. **What happened?** Stopping an unstarted legacy run committed its cancellation, then tried to promote a deferred wake. When another cycle of the same routine was already active, the issue lock update failed against `issues_open_routine_execution_uq`. The cancel request returned HTTP 500 even though the run was cancelled. Without a competing cycle, the Stop launched the old follow-up. **What did you expect to happen?** Release the stopped run's own locks and preserve deferred input for a later eligible instruction. Do not start the old queue as a side effect of Stop. **Steps to reproduce** Create two cycles of the same routine. Keep the current cycle active. Give the old cycle an unstarted legacy run with a verified bootstrap record and a deferred wake. Cancel that run as a board operator, then release its issue execution. The added database regression reproduces the failed lock update before the fix. **Paperclip version / commit** Reproduced on parent commit `976cdc77ed564dc3e014621505890b647e554a7f`. **Deployment mode** A remote runtime exposed the symptom. The regression uses embedded PostgreSQL and the real wake-queue adapter. ## What Changed - Recognize an expected operator Stop only when the legacy run has no runtime-selection time, start time, process-start time, or process identity. - Use the existing release boundary to preserve deferred input without promoting it. - Add database regressions for ordinary Stop, a competing routine cycle, repeated release, provider cancellation, runtime selection, and dispatched-run reconciliation. - Document the behavior and update both recovery maps. ## Verification - Before the fix, both queued Stop cases failed. The competing-cycle case raised `issues_open_routine_execution_uq`. The three control cases passed. - `node node_modules/vitest/vitest.mjs run --project @paperclipai/server server/src/modules/wake-queue`: 7 files and 214 tests passed, including all five new database cases. - `node scripts/check-maps.mjs` and `git diff --check` passed. - `pnpm -r typecheck`, `pnpm build`, and all 10 map-script tests passed locally. The complete CI suite is green on `8e204eef87f1a593df5c183abef764e4b9eb5fd1`. The remaining serial local full-suite run was stopped once that unchanged-head CI verification passed; a local full-suite pass is not claimed. Apex review passed at 5/5 on the same head with no review threads. ## Risks - The change is limited to recorded operator cancellation before legacy runtime selection. Selected runtimes, dispatched work, provider cancellation, and native stop acknowledgement keep their existing paths. - Deferred work is retained. A later eligible instruction must still use normal admission and all current gates. - No schema migration or new authority is required. This regression is not proof of a live provider stop. ## Model Used OpenAI GPT-6 through Codex, with reasoning, tool use, and code execution. The exact serving snapshot ID and context window size are not exposed in this session. ## 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 - [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 --- doc/execution-semantics.md | 7 ++ maps/agent-access-map/recovery.md | 7 ++ maps/feature-map/recovery.md | 6 ++ .../modules/wake-queue/adapters/postgres.ts | 8 +- .../queued-operator-stop-postgres.test.ts | 95 +++++++++++++++++++ 5 files changed, 122 insertions(+), 1 deletion(-) create mode 100644 server/src/modules/wake-queue/adapters/queued-operator-stop-postgres.test.ts diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index d008805a395..6dff82acc7b 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -212,6 +212,13 @@ The active-lock lifecycle is part of the checkout contract: Stale-lock recovery is crash recovery, not a retry loop. Paperclip must not clear or adopt locks held by non-terminal runs. After stale cleanup, a checkout `409` should mean a real live owner, status/assignee mismatch, unresolved blocker, or active gate still prevents checkout. Agents must treat that `409` as an ownership conflict and stop rather than retrying the same checkout. +Stopping a queued legacy run before runtime selection clears only its own task +locks. It preserves deferred messages without promoting another run. The stop +must have a recorded, expected operator cancellation and no runtime selection, +start timestamp, or process identity. A later eligible instruction can adopt the +saved messages through normal admission. This rule does not prove a dispatched +provider has stopped or change native cancellation acknowledgement. + ### Known execution waits at admission A known execution hold is a waiting condition, not a new execution attempt. Every issue wake must read the current effective reconciliation hold under the issue admission lock before creating a run. Resolved recovery bookkeeping can still carry a no-replay hold; only clearing the effective hold makes admission eligible again. The final dispatch gate remains required for changes after admission. diff --git a/maps/agent-access-map/recovery.md b/maps/agent-access-map/recovery.md index e6fc1098fe6..70615f26782 100644 --- a/maps/agent-access-map/recovery.md +++ b/maps/agent-access-map/recovery.md @@ -126,6 +126,13 @@ and must not create reader participation. No live recovery proof is claimed. Gaps: Permitted resolutions use native `call_api` under existing eligibility and ownership checks. Source/action policy determines whether continuation is allowed. +A recorded operator Stop before legacy runtime selection releases only that +run's task locks and preserves deferred input for normal later admission. This +does not grant an agent a new retry or provider-stop authority. Local +[database regressions](../../server/src/modules/wake-queue/adapters/queued-operator-stop-postgres.test.ts) +cover the release boundary and a competing active routine cycle; they are not +live provider-stop proof. + ## `exhaustion` Review: `baseline`. diff --git a/maps/feature-map/recovery.md b/maps/feature-map/recovery.md index 251c79327b1..7ca8f4be28b 100644 --- a/maps/feature-map/recovery.md +++ b/maps/feature-map/recovery.md @@ -203,6 +203,12 @@ Automated: [legacy continuation](../../server/src/services/recovery/legacy-conti cover their respective recovery paths. [Legacy failure continuation](../../tests/e2e/legacy-failure-continuation.spec.ts) adds a browser journey; it is not native runtime proof. +[Queued Stop database regressions](../../server/src/modules/wake-queue/adapters/queued-operator-stop-postgres.test.ts) +verify that an operator cancellation before runtime selection releases its own +locks, preserves deferred input without launching it, and leaves another active +cycle of the same routine unchanged. Provider cancellations and already selected +or dispatched runs retain their existing release behavior. + Manual: stop a disposable run, submit a distinguishable follow-up, and observe any cleanup wait. After confirmed stop, verify the follow-up reaches one eligible continuation and produces the requested answer. Repeat with a task/ancestor diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index d2624b2512f..caf6372b2f5 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -1,4 +1,5 @@ import { isAcknowledgedNativeReassignmentStop, isAcknowledgedNativeStop } from "../../../services/acknowledged-native-stop.js"; +import { readRunCancellation } from "../../../services/run-cancellation.js"; import { isCompletedOnboardingHandoffWake } from "../../../services/chat-completion-delivery.js"; import { instanceSettingsService } from "../../../services/instance-settings.js"; import { currentConversationCommentCondition } from "../../../services/agent-conversations.js"; @@ -1096,6 +1097,11 @@ export function createPostgresWakeQueueAdapter(db: Db, deps: WakeQueuePostgresAd sql`${agentWakeupRequests.payload}->>'issueId' = ${issueRow.id}`, )).limit(1) : []; + const cancellation = readRunCancellation(run.resultJson); + const queuedOperatorStop = run.runtimeMode === "legacy" && run.status === "cancelled" && + run.runtimeModeResolvedAt === null && run.startedAt === null && run.processStartedAt === null && + run.processPid === null && run.processGroupId === null && + cancellation?.source === "operator" && cancellation.expected; const preDrainFacts: PreDrainFacts = { issueRowPresent: issueRow !== null, executionRunIdMatchesRun: !issueRow || !issueRow.executionRunId || issueRow.executionRunId === run.id, @@ -1111,7 +1117,7 @@ export function createPostgresWakeQueueAdapter(db: Db, deps: WakeQueuePostgresAd executionCancellationAcknowledged: isAcknowledgedNativeReassignmentStop(run) || (run.status === "cancelled" && - (parseObject(run.resultJson?.executionCancellation).state === "acknowledged" || isAcknowledgedNativeStop(run)) && + (queuedOperatorStop || parseObject(run.resultJson?.executionCancellation).state === "acknowledged" || isAcknowledgedNativeStop(run)) && !interruptedQueue), }; const preDrain = decidePreDrain(preDrainFacts); diff --git a/server/src/modules/wake-queue/adapters/queued-operator-stop-postgres.test.ts b/server/src/modules/wake-queue/adapters/queued-operator-stop-postgres.test.ts new file mode 100644 index 00000000000..bd510d15f1b --- /dev/null +++ b/server/src/modules/wake-queue/adapters/queued-operator-stop-postgres.test.ts @@ -0,0 +1,95 @@ +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { agentWakeupRequests, agents, companies, createDb, heartbeatRuns, issues } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../../__tests__/helpers/embedded-postgres.js"; +import { createReleaseIssueExecution } from "../application/use-cases.js"; +import { createPostgresWakeQueueAdapter } from "./postgres.js"; + +const support = await getEmbeddedPostgresTestSupport(); +const describePostgres = support.supported ? describe : describe.skip; + +describePostgres("queued operator stop release", () => { + let db: ReturnType; + let database: Awaited>; + + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("paperclip-queued-operator-stop-"); + db = createDb(database.connectionString); + }, 20_000); + + afterAll(async () => { + await database?.cleanup(); + }); + + async function seed(input: { activeCycle?: boolean; source?: string; started?: boolean; selected?: boolean }) { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); + const originId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Stop regression", issuePrefix: `T${companyId.slice(0, 6)}`, + defaultResponsibleUserId: "operator", requireBoardApprovalForNewAgents: false }); + await db.insert(agents).values({ id: agentId, companyId, name: "Worker", role: "engineer", status: "active", + adapterType: "codex_local", adapterConfig: {}, runtimeConfig: { heartbeat: { wakeOnDemand: true } } }); + await db.insert(issues).values({ id: issueId, companyId, title: "Stopped cycle", status: "in_progress", + assigneeAgentId: agentId, responsibleUserId: "operator", originKind: "routine_execution", originId }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "on_demand", status: "cancelled", + runtimeMode: "legacy", contextSnapshot: { issueId }, finishedAt: new Date(), + ...(input.started ? { startedAt: new Date(), runtimeModeResolvedAt: new Date() } : input.selected ? { runtimeModeResolvedAt: new Date() } : {}), + resultJson: { ...(!input.started ? { executionRecovery: { kind: "bootstrap", providerWorkStarted: false }, + startupCancellation: { requestedAt: new Date().toISOString(), beforeNativeSelection: false } } : {}), cancellation: { source: input.source ?? "operator", expected: input.source !== "provider", + initiator: { type: input.source === "provider" ? "provider" : "user", id: "operator" }, + reason: "Stopped", recordedAt: new Date().toISOString() } } }); + let activeCycleId: string | null = null; + let activeRunId: string | null = null; + if (input.activeCycle) { + activeCycleId = randomUUID(); activeRunId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: activeRunId, companyId, agentId, invocationSource: "on_demand", status: "running", + contextSnapshot: { issueId: activeCycleId }, startedAt: new Date() }); + await db.insert(issues).values({ id: activeCycleId, companyId, title: "Current cycle", status: "in_progress", + assigneeAgentId: agentId, originKind: "routine_execution", originId, executionRunId: activeRunId }); + } else { + await db.update(issues).set({ executionRunId: runId, checkoutRunId: runId }).where(eq(issues.id, issueId)); + } + const [wake] = await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "automation", reason: "issue_commented", + status: "deferred_issue_execution", payload: { issueId }, requestedByActorType: "user", requestedByActorId: "operator" }).returning(); + const release = createReleaseIssueExecution({ + issueLock: createPostgresWakeQueueAdapter(db, { resolveResponsibleUserId: async () => "operator", + getRoutineEnv: async () => ({ routineId: null, env: null, responsibleUserId: null }), resolveSessionBeforeForWakeup: async () => null }), + recovery: { escalateStrandedAssignedIssue: async () => { throw Error("Unexpected recovery"); }, + escalateStrandedRecoveryIssueInPlace: async () => { throw Error("Unexpected recovery"); } }, + }); + return { companyId, agentId, issueId, runId, wakeId: wake.id, activeCycleId, activeRunId, release }; + } + + it.each([false, true])("preserves pending work after a pre-dispatch Stop (active cycle=%s)", async (activeCycle) => { + const f = await seed({ activeCycle }); + const before = (await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]; + const result = await f.release({ companyId: f.companyId, runId: f.runId, now: new Date() }); + expect(result).toEqual({ outcome: { kind: "released" }, postCommitEffects: [] }); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toMatchObject({ status: "in_progress", + statusVersion: before.statusVersion, assigneeAgentId: f.agentId, executionRunId: null, checkoutRunId: null }); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.wakeId)))[0]) + .toMatchObject({ status: "deferred_issue_execution", runId: null }); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(activeCycle ? 2 : 1); + if (f.activeCycleId) expect((await db.select().from(issues).where(eq(issues.id, f.activeCycleId)))[0]) + .toMatchObject({ status: "in_progress", executionRunId: f.activeRunId }); + expect(await f.release({ companyId: f.companyId, runId: f.runId, now: new Date() })).toEqual(result); + }); + + it("keeps ordinary provider cancellation promotion", async () => { + const f = await seed({ source: "provider" }); + expect((await f.release({ companyId: f.companyId, runId: f.runId, now: new Date() })).outcome.kind).toBe("promoted"); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.wakeId)))[0].status).toBe("queued"); + }); + + it("does not replace stop acknowledgement for a dispatched run", async () => { + const f = await seed({ started: true }); + expect((await f.release({ companyId: f.companyId, runId: f.runId, now: new Date() })).outcome.kind).toBe("released"); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.wakeId)))[0]) + .toMatchObject({ status: "deferred_issue_execution", runId: null }); + }); + + it("keeps the existing release path when runtime selection has occurred", async () => { + const f = await seed({ selected: true }); + expect((await f.release({ companyId: f.companyId, runId: f.runId, now: new Date() })).outcome.kind).toBe("promoted"); + }); +});