mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
fix: preserve conversation retries through execution cleanup (#15463)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Recovery schedules bounded retries after a provider disconnects. > - A stopped run can still hold its environment lease while cleanup runs. > - Retrying before that lease is released cancels the new run before it starts and spends another retry. > - Restoring the task can also send the worker repair instructions from an already resolved recovery action. > - This pull request preserves the waiting retry and removes settled recovery instructions from later wakes. > - The task can continue after cleanup without an operator repairing the same incident again. ## Linked Issues or Issue Description **What happened?** A legacy conversation run disconnected while it was doing ordinary work. Its environment cleanup took longer than the retry delay. Two retries were cancelled before dispatch with `execution_reconciliation_required`. Those cancellations exhausted the failure budget. After an operator restored the task, the wake still told the original worker to repair the runtime and hand the task back to itself. **Expected behavior** Cleanup waits preserve the pending attempt. The same retry can continue after ownership is released, subject to all current gates. Once a recovery action is resolved or cancelled, subsequent task wakes omit its repair instructions. **Steps to reproduce** 1. Fail a legacy conversation run while its environment lease remains in `pending_cleanup`. 2. Schedule a bounded retry and run promotion before cleanup releases that lease. 3. Repeat the scheduler sweep. Before this fix, retries promote and then cancel without starting. 4. Resolve a stranded-task recovery action and build the restored task wake with that action ID. Before this fix, the wake still includes the settled repair instructions. **Paperclip version or commit** Reproduced with database regressions against `ceabc3bc880` on master. Related: #15019 restores a skipped assignment handoff after lease release. #15235 filters stale handoff evidence in the recovery sweep. This change preserves an existing scheduled conversation retry and corrects restored wake content. It does not create a new handoff wake. ## What Changed - Keep an unstarted legacy conversation retry on the same durable row while prior execution ownership remains active. Recheck after 30 seconds without increasing retry accounting. - Return a queued retry to scheduled state if it encounters that hold at the claim gate. Retain its issue claim and publish the status change. - Record one local lifecycle diagnostic per blocking run. Remove that wait marker on promotion. - Include recovery action metadata only while the referenced action is active or escalated. - Return an explicit `waiting` response and the saved schedule when Retry now meets cleanup. Show the wait inline without a false success or disabled button. - Add database, rendered-prompt, route, and UI regressions. Document the execution and run-log contracts. ## Verification - Five cleanup and restored-wake regressions fail against the original production code. The Retry now route and UI regressions also fail before their correction. - Related retry, dispatch, stale-queue, and recovery suites: 321 tests pass across seven files. All ten focused cleanup/restored-wake cases pass after rebase. The final dispatch adjustment passes all 46 adapter tests. - Retry now routes and affected UI suites: all 46 tests pass. The tests cover repeated clicks, the saved schedule, unchanged accounting, promotion after release, and no false success or error state. - `pnpm -r typecheck`, `pnpm build`, `pnpm build-storybook`, and `pnpm check:token-gates` pass. - Full local `pnpm test:run` was started and then stopped after the final commit passed all GitHub CI test shards. No complete local full-suite result is claimed; CI supplies the complete test result for the final commit. - Final head `6f1058a332c039e33c4f002b296d20a5554e760e`: all 55 GitHub checks are green or intentionally skipped. Apex review is 5/5 after two reviews, with no unresolved threads. The PR has no merge conflicts. ## Risks The wait applies only to unstarted legacy conversation retries. Native runs and non-conversation execution keep their existing recovery rules. Cleanup must actually release ownership before execution can resume. The wait does not fix a cleanup service that never finishes. Promotion and dispatch still enforce cancellation, reassignment, pause, budget, and reconciliation gates. No schema change is required. The Retry now response adds a `waiting` outcome; the shared contract and all three UI controls handle it. ## Model Used OpenAI Codex based on GPT-6, with repository analysis, tool use, and local code execution. The runtime does not expose the precise serving model ID or context window. ## 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 (focused suites; complete final-head test coverage in CI) - [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
db0bee19fa
commit
06484b3c41
14 files changed
+341
-31
No files matched your search
@@ -211,6 +211,22 @@ New comments received during an execution hold retain their individual deferred
|
||||
|
||||
The conversation groups repeated empty pre-start reconciliation cancellations into a neutral waiting notice. Started runs, actual startup failures, and run history remain inspectable. No historical run records are deleted.
|
||||
|
||||
An unstarted legacy conversation retry waits on its existing `scheduled_retry`
|
||||
row while a previous execution still owns a process or environment lease. Each
|
||||
sweep moves the next check forward without creating another run or incrementing
|
||||
retry accounting. The wait survives a controller restart. A queued retry that
|
||||
encounters the same ownership hold returns to that scheduled state. Once
|
||||
ownership is released, normal promotion and dispatch gates still apply,
|
||||
including reassignment, cancellation, budget limits, and no-replay holds.
|
||||
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.
|
||||
|
||||
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
|
||||
does not clear any persisted execution reconciliation hold.
|
||||
|
||||
Workspace contention (`workspace_busy`) displays **Waiting for workspace** and
|
||||
continues automatically when the workspace is available. Internal scheduling
|
||||
attempts remain in the run log without conversation cancellation markers,
|
||||
|
||||
@@ -280,6 +280,14 @@ warnings and errors remain visible.
|
||||
|
||||
Recovery lifecycle events retain the original structured failure code, retry attempt, next retry time, and predecessor/successor identifiers. Durable status delivery uses an idempotency marker; delivery grants no provider authority. Failed publication is retried without repeating provider work. These records are not first-party Telemetry.
|
||||
|
||||
An unstarted legacy conversation retry waiting for execution cleanup retains
|
||||
`resultJson.executionWait` with the issue, blocking run, and
|
||||
`execution_owner_active` cause. A local lifecycle event records the wait and the
|
||||
unchanged scheduled retry attempt once per blocking run. Repeated scheduler
|
||||
checks update the same retry row without repeating that event. Promotion removes
|
||||
this ownership-wait marker after cleanup; other execution holds retain their
|
||||
existing dispatch gates.
|
||||
|
||||
If execution-continuation setup finds that a task no longer exists, is closed,
|
||||
or its owner changed, the existing cancellation settlement records
|
||||
`continuation_task_ownership_changed`. The run and wake request become cancelled
|
||||
|
||||
@@ -645,6 +645,7 @@ export interface IssueScheduledRetry {
|
||||
export type IssueRetryNowOutcome =
|
||||
| "promoted"
|
||||
| "already_promoted"
|
||||
| "waiting"
|
||||
| "no_scheduled_retry"
|
||||
| "gate_suppressed";
|
||||
|
||||
|
||||
@@ -6,6 +6,7 @@ import express from "express";
|
||||
import request from "supertest";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { renderPaperclipWakePrompt } from "@paperclipai/adapter-utils/server-utils";
|
||||
import {
|
||||
costEvents,
|
||||
agents,
|
||||
@@ -1785,6 +1786,29 @@ describeEmbeddedPostgres("issue recovery actions", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it.each(["resolved", "cancelled"])("omits %s recovery instructions when the original worker resumes", async (status) => {
|
||||
const { companyId, coderId, sourceIssueId } = await seedCompany();
|
||||
const action = await issueRecoveryActionService(db).upsertSourceScoped({
|
||||
companyId, sourceIssueId, kind: "stranded_assigned_issue", ownerType: "board",
|
||||
returnOwnerAgentId: coderId, cause: "stranded_assigned_issue", fingerprint: "finished-recovery",
|
||||
evidence: { failureSummary: "Retry budget exhausted during cleanup." },
|
||||
nextAction: "Repair the runtime, then retry the original owner.", wakePolicy: null,
|
||||
});
|
||||
await db.update(issueRecoveryActions).set({ status, outcome: "handed_back", resolvedAt: new Date() })
|
||||
.where(eq(issueRecoveryActions.id, action.id));
|
||||
const payload = await buildPaperclipWakePayload({ db, companyId, contextSnapshot: {
|
||||
issueId: sourceIssueId, wakeReason: "issue_recovery_action_restored",
|
||||
recoveryActionId: action.id, recoveryCause: action.cause,
|
||||
} });
|
||||
expect(payload?.issue?.id).toBe(sourceIssueId);
|
||||
expect(payload?.recovery).toBeNull();
|
||||
const prompt = renderPaperclipWakePrompt(payload);
|
||||
expect(prompt).toContain("Implement backend recovery");
|
||||
expect(prompt).not.toContain("Recovery contract:");
|
||||
expect(prompt).not.toContain(action.nextAction);
|
||||
expect(prompt).not.toContain("Do not produce the deliverable");
|
||||
});
|
||||
|
||||
it("accepts new verified evidence after an automatic no-replay disposition without reopening on duplicate requests", async () => {
|
||||
const { companyId, coderId, sourceIssueId } = await seedCompany();
|
||||
const runId = randomUUID();
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import express from "express";
|
||||
import request from "supertest";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { and, eq, inArray } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
|
||||
import {
|
||||
activityLog,
|
||||
@@ -9,6 +9,7 @@ import {
|
||||
agentWakeupRequests,
|
||||
companies,
|
||||
createDb,
|
||||
environmentLeases,
|
||||
heartbeatRunEvents,
|
||||
heartbeatRuns,
|
||||
issueComments,
|
||||
@@ -44,6 +45,7 @@ describeEmbeddedPostgres("issue scheduled retry routes", () => {
|
||||
}, 20_000);
|
||||
|
||||
afterEach(async () => {
|
||||
await db.delete(environmentLeases);
|
||||
await db.delete(issueComments);
|
||||
await db.delete(issueRelations);
|
||||
await db.delete(issueTreeHolds);
|
||||
@@ -304,6 +306,37 @@ describeEmbeddedPostgres("issue scheduled retry routes", () => {
|
||||
expect(retryRuns[0]).toMatchObject({ id: retryRunId, status: "queued" });
|
||||
});
|
||||
|
||||
it("reports cleanup waits with the saved schedule and promotes the same retry after release", async () => {
|
||||
const f = await seedIssueWithRetry();
|
||||
await db.update(heartbeatRuns)
|
||||
.set({ resultJson: { conversationContinuation: "continue_conversation_v1" } })
|
||||
.where(inArray(heartbeatRuns.id, [f.sourceRunId, f.retryRunId]));
|
||||
const [lease] = await db.insert(environmentLeases).values({
|
||||
companyId: f.companyId, issueId: f.issueId, heartbeatRunId: f.sourceRunId,
|
||||
status: "pending_cleanup", cleanupStatus: "failed",
|
||||
}).returning();
|
||||
const app = createApp(boardActor(f.companyId));
|
||||
for (let click = 0; click < 2; click++) {
|
||||
const before = Date.now();
|
||||
const res = await request(app).post(`/api/issues/${f.issueId}/scheduled-retry/retry-now`).send({});
|
||||
expect(res.status, JSON.stringify(res.body)).toBe(200);
|
||||
expect(res.body).toMatchObject({ outcome: "waiting", scheduledRetry: {
|
||||
runId: f.retryRunId, status: "scheduled_retry", scheduledRetryAttempt: 2,
|
||||
} });
|
||||
expect(res.body.message).toContain("Waiting for execution cleanup");
|
||||
const [saved] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.retryRunId));
|
||||
expect(res.body.scheduledRetry.scheduledRetryAt).toBe(saved!.scheduledRetryAt!.toISOString());
|
||||
expect(saved!.scheduledRetryAt!.getTime()).toBeGreaterThanOrEqual(before + 30_000);
|
||||
}
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.sourceRunId))).toHaveLength(1);
|
||||
await db.update(environmentLeases).set({ status: "released", cleanupStatus: "success", releasedAt: new Date() })
|
||||
.where(eq(environmentLeases.id, lease!.id));
|
||||
const resumed = await request(app).post(`/api/issues/${f.issueId}/scheduled-retry/retry-now`).send({});
|
||||
expect(resumed.body).toMatchObject({ outcome: "promoted", scheduledRetry: {
|
||||
runId: f.retryRunId, status: "queued", scheduledRetryAttempt: 2,
|
||||
} });
|
||||
});
|
||||
|
||||
it("returns a clear no-op response when there is no scheduled retry", async () => {
|
||||
const companyId = randomUUID();
|
||||
const issueId = randomUUID();
|
||||
|
||||
@@ -8,6 +8,7 @@ import {
|
||||
createDb,
|
||||
documentRevisions,
|
||||
documents,
|
||||
environmentLeases,
|
||||
heartbeatRunEvents,
|
||||
heartbeatRuns,
|
||||
issueDocuments,
|
||||
@@ -51,6 +52,7 @@ describeEmbeddedPostgres("run-dispatch postgres adapter", () => {
|
||||
}, 20_000);
|
||||
|
||||
afterEach(async () => {
|
||||
await db.delete(environmentLeases);
|
||||
await db.delete(activityLog);
|
||||
await db.delete(issueDocuments);
|
||||
await db.delete(documentRevisions);
|
||||
@@ -154,6 +156,115 @@ describeEmbeddedPostgres("run-dispatch postgres adapter", () => {
|
||||
return runId;
|
||||
}
|
||||
|
||||
async function seedCleanupRetry(status: "scheduled_retry" | "queued" = "scheduled_retry") {
|
||||
const { companyId, agentId } = await seedCompanyAndAgent();
|
||||
const issueId = randomUUID();
|
||||
const now = new Date();
|
||||
await seedIssue({ companyId, issueId, assigneeAgentId: agentId, status: "in_progress" });
|
||||
const sourceId = await seedRun({ companyId, agentId, contextSnapshot: { issueId } });
|
||||
await db.update(heartbeatRuns).set({ status: "failed", finishedAt: now,
|
||||
resultJson: { conversationContinuation: "continue_conversation_v1", workspaceRestoreFailure: "restore_failed" },
|
||||
}).where(eq(heartbeatRuns.id, sourceId));
|
||||
const [lease] = await db.insert(environmentLeases).values({ companyId, issueId,
|
||||
heartbeatRunId: sourceId, status: "pending_cleanup", cleanupStatus: "failed",
|
||||
}).returning();
|
||||
const runId = await seedRun({ companyId, agentId, status, now,
|
||||
contextSnapshot: { issueId, retryOfRunId: sourceId, executionRetryAccounting: { version: 1, failureRetries: 1, maxTurnContinuations: 0 } },
|
||||
scheduledRetryReason: "transient_failure",
|
||||
});
|
||||
await db.update(heartbeatRuns).set({ retryOfRunId: sourceId, scheduledRetryAttempt: 1,
|
||||
resultJson: { conversationContinuation: "continue_conversation_v1" },
|
||||
}).where(eq(heartbeatRuns.id, runId));
|
||||
await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId));
|
||||
return { companyId, agentId, issueId, sourceId, leaseId: lease!.id, runId, now };
|
||||
}
|
||||
|
||||
it("keeps one scheduled retry through cleanup and promotes it once after release", async () => {
|
||||
const f = await seedCleanupRetry();
|
||||
let adapter = createPostgresRunDispatchAdapter(db);
|
||||
for (let attempt = 0; attempt < 4; attempt++) {
|
||||
expect(await adapter.promoteOrCancelDueRetry({ ...f, now: new Date(f.now.getTime() + attempt * 60_000) }))
|
||||
.toMatchObject({ outcome: "not_promoted" });
|
||||
// Reconstruct the service to prove the wait survives controller restart.
|
||||
adapter = createPostgresRunDispatchAdapter(db);
|
||||
}
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
expect(run).toMatchObject({ status: "scheduled_retry", startedAt: null, finishedAt: null,
|
||||
retryOfRunId: f.sourceId, scheduledRetryAttempt: 1,
|
||||
contextSnapshot: { executionRetryAccounting: { failureRetries: 1 } },
|
||||
});
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(2);
|
||||
expect(await db.select().from(issueRecoveryActions)).toHaveLength(0);
|
||||
const events = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.runId));
|
||||
expect(events.filter(e => e.message?.includes("waiting for execution cleanup"))).toHaveLength(1);
|
||||
await db.update(environmentLeases).set({ status: "released", cleanupStatus: "success", releasedAt: new Date() })
|
||||
.where(eq(environmentLeases.id, f.leaseId));
|
||||
const now = new Date(f.now.getTime() + 300_000);
|
||||
const outcomes = await Promise.all([
|
||||
adapter.promoteOrCancelDueRetry({ ...f, now }), adapter.promoteOrCancelDueRetry({ ...f, now }),
|
||||
]);
|
||||
expect(outcomes.filter(o => o.outcome === "promoted")).toHaveLength(1);
|
||||
const [resumed] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
expect(resumed).toMatchObject({ status: "queued", scheduledRetryAttempt: 1 });
|
||||
expect(resumed.resultJson).not.toHaveProperty("executionWait");
|
||||
expect(await adapter.cancelStaleQueuedRun({ ...f, now, expectedStatus: "queued" }))
|
||||
.toMatchObject({ outcome: "not_stale" });
|
||||
});
|
||||
|
||||
it("parks the same unstarted retry if cleanup becomes pending after promotion", async () => {
|
||||
const f = await seedCleanupRetry("queued");
|
||||
const adapter = createPostgresRunDispatchAdapter(db);
|
||||
expect(await adapter.cancelStaleQueuedRun({ ...f, expectedStatus: "queued" }))
|
||||
.toMatchObject({ outcome: "deferred" });
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
expect(run).toMatchObject({ status: "scheduled_retry", scheduledRetryAttempt: 1, startedAt: null, finishedAt: null });
|
||||
expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].executionRunId).toBe(f.runId);
|
||||
});
|
||||
|
||||
it("still suppresses a cleanup-waiting retry when the task is cancelled", async () => {
|
||||
const f = await seedCleanupRetry();
|
||||
await db.update(issues).set({ status: "cancelled" }).where(eq(issues.id, f.issueId));
|
||||
expect(await createPostgresRunDispatchAdapter(db).promoteOrCancelDueRetry(f))
|
||||
.toMatchObject({ outcome: "gate_suppressed", errorCode: "issue_cancelled" });
|
||||
});
|
||||
|
||||
it("still suppresses a cleanup-waiting retry after reassignment", async () => {
|
||||
const f = await seedCleanupRetry();
|
||||
const newOwnerId = randomUUID();
|
||||
await seedAgent({ id: newOwnerId, companyId: f.companyId, name: "New owner" });
|
||||
await db.update(issues).set({ assigneeAgentId: newOwnerId }).where(eq(issues.id, f.issueId));
|
||||
expect(await createPostgresRunDispatchAdapter(db).promoteOrCancelDueRetry(f))
|
||||
.toMatchObject({ outcome: "gate_suppressed", errorCode: "issue_reassigned" });
|
||||
});
|
||||
|
||||
it("retains a no-replay hold after cleanup releases ownership", async () => {
|
||||
const f = await seedCleanupRetry();
|
||||
const adapter = createPostgresRunDispatchAdapter(db);
|
||||
await db.insert(issueRecoveryActions).values({
|
||||
companyId: f.companyId, sourceIssueId: f.issueId, kind: "active_run_watchdog",
|
||||
ownerType: "board", cause: "uncertain_external_action", status: "resolved",
|
||||
evidence: { automaticRecovery: { replay: "blocked" } }, fingerprint: f.sourceId,
|
||||
nextAction: "Verify the external action before continuing.",
|
||||
});
|
||||
expect(await adapter.promoteOrCancelDueRetry(f)).toMatchObject({ outcome: "not_promoted" });
|
||||
await db.update(environmentLeases).set({ status: "released", cleanupStatus: "success", releasedAt: new Date() })
|
||||
.where(eq(environmentLeases.id, f.leaseId));
|
||||
const now = new Date(f.now.getTime() + 60_000);
|
||||
expect(await adapter.promoteOrCancelDueRetry({ ...f, now })).toMatchObject({ outcome: "promoted" });
|
||||
expect(await adapter.cancelStaleQueuedRun({ ...f, now, expectedStatus: "queued" }))
|
||||
.toMatchObject({ outcome: "cancelled", errorCode: "execution_reconciliation_required" });
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ cause: "uncertain_external_action" });
|
||||
});
|
||||
|
||||
it.each(["native", "started", "non-conversation"])("does not park a %s run behind cleanup", async (mode) => {
|
||||
const f = await seedCleanupRetry("queued");
|
||||
await db.update(heartbeatRuns).set(mode === "native" ? { runtimeMode: "native" }
|
||||
: mode === "started" ? { startedAt: f.now } : { resultJson: {} })
|
||||
.where(eq(heartbeatRuns.id, f.runId));
|
||||
expect(await createPostgresRunDispatchAdapter(db).cancelStaleQueuedRun({ ...f, expectedStatus: "queued" }))
|
||||
.toMatchObject({ outcome: "cancelled", errorCode: "execution_reconciliation_required" });
|
||||
});
|
||||
|
||||
async function seedContinuationSummary(input: {
|
||||
companyId: string;
|
||||
issueId: string;
|
||||
|
||||
@@ -643,7 +643,12 @@ export function createPostgresRunDispatchAdapter(
|
||||
> {
|
||||
const [row] = await tx
|
||||
.update(heartbeatRuns)
|
||||
.set({ status: "queued", updatedAt: input.now })
|
||||
.set({
|
||||
status: "queued",
|
||||
updatedAt: input.now,
|
||||
resultJson: sql`case when ${heartbeatRuns.resultJson}->'executionWait'->>'cause' = 'execution_owner_active'
|
||||
then ${heartbeatRuns.resultJson} - 'executionWait' else ${heartbeatRuns.resultJson} end`,
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(heartbeatRuns.id, input.runId),
|
||||
@@ -832,6 +837,11 @@ export function createPostgresRunDispatchAdapter(
|
||||
: { outcome: { outcome: "not_promoted" as const }, telemetryRun: null };
|
||||
}
|
||||
|
||||
const issueId = factsResult.facts.issueId;
|
||||
if (issueId && (await deferRetryForCleanupInTx(tx, run, issueId, now))) {
|
||||
return { outcome: { outcome: "not_promoted" as const }, telemetryRun: null };
|
||||
}
|
||||
|
||||
const promoted = await promoteDueRetryInTx(tx as unknown as Db, {
|
||||
runId: input.runId,
|
||||
companyId: input.companyId,
|
||||
@@ -955,9 +965,9 @@ export function createPostgresRunDispatchAdapter(
|
||||
async function decideCurrentRunStaleness(tx: Db, run: HeartbeatRun, now: Date) {
|
||||
const contextSnapshot = parseObject(run.contextSnapshot);
|
||||
const issueId = readNonEmptyString(contextSnapshot.issueId);
|
||||
if (!issueId) return { issueId: null, facts: null, decision: { stale: false as const } };
|
||||
if (!issueId) return { issueId: null, facts: null, executionBlocker: null, decision: { stale: false as const } };
|
||||
const recovery = await getExecutionBlocker(tx, run.companyId, issueId, { conversationResetCommentId: deriveCommentId(contextSnapshot) });
|
||||
if (recovery) return { issueId, facts: null, decision: { stale: true as const,
|
||||
if (recovery) return { issueId, facts: null, executionBlocker: recovery, decision: { stale: true as const,
|
||||
errorCode: "execution_reconciliation_required" as const, reason: recovery.nextAction,
|
||||
details: { issueId, recoveryActionId: recovery.recoveryActionId },
|
||||
} };
|
||||
@@ -973,7 +983,67 @@ export function createPostgresRunDispatchAdapter(
|
||||
now,
|
||||
tx,
|
||||
);
|
||||
return { issueId, facts, decision: decideQueuedRunStaleness(facts, now) };
|
||||
return { issueId, facts, executionBlocker: null, decision: decideQueuedRunStaleness(facts, now) };
|
||||
}
|
||||
|
||||
/**
|
||||
* Keep the existing attempt durable while the previous execution releases
|
||||
* ownership. Only an unstarted legacy conversation retry can wait here;
|
||||
* reconciliation of uncertain actions and native execution retain their gates.
|
||||
* Both callers hold the issue and run locks, in that order.
|
||||
*/
|
||||
async function deferRetryForCleanupInTx(
|
||||
tx: Db,
|
||||
run: HeartbeatRun,
|
||||
issueId: string,
|
||||
now: Date,
|
||||
observedBlocker?: Awaited<ReturnType<typeof getExecutionBlocker>>,
|
||||
) {
|
||||
if (
|
||||
run.runtimeMode !== "legacy" || !run.retryOfRunId || run.startedAt ||
|
||||
run.processPid || run.processGroupId ||
|
||||
!hasConversationContinuationPolicy(run.resultJson) ||
|
||||
!["scheduled_retry", "queued"].includes(run.status)
|
||||
) return null;
|
||||
// Use the dispatch gate's observation when supplied. A second read could
|
||||
// see cleanup finish and accidentally cancel the retry for the old hold.
|
||||
const blocker = observedBlocker === undefined
|
||||
? await getExecutionBlocker(tx, run.companyId, issueId)
|
||||
: observedBlocker;
|
||||
if (blocker?.cause !== "execution_owner_active") return null;
|
||||
|
||||
const previousWait = parseObject(run.resultJson?.executionWait);
|
||||
const scheduledRetryAt = new Date(now.getTime() + 30_000);
|
||||
const executionWait = { issueId, runId: blocker.runId, cause: blocker.cause };
|
||||
const [waiting] = await tx.update(heartbeatRuns)
|
||||
.set({
|
||||
status: "scheduled_retry",
|
||||
scheduledRetryAt,
|
||||
updatedAt: now,
|
||||
contextSnapshot: { ...parseObject(run.contextSnapshot), scheduledRetryAt: scheduledRetryAt.toISOString() },
|
||||
resultJson: { ...parseObject(run.resultJson), executionWait },
|
||||
})
|
||||
.where(and(
|
||||
eq(heartbeatRuns.id, run.id),
|
||||
eq(heartbeatRuns.companyId, run.companyId),
|
||||
eq(heartbeatRuns.status, run.status),
|
||||
))
|
||||
.returning();
|
||||
if (!waiting) return null;
|
||||
// Repeated sweeps retain one wait diagnostic, not a new attempt or event.
|
||||
if (previousWait.cause !== blocker.cause || previousWait.runId !== blocker.runId) {
|
||||
await appendHeartbeatRunEvent(tx, {
|
||||
companyId: run.companyId,
|
||||
runId: run.id,
|
||||
agentId: run.agentId,
|
||||
eventType: "lifecycle",
|
||||
stream: "system",
|
||||
level: "info",
|
||||
message: "Scheduled retry is waiting for execution cleanup; the same attempt will continue after ownership is released",
|
||||
payload: { ...executionWait, scheduledRetryAttempt: run.scheduledRetryAttempt },
|
||||
});
|
||||
}
|
||||
return { outcome: "deferred" as const, postCommitEffects: [statusEffect(waiting, run.status)] };
|
||||
}
|
||||
|
||||
async function cancelStaleQueuedRun(
|
||||
@@ -981,7 +1051,7 @@ export function createPostgresRunDispatchAdapter(
|
||||
): Promise<CancelStaleQueuedRunOutcome> {
|
||||
const cancelLockedRun = async (tx: Db, run: HeartbeatRun) => {
|
||||
if (run.status !== input.expectedStatus) return { outcome: "lost_race" as const };
|
||||
const { issueId, facts, decision } = await decideCurrentRunStaleness(tx, run, input.now);
|
||||
const { issueId, facts, decision, executionBlocker } = await decideCurrentRunStaleness(tx, run, input.now);
|
||||
if (!decision.stale || !issueId) {
|
||||
if (input.expectedStatus === "queued" && facts?.isInteractionWake) {
|
||||
// Preserve the authority accepted under the issue/run locks. Later
|
||||
@@ -996,6 +1066,10 @@ export function createPostgresRunDispatchAdapter(
|
||||
}
|
||||
return { outcome: "not_stale" as const };
|
||||
}
|
||||
if (input.expectedStatus === "queued" && decision.errorCode === "execution_reconciliation_required") {
|
||||
const deferred = await deferRetryForCleanupInTx(tx, run, issueId, input.now, executionBlocker);
|
||||
if (deferred) return deferred;
|
||||
}
|
||||
return cancelStaleRunInTx(tx, run, issueId, decision, input.expectedStatus, input.now);
|
||||
};
|
||||
|
||||
|
||||
@@ -46,6 +46,7 @@ export type PromoteScheduledRetryOutcome =
|
||||
|
||||
export type CancelStaleQueuedRunOutcome =
|
||||
| { outcome: "not_stale" }
|
||||
| { outcome: "deferred"; postCommitEffects: PostCommitEffect[] }
|
||||
/** The run's status no longer matched the phase the caller expected; a concurrent writer already won. */
|
||||
| { outcome: "lost_race" }
|
||||
| {
|
||||
|
||||
@@ -8162,9 +8162,11 @@ export async function buildPaperclipWakePayload(input: {
|
||||
.then((rows) => rows[0] ?? null)
|
||||
: null;
|
||||
const recoveryEvidence = parseObject(recoveryAction?.evidence);
|
||||
const executionAlreadyReconciled =
|
||||
recoveryAction?.status === "resolved" &&
|
||||
Boolean(recoveryEvidence.executionReconciliation);
|
||||
// A restored task resumes its deliverable, not the completed repair. A
|
||||
// referenced settled/missing action must not fall back to a stale cause.
|
||||
const recoveryStillActive = recoveryAction
|
||||
? ["active", "escalated"].includes(recoveryAction.status)
|
||||
: !recoveryActionId && Boolean(recoveryCause);
|
||||
const originalAssigneeId =
|
||||
recoveryAction?.returnOwnerAgentId ??
|
||||
recoveryAction?.previousOwnerAgentId ??
|
||||
@@ -8208,7 +8210,7 @@ export async function buildPaperclipWakePayload(input: {
|
||||
attachmentOmissions,
|
||||
externalChatProvider,
|
||||
recovery:
|
||||
!executionAlreadyReconciled && (recoveryAction || recoveryCause)
|
||||
recoveryStillActive
|
||||
? {
|
||||
cause: recoveryAction?.cause ?? recoveryCause,
|
||||
failureSummary: readNonEmptyString(recoveryEvidence.failureSummary),
|
||||
@@ -16876,17 +16878,25 @@ export function heartbeatService(
|
||||
if (promotion.outcome === "promoted") {
|
||||
applyRunDispatchPostCommitEffects(promotion.postCommitEffects);
|
||||
}
|
||||
const promotedRow = await getIssueRetryRun(issue.companyId, issue.id, [
|
||||
"queued",
|
||||
"running",
|
||||
"cancelled",
|
||||
]);
|
||||
const scheduledRetry = promotedRow
|
||||
? summarizeIssueScheduledRetryRun(promotedRow)
|
||||
: summarizeIssueScheduledRetryRun({
|
||||
run: updated,
|
||||
// Promotion can preserve this row as a cleanup wait. Read that exact run,
|
||||
// not an older cancelled retry or the pre-promotion schedule.
|
||||
const currentRun = await getRun(updated.id);
|
||||
const scheduledRetry = currentRun
|
||||
? summarizeIssueScheduledRetryRun({
|
||||
run: currentRun,
|
||||
agentName: scheduled.agentName,
|
||||
});
|
||||
})
|
||||
: null;
|
||||
|
||||
if (currentRun?.status === "scheduled_retry") {
|
||||
return {
|
||||
outcome: "waiting" as const,
|
||||
message: parseObject(currentRun.resultJson?.executionWait).cause === "execution_owner_active"
|
||||
? "Waiting for execution cleanup. Paperclip will retry automatically once cleanup finishes."
|
||||
: "The retry remains scheduled. Paperclip will check again at the scheduled time.",
|
||||
scheduledRetry,
|
||||
};
|
||||
}
|
||||
|
||||
if (promotion.outcome === "promoted") {
|
||||
return {
|
||||
@@ -16902,10 +16912,17 @@ export function heartbeatService(
|
||||
scheduledRetry,
|
||||
};
|
||||
}
|
||||
if (currentRun && ["queued", "running"].includes(currentRun.status)) {
|
||||
return {
|
||||
outcome: "already_promoted" as const,
|
||||
message: "Scheduled retry was already promoted",
|
||||
scheduledRetry,
|
||||
};
|
||||
}
|
||||
return {
|
||||
outcome: "already_promoted" as const,
|
||||
message: "Scheduled retry was already promoted",
|
||||
scheduledRetry,
|
||||
outcome: "no_scheduled_retry" as const,
|
||||
message: "No live scheduled retry exists for this issue",
|
||||
scheduledRetry: null,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -17634,11 +17651,16 @@ export function heartbeatService(
|
||||
companyId: run.companyId,
|
||||
expectedStatus: "queued",
|
||||
});
|
||||
if (staleness.outcome === "cancelled") {
|
||||
if (staleness.outcome === "cancelled" || staleness.outcome === "deferred") {
|
||||
applyRunDispatchPostCommitEffects(staleness.postCommitEffects);
|
||||
logger.info(
|
||||
{ runId: run.id, issueId, errorCode: staleness.errorCode },
|
||||
"claimQueuedRun: cancelled stale queued run",
|
||||
{
|
||||
runId: run.id,
|
||||
issueId,
|
||||
outcome: staleness.outcome,
|
||||
errorCode: staleness.outcome === "cancelled" ? staleness.errorCode : undefined,
|
||||
},
|
||||
"claimQueuedRun: withheld queued run at the execution gate",
|
||||
);
|
||||
return null;
|
||||
}
|
||||
|
||||
@@ -95,7 +95,9 @@ function SuccessfulRunRetryNowControl({
|
||||
<div className="mt-2 rounded-md border border-amber-300/70 bg-background/80 p-2 dark:border-amber-500/40 dark:bg-background/40">
|
||||
<div className="flex flex-col gap-2 sm:flex-row sm:items-center sm:justify-between">
|
||||
<div className="min-w-0 text-xs leading-5 text-amber-900 dark:text-amber-100">
|
||||
Paperclip will ask the assignee to choose the next step {scheduleLabel}. Retry now starts that follow-up immediately.
|
||||
{retryNow.data?.outcome === "waiting" && retryNow.data.scheduledRetry?.runId === scheduledRetry.runId
|
||||
? retryNow.data.message
|
||||
: <>Paperclip will ask the assignee to choose the next step {scheduleLabel}. Retry now starts that follow-up immediately.</>}
|
||||
</div>
|
||||
<Button
|
||||
type="button"
|
||||
|
||||
@@ -221,6 +221,19 @@ describe("IssueScheduledRetryCard", () => {
|
||||
expect(container.querySelector('[data-testid="issue-scheduled-retry-error-band"]')).toBeNull();
|
||||
});
|
||||
|
||||
it("shows cleanup waiting without claiming success or disabling Retry now", async () => {
|
||||
const message = "Waiting for execution cleanup. Paperclip will retry automatically once cleanup finishes.";
|
||||
retryNowMock.mockResolvedValue({ outcome: "waiting", message, scheduledRetry: baseRetry });
|
||||
renderWithProviders(<IssueScheduledRetryCard issueId="issue-1" scheduledRetry={baseRetry} />);
|
||||
act(() => { getRetryNowButton()!.click(); });
|
||||
await waitForUi(() => {
|
||||
expect(getCard()?.textContent).toContain(message);
|
||||
expect(getCard()?.textContent).not.toContain("run starting");
|
||||
expect(getRetryNowButton()!.disabled).toBe(false);
|
||||
expect(container.querySelector('[data-testid="issue-scheduled-retry-error-band"]')).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
it("renders an inline error band on backend failure", async () => {
|
||||
retryNowMock.mockRejectedValue(new Error("Server error"));
|
||||
renderWithProviders(
|
||||
|
||||
@@ -162,7 +162,9 @@ export function IssueScheduledRetryCard({
|
||||
? retryNow.data?.outcome === "already_promoted"
|
||||
? "Already promoted — run starting"
|
||||
: "Promoted — run starting"
|
||||
: helperIdle}
|
||||
: retryNow.data?.outcome === "waiting" && retryNow.data.scheduledRetry?.runId === scheduledRetry.runId
|
||||
? retryNow.data.message
|
||||
: helperIdle}
|
||||
</span>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
@@ -1542,9 +1542,11 @@ export function IssueProperties({
|
||||
? retryNow.data?.outcome === "already_promoted"
|
||||
? "Already promoted — run starting"
|
||||
: "Promoted — run starting"
|
||||
: scheduledRetryIsContinuation
|
||||
? "Pulls continuation forward immediately"
|
||||
: "Pulls retry forward immediately"}
|
||||
: retryNow.data?.outcome === "waiting" && retryNow.data.scheduledRetry?.runId === scheduledRetry.runId
|
||||
? retryNow.data.message
|
||||
: scheduledRetryIsContinuation
|
||||
? "Pulls continuation forward immediately"
|
||||
: "Pulls retry forward immediately"}
|
||||
</span>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
@@ -24,6 +24,7 @@ function readErrorMessage(error: unknown): string {
|
||||
export const RETRY_NOW_OUTCOME_HEADLINE: Record<IssueRetryNowOutcome, string> = {
|
||||
promoted: "Retry promoted",
|
||||
already_promoted: "Retry already running",
|
||||
waiting: "Retry waiting",
|
||||
no_scheduled_retry: "No scheduled retry",
|
||||
gate_suppressed: "Couldn't retry now",
|
||||
};
|
||||
|
||||
Reference in new issue
Block a user