mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-23 00:53:38 +02:00
fix(heartbeat): preserve active retries when comments arrive (#16855)
## Thinking Path > - Paperclip manages agents and their work. > - Heartbeat retries preserve task ownership after a recoverable failure. > - The issue read model includes retries that are scheduled, queued, or running. > - The human comment route treated every retry in that model as parked work. > - A normal status comment could therefore cancel a running continuation and return its task to Todo. > - This pull request limits comment supersession to parked retries and checks that state at the cancellation fence. > - The agent can keep working when a person adds context. Explicit Stop and Interrupt remain available. ## Linked Issues or Issue Description **What happened?** A Board comment on an in-progress task cancelled its running recovery continuation. The handler read a live retry from `getCurrentScheduledRetry` but did not check its status. The cancellation also forced workspace recovery when copy-back did not complete. **Expected behavior** A comment must preserve queued and running execution. Only a retry that is still `scheduled_retry` may be superseded. If promotion wins the race, the comment must keep the task in progress and follow the normal wake path. **Steps to reproduce** 1. Let a task retry advance from `scheduled_retry` to `running`. 2. Keep its retry reason and source run identity in the stored row. 3. Post a normal human comment, without an Interrupt request. 4. Observe the old handler cancel the run and move the task to Todo. **Paperclip version or commit** Reproduced on master base `9966f31bee`. The staging incident served `f3cd3f4ea89539b68410841d39ec15a5c626007f`. **Deployment mode** Staging server with a remote runner. The defect is in the core comment and cancellation path. Refs #16835 and #16843. Those repairs cover runner health and workspace preservation. This change prevents a normal comment from requesting the wrong cancellation. Related #9732 promotes human wakes on parked retries; it does not protect a retry that is already running. ## What Changed - Check the live retry status in both comment routes: POST comments and PATCH issue with a comment. - Add an optional expected-status fence to cancellation. A concurrent promotion or terminal outcome returns a no-op result. - Make due-retry promotion honor a cancellation claim under its issue and run row locks. A scheduler cannot promote work after cancellation has won. - Finish persisted cancellations of parked retries through the existing startup and periodic recovery sweep. Preserve their original request and retry a failed settlement without starting provider work. - Keep the task active when guarded cancellation does not win. Do not record a cancellation that did not occur. - Add route regressions, real PostgreSQL race tests, and pending-retry and explicit Stop safety coverage. - Exercise both normal HTTP comment routes with real services, stored Board membership, PostgreSQL, and a live child process. Assert that the process can still perform work, the task keeps its original execution, and no second executor starts. - Document the comment supersession contract and update the task feature/access maps with the local regression coverage and deployed acceptance recipe. ## Verification - Before the fix, all four POST/PATCH regressions cancelled queued or running work. - Before the fix, the real PostgreSQL race test cancelled a retry that became running after lookup. - The comment-route suite passes **103 tests**. The final dispatch, cancellation, restart recovery and HTTP/live-process suites pass **119 tests**, for **222 affected tests** across these runs. They cover comment delivery, both sides of the promotion race, concurrent terminal outcomes, pending retries, explicit Stop, and a live process that can continue work. - Command: `node node_modules/vitest/vitest.mjs run server/src/__tests__/issue-comment-reopen-routes.test.ts server/src/services/heartbeat/run-control.test.ts`. - The normal HTTP/live-process acceptance suite passes all **14 tests**, including both comment endpoints with stored Board membership and deterministic wakeup teardown. - Review exposed a second race after the cancellation claim. Its new PostgreSQL test failed before the dispatch guard with `promoted` and passes afterward with `not_promoted`, followed by completed cancellation. - A real PostgreSQL constraint interrupts the terminal cancellation write after its intent commits. The assembled recovery service failed to settle that row before the repair. Afterward a fresh service settles it, clears its task lock, cancels its saved wake, preserves the original request and leaves unrelated parked retries alone. A repeated database failure preserves the intent for a later successful sweep; a second successful sweep adds no duplicate event. - Full `pnpm -r typecheck` and `pnpm build` pass. Server typecheck and build pass again after the final server-only review fix. A local full-suite attempt started on `f8773bf` did not return an overall result; its log is preserved and no local full-suite pass is claimed. The new recovery change passes the focused suites and fresh server typecheck/build. Latest-head CI on `8f928adee67c089e1ac405d2ad3b169eedb8da27` is green: the full PR workflow and commitperclip review passed; only the two optional Storybook jobs skipped. - Greptile reviewed final head `8f928adee67c089e1ac405d2ad3b169eedb8da27` at **5/5**, with no unresolved findings. All actual latest-head CI checks passed; the two optional Storybook jobs skipped. - `node scripts/check-maps.mjs` passes after the map updates (34 areas, 172 capability IDs); it checks structure only. - Fixed-revision staging acceptance is pending deployment. The original workspace remains held and its latest source checkpoint is backed up and verified. ## Risks - A promoted retry must keep its execution and ownership when a comment arrives. The database status predicate and route no-op handling enforce this at the race boundary. - A cancellation intent must prevent promotion without leaving work parked forever after a restart. The recovery sweep consumes at most 20 such intents per pass through guarded cancellation, and a failed settlement keeps its intent for retry. - A parked retry must still accept fresh instructions. Its existing cancellation and wake behavior remains covered. - Ordinary Stop and Interrupt calls omit the expected-status option. Their cancellation authority is unchanged and remains covered by the run-control suite. - No schema migration or API request contract changes are required. ## Model Used OpenAI `gpt-6.1-sol` in Codex, with extra-high reasoning, tool use, and code execution. The session does not expose its context-window limit. ## 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 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
36e4c6657f
commit
d20cb69ce5
10 files changed
+316
-24
No files matched your search
@@ -242,6 +242,17 @@ The **Retry now** action returns `waiting` with the saved retry schedule while
|
||||
cleanup holds ownership. Its controls show the wait without claiming that a run
|
||||
started, and another click can check the gate again.
|
||||
|
||||
A human comment may supersede a retry only while it remains `scheduled_retry`.
|
||||
Queued and running retries retain their execution and receive the comment through
|
||||
normal wake coalescing. Supersession rechecks the scheduled status atomically with
|
||||
the cancellation fence; a concurrent promotion must not cancel the active run or
|
||||
move its task back to `todo`. Explicit Stop and Interrupt keep their authority.
|
||||
Once cancellation wins that fence, due-retry promotion respects its claim until
|
||||
the cancellation owner finishes settlement. Startup and periodic recovery finish
|
||||
persisted cancellations of parked retries through the normal guarded cancellation
|
||||
path if the original owner stops or its final database write fails. They preserve
|
||||
the original cancellation request and never start a provider turn for that work.
|
||||
|
||||
Recovery instructions in a wake describe only an active or escalated recovery
|
||||
action. Resolving or cancelling that action removes its repair instructions from
|
||||
subsequent wakes, so the original worker can resume the task. This prompt change
|
||||
|
||||
@@ -22,7 +22,10 @@ API: `present` — representative operation IDs:
|
||||
Agent access: `available` — Native create_task; REST / CLI / MCP.
|
||||
|
||||
Policy: Company/task visibility, assignment/change policy, active-run checkout ownership
|
||||
and lifecycle rules.
|
||||
and lifecycle rules. Ordinary Board comments preserve queued/running retries;
|
||||
only a parked retry can be superseded. Explicit Stop/Interrupt retain their
|
||||
existing actor and execution authority. Agent comment tools do not acquire Board
|
||||
identity or cancellation authority through this behavior.
|
||||
|
||||
Evidence: `source-only` — [implementation](../../server/src/routes/issues.ts) at the
|
||||
reviewed commit above.
|
||||
|
||||
@@ -70,6 +70,14 @@ reassignment and confirm that the outgoing run still stops.
|
||||
|
||||
Manual: Add a comment and edit authorized properties. Reload and verify history, actor, and values. Complete a disposable task and inspect its deliverable; reopen through an available action and confirm the next execution follows current policy.
|
||||
|
||||
Manual: Add an ordinary Board comment while a task retry is queued or running.
|
||||
Confirm that the same execution keeps working and receives the saved message,
|
||||
without a second executor or an implicit move to Todo. Repeat with a parked retry
|
||||
and verify that fresh instructions can supersede it. Explicit Stop must still
|
||||
cancel active work. The [comment route regressions](../../server/src/__tests__/issue-comment-reopen-routes.test.ts)
|
||||
and [HTTP/process and interrupted-cancellation tests](../../server/src/__tests__/issue-scheduled-retry-routes.test.ts)
|
||||
cover these boundaries locally; deployed runner acceptance remains separate.
|
||||
|
||||
## Gotchas
|
||||
|
||||
- Priority remains an API field, but ordinary priority UI editing is hidden.
|
||||
|
||||
@@ -1215,7 +1215,7 @@ describe("issue comment reopen routes", () => {
|
||||
"11111111-1111-4111-8111-111111111111",
|
||||
{ status: "todo" },
|
||||
);
|
||||
expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("retry-run-1");
|
||||
expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("retry-run-1", undefined, { expectedStatus: "scheduled_retry" });
|
||||
expect(mockLogActivity).toHaveBeenCalledWith(
|
||||
expect.anything(),
|
||||
expect.objectContaining({
|
||||
@@ -1246,6 +1246,58 @@ describe("issue comment reopen routes", () => {
|
||||
);
|
||||
});
|
||||
|
||||
it.each([
|
||||
["post", "queued"],
|
||||
["post", "running"],
|
||||
["patch", "queued"],
|
||||
["patch", "running"],
|
||||
] as const)("preserves a %s comment's %s retry execution", async (method, status) => {
|
||||
const issue = {
|
||||
...makeIssue("in_progress"),
|
||||
executionRunId: "retry-run-1",
|
||||
};
|
||||
mockIssueService.getById.mockResolvedValue(issue);
|
||||
mockIssueService.getCurrentScheduledRetry.mockResolvedValue({
|
||||
...createScheduledRetry(),
|
||||
status,
|
||||
});
|
||||
mockIssueService.update.mockImplementation(async (_id, patch) => ({ ...issue, ...patch }));
|
||||
|
||||
const app = await installActor(createApp());
|
||||
const res = method === "post"
|
||||
? await request(app).post(`/api/issues/${issue.id}/comments`).send({ body: "Recovery evidence is attached." })
|
||||
: await request(app).patch(`/api/issues/${issue.id}`).send({ comment: "Recovery evidence is attached." });
|
||||
|
||||
expect(res.status).toBe(method === "post" ? 201 : 200);
|
||||
expect(mockHeartbeatService.cancelRun).not.toHaveBeenCalled();
|
||||
expect(mockIssueService.update).not.toHaveBeenCalledWith(issue.id, expect.objectContaining({ status: "todo" }));
|
||||
await waitForWakeup(() => expect(mockHeartbeatService.wakeup).toHaveBeenCalledWith(
|
||||
issue.assigneeAgentId,
|
||||
expect.objectContaining({ reason: "issue_commented" }),
|
||||
));
|
||||
});
|
||||
|
||||
it.each(["post", "patch"] as const)("keeps the task active when a retry is promoted during a %s comment", async method => {
|
||||
const issue = { ...makeIssue("in_progress"), executionRunId: "retry-run-1" };
|
||||
mockIssueService.getById.mockResolvedValue(issue);
|
||||
mockIssueService.getCurrentScheduledRetry.mockResolvedValue(createScheduledRetry());
|
||||
mockHeartbeatService.cancelRun.mockResolvedValue({ id: "retry-run-1", status: "running" });
|
||||
mockIssueService.update.mockImplementation(async (_id, patch) => ({ ...issue, ...patch }));
|
||||
const app = await installActor(createApp());
|
||||
|
||||
const res = method === "post"
|
||||
? await request(app).post(`/api/issues/${issue.id}/comments`).send({ body: "Additional context for the active run." })
|
||||
: await request(app).patch(`/api/issues/${issue.id}`).send({ comment: "Additional context for the active run." });
|
||||
|
||||
expect(res.status).toBe(method === "post" ? 201 : 200);
|
||||
expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("retry-run-1", undefined, { expectedStatus: "scheduled_retry" });
|
||||
expect(mockIssueService.update).not.toHaveBeenCalledWith(issue.id, expect.objectContaining({ status: "todo" }));
|
||||
expect(mockLogActivity).not.toHaveBeenCalledWith(expect.anything(), expect.objectContaining({ action: "heartbeat.cancelled" }));
|
||||
await waitForWakeup(() => expect(mockHeartbeatService.wakeup).toHaveBeenCalledWith(
|
||||
issue.assigneeAgentId, expect.objectContaining({ reason: "issue_commented" }),
|
||||
));
|
||||
});
|
||||
|
||||
it("does not move scheduled-retry issues to todo when POST comment retry cancellation fails", async () => {
|
||||
const issue = {
|
||||
...makeIssue("in_progress"),
|
||||
@@ -1262,7 +1314,7 @@ describe("issue comment reopen routes", () => {
|
||||
.send({ body: "I added the missing detail; please continue." });
|
||||
|
||||
expect(res.status).toBe(500);
|
||||
expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("retry-run-1");
|
||||
expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("retry-run-1", undefined, { expectedStatus: "scheduled_retry" });
|
||||
expect(mockIssueService.update).not.toHaveBeenCalled();
|
||||
expect(mockIssueService.addComment).not.toHaveBeenCalled();
|
||||
expect(mockLogActivity).not.toHaveBeenCalledWith(
|
||||
@@ -1945,7 +1997,7 @@ describe("issue comment reopen routes", () => {
|
||||
actorUserId: "local-board",
|
||||
}),
|
||||
);
|
||||
expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("retry-run-1");
|
||||
expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("retry-run-1", undefined, { expectedStatus: "scheduled_retry" });
|
||||
await waitForWakeup(() =>
|
||||
expect(mockHeartbeatService.wakeup).toHaveBeenCalledWith(
|
||||
"22222222-2222-4222-8222-222222222222",
|
||||
@@ -1976,7 +2028,7 @@ describe("issue comment reopen routes", () => {
|
||||
.send({ comment: "Retry window is over; please continue." });
|
||||
|
||||
expect(res.status).toBe(500);
|
||||
expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("retry-run-1");
|
||||
expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("retry-run-1", undefined, { expectedStatus: "scheduled_retry" });
|
||||
expect(mockIssueService.update).not.toHaveBeenCalled();
|
||||
expect(mockIssueService.addComment).not.toHaveBeenCalled();
|
||||
expect(mockLogActivity).not.toHaveBeenCalledWith(
|
||||
|
||||
@@ -1,13 +1,16 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { spawn } from "node:child_process";
|
||||
import { once } from "node:events";
|
||||
import express from "express";
|
||||
import request from "supertest";
|
||||
import { and, eq, inArray } from "drizzle-orm";
|
||||
import { and, eq, inArray, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
|
||||
import {
|
||||
activityLog,
|
||||
agents,
|
||||
agentWakeupRequests,
|
||||
companies,
|
||||
companyMemberships,
|
||||
createDb,
|
||||
environmentLeases,
|
||||
heartbeatRunEvents,
|
||||
@@ -23,6 +26,8 @@ import {
|
||||
} from "./helpers/embedded-postgres.js";
|
||||
import { errorHandler } from "../middleware/index.js";
|
||||
import { issueRoutes } from "../routes/issues.js";
|
||||
import { runningProcesses } from "../adapters/index.js";
|
||||
import { heartbeatService } from "../services/heartbeat.js";
|
||||
|
||||
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
|
||||
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
|
||||
@@ -55,6 +60,7 @@ describeEmbeddedPostgres("issue scheduled retry routes", () => {
|
||||
await db.delete(heartbeatRuns);
|
||||
await db.delete(agentWakeupRequests);
|
||||
await db.delete(agents);
|
||||
await db.delete(companyMemberships);
|
||||
await db.delete(companies);
|
||||
});
|
||||
|
||||
@@ -207,6 +213,90 @@ describeEmbeddedPostgres("issue scheduled retry routes", () => {
|
||||
return { companyId, agentId, issueId, sourceRunId, retryRunId, scheduledRetryAt };
|
||||
}
|
||||
|
||||
it("finishes a parked retry cancellation after its owner loses the final database write", async () => {
|
||||
const f = await seedIssueWithRetry({ agentStatus: "paused" });
|
||||
const unrelated = await seedIssueWithRetry({ agentStatus: "paused" });
|
||||
await db.execute(sql`ALTER TABLE heartbeat_runs ADD CONSTRAINT reject_retry_cancellation CHECK (status <> 'cancelled') NOT VALID`);
|
||||
try {
|
||||
await expect(heartbeatService(db).cancelRun(f.retryRunId, "Superseded by a human comment", {
|
||||
expectedStatus: "scheduled_retry",
|
||||
})).rejects.toThrow();
|
||||
// A failed sweep leaves the same durable intent available for a later one.
|
||||
await heartbeatService(db).reapOrphanedRuns();
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.retryRunId)))
|
||||
.toEqual([expect.objectContaining({ status: "scheduled_retry" })]);
|
||||
} finally {
|
||||
await db.execute(sql`ALTER TABLE heartbeat_runs DROP CONSTRAINT reject_retry_cancellation`);
|
||||
}
|
||||
const [interrupted] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.retryRunId));
|
||||
expect(interrupted).toMatchObject({ status: "scheduled_retry", resultJson: {
|
||||
startupCancellation: expect.any(Object), cancellation: { reason: "Superseded by a human comment" },
|
||||
} });
|
||||
|
||||
// A fresh service has no process-local cancellation owner or settlement promise.
|
||||
await heartbeatService(db).reapOrphanedRuns();
|
||||
const [cancelled] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.retryRunId));
|
||||
expect(cancelled).toMatchObject({ status: "cancelled", error: "Superseded by a human comment" });
|
||||
expect(cancelled!.resultJson!.cancellation).toEqual(interrupted!.resultJson!.cancellation);
|
||||
expect(await db.select().from(issues).where(eq(issues.id, f.issueId)))
|
||||
.toEqual([expect.objectContaining({ executionRunId: null, assigneeAgentId: f.agentId })]);
|
||||
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.runId, f.retryRunId)))
|
||||
.toEqual([expect.objectContaining({ status: "cancelled" })]);
|
||||
const events = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.retryRunId));
|
||||
await heartbeatService(db).reapOrphanedRuns();
|
||||
expect(await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.retryRunId))).toEqual(events);
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, unrelated.retryRunId)))
|
||||
.toEqual([expect.objectContaining({ status: "scheduled_retry", resultJson: null })]);
|
||||
});
|
||||
|
||||
it.each(["post", "patch"] as const)("keeps a retry's live process and saved work owner when a %s comment arrives", async method => {
|
||||
const f = await seedIssueWithRetry({ retryStatus: "running" });
|
||||
await db.insert(companyMemberships).values({
|
||||
companyId: f.companyId, principalType: "user", principalId: "board-user",
|
||||
membershipRole: "admin", status: "active",
|
||||
});
|
||||
const child = spawn(process.execPath, ["-e", `
|
||||
process.on('message', message => {
|
||||
if (message === 'continue-work') process.send('work-continues');
|
||||
});
|
||||
process.send('ready');
|
||||
`], { stdio: ["ignore", "ignore", "pipe", "ipc"] });
|
||||
const exited = once(child, "exit");
|
||||
const heartbeat = heartbeatService(db);
|
||||
try {
|
||||
await once(child, "message", { signal: AbortSignal.timeout(5_000) });
|
||||
runningProcesses.set(f.retryRunId, { child, graceSec: 1, processGroupId: null });
|
||||
await db.update(heartbeatRuns).set({
|
||||
startedAt: new Date(), runtimeModeResolvedAt: new Date(), processPid: child.pid,
|
||||
}).where(eq(heartbeatRuns.id, f.retryRunId));
|
||||
const app = createApp(boardActor(f.companyId));
|
||||
|
||||
const res = method === "post"
|
||||
? await request(app).post(`/api/issues/${f.issueId}/comments`).send({ body: "Verified source recovery; continue the original work." })
|
||||
: await request(app).patch(`/api/issues/${f.issueId}`).send({ comment: "Verified source recovery; continue the original work." });
|
||||
|
||||
expect(res.status, JSON.stringify(res.body)).toBe(method === "post" ? 201 : 200);
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
const progress = once(child, "message", { signal: AbortSignal.timeout(5_000) });
|
||||
child.send("continue-work");
|
||||
expect((await progress)[0]).toBe("work-continues");
|
||||
const [savedIssue] = await db.select().from(issues).where(eq(issues.id, f.issueId));
|
||||
const [savedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.retryRunId));
|
||||
expect(savedIssue).toMatchObject({ status: "in_progress", executionRunId: f.retryRunId });
|
||||
expect(savedRun).toMatchObject({ status: "running", processPid: child.pid });
|
||||
expect(await db.select().from(issueComments).where(eq(issueComments.issueId, f.issueId)))
|
||||
.toHaveLength(1);
|
||||
expect(await db.select().from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "running"),
|
||||
))).toHaveLength(1);
|
||||
} finally {
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
runningProcesses.delete(f.retryRunId);
|
||||
if (child.exitCode === null && child.signalCode === null) child.kill();
|
||||
await exited;
|
||||
}
|
||||
});
|
||||
|
||||
it("surfaces the current scheduled retry in the issue read model", async () => {
|
||||
const { companyId, issueId, agentId, sourceRunId, retryRunId, scheduledRetryAt } = await seedIssueWithRetry();
|
||||
|
||||
|
||||
@@ -760,6 +760,9 @@ export function createPostgresRunDispatchAdapter(
|
||||
const promoteLockedRun = async (tx: Db, run: HeartbeatRun, publications: ActivityPublication[]) => {
|
||||
if (
|
||||
run.status !== "scheduled_retry" ||
|
||||
// The locked row is also the cancellation claim. If Stop won first,
|
||||
// leave settlement to that owner rather than promote its parked run.
|
||||
run.resultJson?.startupCancellation ||
|
||||
!run.scheduledRetryAt ||
|
||||
new Date(run.scheduledRetryAt).getTime() > now.getTime()
|
||||
) {
|
||||
|
||||
+23
-13
@@ -4134,8 +4134,11 @@ export function issueRoutes(
|
||||
if (!scheduledRetryRunId) return null;
|
||||
|
||||
try {
|
||||
const cancelled = await heartbeat.cancelRun(scheduledRetryRunId);
|
||||
const cancelledRunId = cancelled?.id ?? scheduledRetryRunId;
|
||||
const cancelled = await heartbeat.cancelRun(scheduledRetryRunId, undefined, {
|
||||
expectedStatus: "scheduled_retry",
|
||||
});
|
||||
if (cancelled?.status !== "cancelled") return null;
|
||||
const cancelledRunId = cancelled.id;
|
||||
await logActivity(db, {
|
||||
companyId: input.issue.companyId,
|
||||
actorType: input.actor.actorType,
|
||||
@@ -13791,8 +13794,8 @@ export function issueRoutes(
|
||||
})
|
||||
? await svc.getCurrentScheduledRetry(existing.id)
|
||||
: null;
|
||||
const shouldResumeInProgressScheduledRetry =
|
||||
!!scheduledRetryForHumanComment &&
|
||||
let shouldResumeInProgressScheduledRetry =
|
||||
scheduledRetryForHumanComment?.status === "scheduled_retry" &&
|
||||
scheduledRetryForHumanComment.agentId === requestedAssigneeAgentId;
|
||||
const assigneeSelfCommentOnTerminal =
|
||||
isAssigneeSelfCommentOnTerminalIssue({
|
||||
@@ -13945,6 +13948,12 @@ export function issueRoutes(
|
||||
issue: existing,
|
||||
actor,
|
||||
});
|
||||
if (!cancelledScheduledRetryRunId) {
|
||||
shouldResumeInProgressScheduledRetry = false;
|
||||
if (!explicitMoveToTodoRequested && req.body.status === undefined) {
|
||||
delete updateFields.status;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (req.body.executionPolicy !== undefined) {
|
||||
updateFields.executionPolicy = applyActorMonitorScheduledBy(
|
||||
@@ -18265,8 +18274,8 @@ export function issueRoutes(
|
||||
})
|
||||
? await svc.getCurrentScheduledRetry(issue.id)
|
||||
: null;
|
||||
const shouldResumeInProgressScheduledRetry =
|
||||
!!scheduledRetryForHumanComment &&
|
||||
let shouldResumeInProgressScheduledRetry =
|
||||
scheduledRetryForHumanComment?.status === "scheduled_retry" &&
|
||||
scheduledRetryForHumanComment.agentId === issue.assigneeAgentId;
|
||||
const assigneeSelfCommentOnTerminal =
|
||||
isAssigneeSelfCommentOnTerminalIssue({
|
||||
@@ -18366,6 +18375,14 @@ export function issueRoutes(
|
||||
|
||||
let scheduledRetrySupersededByComment = false;
|
||||
let cancelledScheduledRetryRunId: string | null = null;
|
||||
if (effectiveMoveToTodoRequested && shouldResumeInProgressScheduledRetry) {
|
||||
cancelledScheduledRetryRunId = await cancelScheduledRetrySupersededByComment({
|
||||
scheduledRetryRunId: scheduledRetryForHumanComment?.runId,
|
||||
issue,
|
||||
actor,
|
||||
});
|
||||
shouldResumeInProgressScheduledRetry = cancelledScheduledRetryRunId !== null;
|
||||
}
|
||||
if (
|
||||
effectiveMoveToTodoRequested &&
|
||||
(isClosed ||
|
||||
@@ -18375,13 +18392,6 @@ export function issueRoutes(
|
||||
scheduledRetrySupersededByComment =
|
||||
shouldResumeInProgressScheduledRetry &&
|
||||
issue.status === "in_progress";
|
||||
cancelledScheduledRetryRunId = scheduledRetrySupersededByComment
|
||||
? await cancelScheduledRetrySupersededByComment({
|
||||
scheduledRetryRunId: scheduledRetryForHumanComment?.runId,
|
||||
issue,
|
||||
actor,
|
||||
})
|
||||
: null;
|
||||
const reopenedIssue = await svc.update(id, { status: "todo" });
|
||||
if (!reopenedIssue) {
|
||||
res.status(404).json({ error: "Issue not found" });
|
||||
|
||||
@@ -75,6 +75,7 @@ import {
|
||||
} from "../hot-restart.js";
|
||||
import { serverVersion } from "../../version.js";
|
||||
import { adapterExecutionControls, waitForAdapterStop } from "../adapter-execution-control.js";
|
||||
import { readRunCancellation } from "../run-cancellation.js";
|
||||
import { allowLegacyShutdownWorkspaceCleanup, beginLegacyShutdownWorkspaceSettlement, finishLegacyShutdownWorkspaceSettlement, maintainLegacyShutdownWorkspaceSettlement, sweepLegacyShutdownWorkspaceSettlements } from "../legacy-shutdown-workspace-settlement.js";
|
||||
|
||||
import type { environmentService } from "../environments.js";
|
||||
@@ -82,6 +83,7 @@ import type { environmentRuntimeService, ProviderResourceDisposition } from "../
|
||||
import type { agentInstructionWorkingCopyService } from "../agent-instruction-working-copies.js";
|
||||
import type { createHeartbeatRunState } from "./run-state.js";
|
||||
import type { HeartbeatRetryDependencies, createHeartbeatRetries } from "./retries.js";
|
||||
import type { CancelRunOptions } from "./run-control.js";
|
||||
|
||||
type HeartbeatRun = typeof heartbeatRuns.$inferSelect;
|
||||
type HeartbeatRetries = ReturnType<typeof createHeartbeatRetries>;
|
||||
@@ -144,7 +146,7 @@ export interface HeartbeatRecoveryDependencies extends Pick<HeartbeatRetryDepend
|
||||
acknowledgeRemoteStop: (runId: string, companyId: string) => Promise<void>;
|
||||
resumeRemoteStopComments: (run: HeartbeatRun) => Promise<void>;
|
||||
dispatchPendingNativeStatusWakeups: () => Promise<unknown>;
|
||||
cancelRunInternal: (runId: string, reason?: string, options?: { resultJson?: Record<string, unknown> }) => Promise<unknown>;
|
||||
cancelRunInternal: (runId: string, reason?: string, options?: Pick<CancelRunOptions, "resultJson" | "expectedStatus" | "errorCode">) => Promise<unknown>;
|
||||
startNextQueuedRunForAgent: (agentId: string) => Promise<unknown>;
|
||||
}
|
||||
|
||||
@@ -1916,6 +1918,31 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
|
||||
await preserveCancelledNativeWorkspaces(db);
|
||||
const staleThresholdMs = opts?.staleThresholdMs ?? 0;
|
||||
const now = new Date();
|
||||
// A parked retry has no executor to finish a cancellation after a restart
|
||||
// or a failed status write. Consume its durable intent through normal run
|
||||
// control; the guarded CAS and terminal settlement make competing sweeps
|
||||
// safe, and dispatch cannot promote a fenced retry in the meantime.
|
||||
const interruptedRetryCancellations = await db.select().from(heartbeatRuns)
|
||||
.where(and(
|
||||
eq(heartbeatRuns.status, "scheduled_retry"),
|
||||
sql`jsonb_typeof(${heartbeatRuns.resultJson}->'startupCancellation') = 'object'`,
|
||||
))
|
||||
.orderBy(asc(heartbeatRuns.updatedAt), asc(heartbeatRuns.id))
|
||||
.limit(20);
|
||||
for (const run of interruptedRetryCancellations) {
|
||||
const cancellation = readRunCancellation(run.resultJson);
|
||||
if (!cancellation?.expected) continue;
|
||||
try {
|
||||
await cancelRunInternal(run.id, cancellation.reason, {
|
||||
expectedStatus: "scheduled_retry",
|
||||
errorCode: readNonEmptyString(parseObject(run.resultJson?.startupCancellation).errorCode) ?? "cancelled",
|
||||
resultJson: parseObject(run.resultJson),
|
||||
});
|
||||
} catch (error) {
|
||||
// Leave the durable intent for the next bounded periodic sweep.
|
||||
logger.warn({ err: error, runId: run.id }, "failed to finish interrupted retry cancellation");
|
||||
}
|
||||
}
|
||||
// Recovery never launches a provider or infers stopped ownership from
|
||||
// terminal status. Uncaptured local copies require durable stop evidence.
|
||||
await instructionCopies.recoverStopped().catch(error => {
|
||||
|
||||
@@ -20,6 +20,7 @@ import { createHeartbeatRunPreparation } from "./run-preparation.js";
|
||||
import { createHeartbeatLifecycle, type HeartbeatLifecycleDependencies } from "./run-lifecycle.js";
|
||||
import { createHeartbeatRunControl, type HeartbeatRunControlDependencies } from "./run-control.js";
|
||||
import { preserveCancelledNativeWorkspaces } from "../native-runtime/native-failed-workspace-recovery.js";
|
||||
import { createPostgresRunDispatchAdapter } from "../../modules/run-dispatch/adapters/postgres.js";
|
||||
|
||||
vi.mock("../live-events.js", () => ({ publishLiveEvent: vi.fn() }));
|
||||
vi.mock("../adapter-execution-control.js", async (importOriginal) => {
|
||||
@@ -197,6 +198,88 @@ describe.skipIf(!support.supported)("heartbeat run-control database wiring", ()
|
||||
return run;
|
||||
}
|
||||
|
||||
it.each(["queued", "running", "cancelled", "succeeded"] as const)("does not supersede a retry that becomes %s between lookup and the cancellation fence", async status => {
|
||||
const { company, agent } = await fixture();
|
||||
const [run] = await db.insert(heartbeatRuns).values({
|
||||
companyId: company.id, agentId: agent.id, status: "scheduled_retry",
|
||||
runtimeMode: "legacy", runtimeModeResolvedAt: new Date(),
|
||||
scheduledRetryReason: "execution_reconciliation_continuation",
|
||||
}).returning();
|
||||
const deps = callbacks(db);
|
||||
const originalGetRun = deps.getRun.getMockImplementation()!;
|
||||
deps.getRun.mockImplementationOnce(async id => {
|
||||
const snapshot = await originalGetRun(id);
|
||||
await db.update(heartbeatRuns).set({ status, startedAt: new Date() }).where(eq(heartbeatRuns.id, id));
|
||||
return snapshot;
|
||||
});
|
||||
|
||||
const control = createHeartbeatRunControl(db, deps);
|
||||
const result = await control.cancelRunInternal(run.id, undefined, { expectedStatus: "scheduled_retry" });
|
||||
|
||||
expect(result).toBeNull();
|
||||
expect(await originalGetRun(run.id)).toMatchObject({ status, resultJson: null });
|
||||
expect(deps.envOrchestrator.releaseForRun).not.toHaveBeenCalled();
|
||||
expect(deps.options.closeWarmNativeSessionsForRun).not.toHaveBeenCalled();
|
||||
expect(deps.appendRunEvent).not.toHaveBeenCalled();
|
||||
|
||||
// An actual Stop still has authority over promoted, nonterminal work.
|
||||
if (status === "queued" || status === "running") {
|
||||
expect(await control.cancelRunInternal(run.id)).toMatchObject({ status: "cancelled" });
|
||||
}
|
||||
});
|
||||
|
||||
it("can supersede a retry that is still scheduled", async () => {
|
||||
const { company, agent } = await fixture();
|
||||
const [run] = await db.insert(heartbeatRuns).values({
|
||||
companyId: company.id, agentId: agent.id, status: "scheduled_retry",
|
||||
runtimeMode: "legacy", runtimeModeResolvedAt: new Date(),
|
||||
}).returning();
|
||||
const deps = callbacks(db);
|
||||
|
||||
expect(await createHeartbeatRunControl(db, deps).cancelRunInternal(
|
||||
run.id, undefined, { expectedStatus: "scheduled_retry" },
|
||||
)).toMatchObject({ status: "cancelled" });
|
||||
});
|
||||
|
||||
it("blocks promotion after guarded cancellation wins but before it settles", async () => {
|
||||
const { company, agent } = await fixture();
|
||||
await db.update(companies).set({ defaultResponsibleUserId: "operator" }).where(eq(companies.id, company.id));
|
||||
const [issue] = await db.insert(issues).values({
|
||||
companyId: company.id, title: "Keep the cancellation claim", status: "in_progress", assigneeAgentId: agent.id,
|
||||
}).returning();
|
||||
const [run] = await db.insert(heartbeatRuns).values({
|
||||
companyId: company.id, agentId: agent.id, issueId: issue.id, status: "scheduled_retry",
|
||||
runtimeMode: "legacy", runtimeModeResolvedAt: new Date(), responsibleUserId: "operator",
|
||||
scheduledRetryAt: new Date(), scheduledRetryReason: "transient_failure", contextSnapshot: { issueId: issue.id },
|
||||
}).returning();
|
||||
await db.update(issues).set({ executionRunId: run.id }).where(eq(issues.id, issue.id));
|
||||
const deps = callbacks(db);
|
||||
const settle = deps.setRunStatusFromLive.getMockImplementation()!;
|
||||
let claimed!: () => void, release!: () => void;
|
||||
const claimReached = new Promise<void>(resolve => { claimed = resolve; });
|
||||
const continueSettlement = new Promise<void>(resolve => { release = resolve; });
|
||||
deps.setRunStatusFromLive.mockImplementation(async (...args) => {
|
||||
claimed();
|
||||
await continueSettlement;
|
||||
return settle(...args);
|
||||
});
|
||||
const cancellation = createHeartbeatRunControl(db, deps).cancelRunInternal(
|
||||
run.id, undefined, { expectedStatus: "scheduled_retry" },
|
||||
);
|
||||
try {
|
||||
await Promise.race([claimReached, cancellation.then(() => { throw new Error("Cancellation did not reach settlement"); })]);
|
||||
expect((await deps.getRun(run.id))?.resultJson?.startupCancellation).toBeTruthy();
|
||||
expect(await createPostgresRunDispatchAdapter(db).promoteOrCancelDueRetry({
|
||||
companyId: company.id, runId: run.id, now: new Date(),
|
||||
})).toMatchObject({ outcome: "not_promoted" });
|
||||
expect(await deps.getRun(run.id)).toMatchObject({ status: "scheduled_retry" });
|
||||
} finally {
|
||||
release();
|
||||
await cancellation;
|
||||
}
|
||||
expect(await deps.getRun(run.id)).toMatchObject({ status: "cancelled" });
|
||||
});
|
||||
|
||||
it.each(["ephemeral", "reuse_by_environment"] as const)("keeps %s cancellation eligible for retention when teardown loses the task lock", async leasePolicy => {
|
||||
const { company, agent } = await fixture();
|
||||
const [issue] = await db.insert(issues).values({ companyId: company.id, title: "Retain cancelled work",
|
||||
|
||||
@@ -233,6 +233,8 @@ function readNonEmptyString(value: unknown): string | null {
|
||||
}
|
||||
|
||||
export type CancelRunOptions = {
|
||||
/** Return null if a parked retry was promoted or settled before this cancellation. */
|
||||
expectedStatus?: "scheduled_retry";
|
||||
budgetEnforcement?: BudgetEnforcementScope;
|
||||
/** Optional board request identity, atomically reserved for native Stop. */
|
||||
cancellationRequestId?: string;
|
||||
@@ -899,6 +901,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont
|
||||
) {
|
||||
let run = await getRun(runId);
|
||||
if (!run) throw notFound("Heartbeat run not found");
|
||||
if (options.expectedStatus && run.status !== options.expectedStatus) return null;
|
||||
if (options.cancellationRequestId) {
|
||||
run = await claimCancellationRequest(db, runId, run.companyId, options.cancellationRequestId, options.cancellationRequestedByUserId ?? null);
|
||||
}
|
||||
@@ -929,7 +932,8 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont
|
||||
return run;
|
||||
const agent = await getAgent(run.agentId);
|
||||
const errorCode = options.errorCode ?? "cancelled";
|
||||
const cancellation = requestedRunCancellation(options.resultJson ?? {}, reason);
|
||||
const cancellation = (options.expectedStatus && run.resultJson?.startupCancellation
|
||||
? readRunCancellation(run.resultJson) : null) ?? requestedRunCancellation(options.resultJson ?? {}, reason);
|
||||
|
||||
const pendingProcessCancellation = processRunCancellationSettlements.get(
|
||||
run.id,
|
||||
@@ -952,7 +956,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont
|
||||
// Established legacy processes must still be stopped if the database is
|
||||
// unavailable. Only native or not-yet-dispatched preparation needs this
|
||||
// additional durable fence before its existing cancellation path.
|
||||
if (options.budgetEnforcement || (!options.cancellationRequestId && (run.runtimeMode === "native" || (!run.runtimeModeResolvedAt && !running && !control)))) {
|
||||
if (options.expectedStatus || options.budgetEnforcement || (!options.cancellationRequestId && (run.runtimeMode === "native" || (!run.runtimeModeResolvedAt && !running && !control)))) {
|
||||
const fence = async (tx: Db) => {
|
||||
const [fenced] = await tx.update(heartbeatRuns).set({
|
||||
// Record handoff intent before native cancellation can finalize and release
|
||||
@@ -963,13 +967,14 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont
|
||||
${JSON.stringify({ cancellation })}::jsonb ||
|
||||
jsonb_build_object('startupCancellation', jsonb_build_object(
|
||||
'requestedAt', ${new Date().toISOString()}::text,
|
||||
'errorCode', ${errorCode}::text,
|
||||
'beforeNativeSelection', ${heartbeatRuns.runtimeMode} = 'legacy'
|
||||
and ${heartbeatRuns.runtimeModeResolvedAt} is null
|
||||
and ${heartbeatRuns.executionStage} = 'preparing'
|
||||
and coalesce(${heartbeatRuns.runnerProfileJson}->'adapterDispatch'->>'adapterType' = 'paperclip_runner', false)
|
||||
))`,
|
||||
}).where(and(eq(heartbeatRuns.id, runId), inArray(heartbeatRuns.status,
|
||||
pendingNativeRetry ? [...CANCELLABLE_HEARTBEAT_RUN_STATUSES, "failed"] : [...CANCELLABLE_HEARTBEAT_RUN_STATUSES],
|
||||
options.expectedStatus ? [options.expectedStatus] : pendingNativeRetry ? [...CANCELLABLE_HEARTBEAT_RUN_STATUSES, "failed"] : [...CANCELLABLE_HEARTBEAT_RUN_STATUSES],
|
||||
))).returning();
|
||||
return fenced ?? null;
|
||||
};
|
||||
@@ -984,7 +989,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont
|
||||
}
|
||||
if (!fenced) {
|
||||
stopOwnership?.release();
|
||||
return getRun(runId);
|
||||
return options.expectedStatus ? null : getRun(runId);
|
||||
}
|
||||
run = fenced;
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user