fix(recovery): reject stale productive continuation wakes (#13173)
## Thinking Path > - Paperclip manages agent work through tasks and runs. > - Recovery continues assigned work when no live execution path remains. > - A recovery sweep can read an in-progress task before its run completes. > - The sweep can then observe the successful run after completion has changed the task status. > - This pull request checks current status and assignment under the existing enqueue lock. > - A stale continuation leaves a skipped wake receipt and creates no run. > - Task chat also omits an empty continuation cancelled before it started because its task had become terminal. ## Linked Issues or Issue Description Related public work: #10779 and #8419. Those older open changes address terminal disposition across other recovery paths. This change uses the existing scheduler guard for productive successful-run continuation and adds real database lock contention coverage. **What happened?** Recovery could combine an old in-progress task snapshot with a newer successful run. It queued an automatic continuation after the task was done. Dispatch cancelled that run before it started, but task chat displayed “Couldn't start” below the successful answer. This can happen after the native runner's finish result has already been accepted. It does not require a missing comment. **Expected behavior** Productive continuation must remain eligible when enqueueing acquires the task lock. Completion, cancellation, reassignment, or a move away from in-progress must prevent creation of the run. Actual execution stops must remain visible. **Steps to reproduce** 1. Let recovery select an assigned in-progress task whose latest run succeeded with productive progress. 2. Hold the task row lock in another transaction and change the task to done. 3. Let recovery attempt to enqueue while that transaction holds the lock. 4. Commit completion. Before this fix, recovery creates a redundant run from the stale snapshot. **Paperclip version or commit** Reproduced against master at `4042eb1c4` with deterministic integration tests. **Deployment mode** Built from source with PostgreSQL. The bug is in core recovery and is not adapter-specific. ## What Changed - Pass the existing status-and-assignee guard for productive terminal continuation recovery. - Preserve a skipped wake receipt with the expected and actual task state, without creating a run. - Test actual PostgreSQL lock contention for native and legacy completion, cancellation, backlog, review, blocked state, and reassignment. - Omit empty redundant pre-start cancellations from native and legacy task chat. Preserve stop markers for runs that started. - Document recovery eligibility at enqueue time. ## Verification - All seven new race cases failed before the guard was connected. - `pnpm -r typecheck` passed. - `pnpm build` passed. - `pnpm check:token-gates` passed. - Task chat suite: 98 tests passed. - Recovery integration suites: 290 tests passed, including 31 stale-queue tests. - Full CI verification passed on `7ea71f04d`: all 31 active checks succeeded, including all test shards, browser tests, build, typecheck, and canary release dry run. Storybook visual regression was skipped by its path filter. - Greptile reviewed this commit at 5/5 with no review threads. - The local `pnpm test:run` aggregate reported a setup failure in the unchanged `tool-access-service.test.ts` suite. Its isolated rerun passed all 231 tests without edits. The duplicate aggregate was stopped after the complete CI matrix passed; it is not counted as a successful local full-suite run. ## Risks Low risk. The backend guard applies only to productive successful-run recovery. It requires the task to remain in-progress with the same agent. Other wake sources keep their current policy. The UI change only suppresses empty redundant cancellations; run records remain available. No schema change or migration is required. ## Model Used OpenAI GPT-6 through Codex, with reasoning, repository inspection, code edits, and local test execution. The exact deployment snapshot and context window are not exposed by 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 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:
Dotta
GitHub
parent
a05b828bcd
commit
d10cbde815
|
|
@ -552,6 +552,8 @@ Recovery rule:
|
|||
|
||||
This is an active-work continuity recovery.
|
||||
|
||||
After a productive successful run, recovery checks that the issue is still `in_progress` and assigned to the same agent under the enqueue transaction's issue lock. The sweep's earlier snapshot cannot authorize a continuation after completion, cancellation, reassignment, or a move to another status. A mismatch records a skipped wake receipt without creating a run. An empty queued continuation cancelled because the issue became terminal is omitted from task chat; its cancellation remains in the run log. Runs that actually started still show their stop state.
|
||||
|
||||
The same bounded rule applies when the previous heartbeat reported waiting on a local/background watcher and that watcher was killed, disappeared, or was never represented by a durable Paperclip primitive. Paperclip queues at most one continuation for the same recovery fingerprint. If the continuation also leaves only local watcher evidence, Paperclip must surface a real blocker or explicit recovery action instead of repeating continuation recovery. A new monitor, scheduled wake, healthy delegated blocker issue, or other durable source mutation resolves that recovery fingerprint normally.
|
||||
|
||||
#### Deliberate wait is not a lost run
|
||||
|
|
|
|||
|
|
@ -25,6 +25,7 @@ import {
|
|||
heartbeatService,
|
||||
} from "../services/heartbeat.ts";
|
||||
import { runningProcesses } from "../adapters/index.ts";
|
||||
import { recoveryService } from "../services/recovery/service.ts";
|
||||
|
||||
const mockAdapterExecute = vi.hoisted(() =>
|
||||
vi.fn(async () => ({
|
||||
|
|
@ -395,6 +396,93 @@ describeEmbeddedPostgres("heartbeat stale queued-run invalidation", () => {
|
|||
]);
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ runtimeMode: "native", status: "done", reassigned: false },
|
||||
{ runtimeMode: "legacy", status: "done", reassigned: false },
|
||||
{ runtimeMode: "native", status: "cancelled", reassigned: false },
|
||||
{ runtimeMode: "native", status: "backlog", reassigned: false },
|
||||
{ runtimeMode: "native", status: "in_review", reassigned: false },
|
||||
{ runtimeMode: "native", status: "blocked", reassigned: false },
|
||||
{ runtimeMode: "native", status: "in_progress", reassigned: true },
|
||||
] as const)("skips stale $runtimeMode productive recovery after status=$status reassigned=$reassigned commits under the enqueue lock", async ({ runtimeMode, status, reassigned }) => {
|
||||
const { companyId, agentId } = await seedCompanyAndAgent();
|
||||
const issueId = randomUUID();
|
||||
const runId = randomUUID();
|
||||
await db.insert(issues).values({
|
||||
id: issueId,
|
||||
companyId,
|
||||
title: "Completion racing with productive recovery",
|
||||
status: "in_progress",
|
||||
assigneeAgentId: agentId,
|
||||
});
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: runId,
|
||||
companyId,
|
||||
agentId,
|
||||
invocationSource: "assignment",
|
||||
runtimeMode,
|
||||
status: "succeeded",
|
||||
livenessState: "completed",
|
||||
contextSnapshot: { issueId, wakeReason: "issue_assigned" },
|
||||
startedAt: new Date(),
|
||||
finishedAt: new Date(),
|
||||
});
|
||||
|
||||
// Let the real sweep select its in-progress snapshot, then hold the issue
|
||||
// lock until the real enqueue transaction is waiting on the newer state.
|
||||
const enqueueWakeup = vi.fn(async (...[targetAgentId, options]: Parameters<typeof heartbeat.wakeup>) => {
|
||||
let pendingWake!: ReturnType<typeof heartbeat.wakeup>;
|
||||
await db.transaction(async (tx) => {
|
||||
await tx.update(issues).set({
|
||||
status,
|
||||
...(reassigned ? { assigneeAgentId: null, assigneeUserId: "responsible-user" } : {}),
|
||||
}).where(eq(issues.id, issueId));
|
||||
const [{ pid }] = await tx.execute<{ pid: number }>(sql`select pg_backend_pid() as pid`);
|
||||
pendingWake = heartbeat.wakeup(targetAgentId, options);
|
||||
try {
|
||||
expect(await waitForCondition(async () => {
|
||||
const [{ waiting }] = await db.execute<{ waiting: boolean }>(sql`
|
||||
select exists (
|
||||
select 1 from pg_stat_activity
|
||||
where ${pid} = any(pg_blocking_pids(pid))
|
||||
) as waiting
|
||||
`);
|
||||
return waiting;
|
||||
})).toBe(true);
|
||||
} catch (error) {
|
||||
// Observe a pending rejection even if the lock assertion fails.
|
||||
void pendingWake.catch(() => {});
|
||||
throw error;
|
||||
}
|
||||
});
|
||||
return pendingWake;
|
||||
});
|
||||
const recovery = recoveryService(db, { enqueueWakeup });
|
||||
const result = await recovery.reconcileStrandedAssignedIssues();
|
||||
|
||||
expect(enqueueWakeup).toHaveBeenCalledOnce();
|
||||
expect(result).toMatchObject({ continuationRequeued: 0, escalated: 0, skipped: 1, issueIds: [] });
|
||||
expect(await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns)).toEqual([{ id: runId }]);
|
||||
expect(await db.select().from(issueComments)).toHaveLength(0);
|
||||
expect(mockAdapterExecute).not.toHaveBeenCalled();
|
||||
const [wakeup] = await db.select().from(agentWakeupRequests);
|
||||
expect(wakeup).toMatchObject({
|
||||
status: "skipped",
|
||||
reason: "issue_state_guard_mismatch",
|
||||
runId: null,
|
||||
payload: {
|
||||
heartbeatSkip: {
|
||||
expectedStatuses: ["in_progress"],
|
||||
actualStatus: status,
|
||||
expectedAssigneeAgentId: agentId,
|
||||
actualAssigneeAgentId: reassigned ? null : agentId,
|
||||
},
|
||||
},
|
||||
});
|
||||
const [issue] = await db.select().from(issues).where(eq(issues.id, issueId));
|
||||
expect(issue).toMatchObject({ status, assigneeAgentId: reassigned ? null : agentId });
|
||||
});
|
||||
|
||||
it("cancels a resolved connection-intent wake parked before queued-run claim", async () => {
|
||||
const { companyId, agentId } = await seedCompanyAndAgent();
|
||||
const issueId = randomUUID();
|
||||
|
|
|
|||
|
|
@ -182,6 +182,10 @@ type RecoveryWakeupOptions = {
|
|||
requestedByActorType?: "user" | "agent" | "system";
|
||||
requestedByActorId?: string | null;
|
||||
contextSnapshot?: Record<string, unknown>;
|
||||
issueStateGuard?: {
|
||||
statuses: string[];
|
||||
assigneeAgentId: string;
|
||||
};
|
||||
};
|
||||
|
||||
type RecoveryWakeup = (
|
||||
|
|
@ -1921,6 +1925,17 @@ export function recoveryService(
|
|||
source: "automation",
|
||||
triggerDetail: "system",
|
||||
reason: input.reason,
|
||||
// The sweep can combine an old in-progress issue snapshot with a newer
|
||||
// successful run. Validate eligibility under the enqueue issue lock so
|
||||
// completion or reassignment cannot create a redundant continuation.
|
||||
...(input.source === "issue.productive_terminal_continuation_recovery"
|
||||
? {
|
||||
issueStateGuard: {
|
||||
statuses: ["in_progress"],
|
||||
assigneeAgentId: input.agentId,
|
||||
},
|
||||
}
|
||||
: {}),
|
||||
payload: withRecoveryContext(
|
||||
{
|
||||
issueId: input.issueId,
|
||||
|
|
|
|||
|
|
@ -1229,31 +1229,72 @@ describe("TaskChatThread runtime transcript selection", () => {
|
|||
},
|
||||
);
|
||||
|
||||
it("does not show a completed-response notice for a redundant cancelled continuation", () => {
|
||||
render(
|
||||
<TaskChatThread
|
||||
comments={[]}
|
||||
onAdd={async () => {}}
|
||||
linkedRuns={[
|
||||
{
|
||||
runId: "connection-continuation-skipped",
|
||||
status: "cancelled",
|
||||
errorCode: "issue_not_in_progress",
|
||||
startedAt: null,
|
||||
agentId: "agent-1",
|
||||
agentName: "Runner",
|
||||
adapterType: "paperclip_runner",
|
||||
createdAt: "2026-09-07T18:00:00.000Z",
|
||||
finishedAt: "2026-09-07T18:00:01.000Z",
|
||||
},
|
||||
]}
|
||||
/>,
|
||||
);
|
||||
expect(container.textContent).not.toContain(
|
||||
"The runner returned no user-facing response.",
|
||||
);
|
||||
expect(container.textContent).not.toContain("Run completed");
|
||||
});
|
||||
it.each([
|
||||
["legacy", "issue_not_in_progress"],
|
||||
["native", "issue_not_in_progress"],
|
||||
["legacy", "issue_terminal_status"],
|
||||
["native", "issue_terminal_status"],
|
||||
] as const)(
|
||||
"hides a redundant cancelled continuation (%s, %s)",
|
||||
(runtimeMode, errorCode) => {
|
||||
render(
|
||||
<TaskChatThread
|
||||
comments={[]}
|
||||
onAdd={async () => {}}
|
||||
linkedRuns={[
|
||||
{
|
||||
runId: "connection-continuation-skipped",
|
||||
runtimeMode,
|
||||
status: "cancelled",
|
||||
errorCode,
|
||||
startedAt: null,
|
||||
agentId: "agent-1",
|
||||
agentName: "Runner",
|
||||
adapterType: "paperclip_runner",
|
||||
createdAt: "2026-09-07T18:00:00.000Z",
|
||||
finishedAt: "2026-09-07T18:00:01.000Z",
|
||||
},
|
||||
]}
|
||||
/>,
|
||||
);
|
||||
expect(container.textContent).not.toContain(
|
||||
"The runner returned no user-facing response.",
|
||||
);
|
||||
expect(container.textContent).not.toContain("Run completed");
|
||||
expect(container.textContent).not.toContain("Couldn't start");
|
||||
expect(container.textContent).not.toContain("Run cancelled");
|
||||
expect(container.textContent).not.toContain("before returning an answer");
|
||||
},
|
||||
);
|
||||
|
||||
it.each(["legacy", "native"] as const)(
|
||||
"keeps a cancellation visible when the %s run had already started",
|
||||
(runtimeMode) => {
|
||||
render(
|
||||
<TaskChatThread
|
||||
comments={[]}
|
||||
onAdd={async () => {}}
|
||||
linkedRuns={[
|
||||
{
|
||||
runId: "started-cancellation",
|
||||
runtimeMode,
|
||||
status: "cancelled",
|
||||
errorCode: "issue_terminal_status",
|
||||
agentId: "agent-1",
|
||||
agentName: "Runner",
|
||||
adapterType: "paperclip_runner",
|
||||
createdAt: "2026-09-07T18:00:00.000Z",
|
||||
startedAt: "2026-09-07T18:00:00.500Z",
|
||||
finishedAt: "2026-09-07T18:00:01.000Z",
|
||||
},
|
||||
]}
|
||||
/>,
|
||||
);
|
||||
expect(container.textContent).toContain(
|
||||
runtimeMode === "native" ? "Run cancelled" : "Stopped",
|
||||
);
|
||||
},
|
||||
);
|
||||
|
||||
it("does not treat a progress comment as the final response of a failed native run", () => {
|
||||
nativeTranscriptState.transcriptByRun.set("native-progress-failed", [
|
||||
|
|
|
|||
|
|
@ -1384,6 +1384,18 @@ export function TaskChatThread(props: TaskChatThreadProps) {
|
|||
if (liveRun && source.id === liveRun.id) continue;
|
||||
const entries = transcriptByRun.get(source.id) ?? [];
|
||||
const meta = linkedRunMetaById.get(source.id);
|
||||
// A queued continuation can become unnecessary while another turn finishes
|
||||
// the task. Keep that cancellation in the run log, not the conversation.
|
||||
// Apply this before native stop markers are assembled as well.
|
||||
if (
|
||||
source.status === "cancelled" &&
|
||||
entries.length === 0 &&
|
||||
(meta?.errorCode === "issue_not_in_progress" ||
|
||||
(meta?.errorCode === "issue_terminal_status" && !meta.startedAt))
|
||||
) {
|
||||
settledRunIds.add(source.id);
|
||||
continue;
|
||||
}
|
||||
const acceptedSummary = acceptedSemanticResultSummary(meta?.resultJson);
|
||||
const parsedSource = transcriptToTaskChatItems(entries, {
|
||||
runId: source.id,
|
||||
|
|
@ -1516,16 +1528,6 @@ export function TaskChatThread(props: TaskChatThreadProps) {
|
|||
});
|
||||
}
|
||||
if (entries.length === 0) {
|
||||
// A queued continuation cancelled after the task was completed or parked
|
||||
// never produced a provider turn. Keep its record in the run log without
|
||||
// presenting it as a completed chat response.
|
||||
if (
|
||||
source.status === "cancelled" &&
|
||||
meta?.errorCode === "issue_not_in_progress"
|
||||
) {
|
||||
settledRunIds.add(source.id);
|
||||
continue;
|
||||
}
|
||||
if (sourceIsPaperclipRunner && sourceYielded) {
|
||||
settledRunIds.add(source.id);
|
||||
continue;
|
||||
|
|
|
|||
Loading…
Reference in New Issue