mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-21 16:27:18 +02:00
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 <noreply@paperclip.ing>
This commit is contained in:
1 parent
976cdc77ed
commit
6fae6f2cb8
5 files changed
+122
-1
No files matched your search
@@ -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.
|
||||
|
||||
@@ -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`.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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<typeof createDb>;
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
|
||||
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");
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user