mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-31 06:04:11 +01:00
fix: continue interrupted task conversations with bounded retries (#13237)
## Thinking Path > - Paperclip manages AI agents and their tasks. > - A task can outlive a provider process or a server restart. > - Legacy recovery treated unknown tool outcomes as a permanent execution hold. > - That hold could also reject a later user message. > - A conversation turn can use prior history without replaying prior tool calls. > - This pull request lets supported conversation adapters continue within the existing retry budget. > - Users can send a new message after automatic attempts stop. ## Linked Issues or Issue Description **What happened?** A server restart could interrupt a local ACP run and leave its task behind a permanent recovery hold. A later user message could be cancelled before the provider answered. The immediate recovery path could also create a successor outside the durable failure counter. **Expected behavior** Continue with a bounded new conversation turn. Preserve a compatible provider session or use full task context when it is unavailable. Do not replay recorded tools. When automatic attempts stop, allow a new user request through the normal execution gates. **Steps to reproduce** 1. Start a task with a local conversation adapter. 2. Restart the server while the provider is working. 3. Let the previous run become interrupted. 4. Send a follow-up message and observe the recovery hold on the old behavior. Related work: Refs #13075 for durable task recovery. Refs #12946 for retry-limit and checkout-lock handling. This change routes conversation recovery through the existing bounded scheduler. ## What Changed - Mark supported local conversation failures for continuation. Keep native-runner and non-conversation recovery rules. - Carry an interruption notice into the next turn. Retain stopped ACP session history even when a write outcome is unknown. - Clear unavailable ACP sessions so the next bounded attempt can use full task context. - Route immediate failure recovery through the same durable scheduler as process-loss recovery. Release only the predecessor checkout when its retry takes ownership. - Retire obsolete conversation holds using immutable run evidence, in bounded batches with an activity record. Preserve outcome evidence and do not wake historical tasks. - Block actual admission and Resume while a predecessor process or environment lease is still active. Keep the original interruption notice after a rejected wake. Preserve the upstream blocked-wake waiting contract: bounded retry planning can happen during cleanup, while deferred messages and execution remain gated. - Add subprocess and database regression tests. Update the execution contract. - Add the current thread-status field to the native recovery provider fixture so its damaged-journal test reaches the intended boundary. Tolerate an already-exited fixture process during test cleanup while still asserting both processes terminate. ## Verification - Workspace typecheck passed: `pnpm -r typecheck`. - Build passed: `pnpm build`. - Module boundaries passed: `pnpm check:module-boundaries`. - Focused tests passed: 293 recovery/session/dispatch tests, 66 retry and response-gate tests, and 37 native-session tests. Some suites overlap. - Tests cover interrupted writes, missing sessions, concurrent retries, restart persistence, pending questions and approvals, execution gates, and historical holds. - Built the Rust test executables with `pnpm --filter @paperclipai/paperclip-runner build:rust` for native-runner verification. - Full Vitest coverage verified locally using the repository’s general and serialized shards, with focused reruns for failures and files not reached after a shard stopped. The ownership-gate regression is fixed and the complete affected server shard passes (1,390 tests). Local parallel runs also hit temporary-directory, resource, and timing failures; those suites pass with canonical temporary paths and sequential reruns. No test timeouts were increased. - Final merged-branch regression run: 577 tests pass across process recovery, retry scheduling, liveness, durable chat, wake-queue application/adapter, dispatch, continuation, native sessions, and task chat. Earlier focused verification also passed 19 native control tests. Token gates and whitespace validation pass. - Browser verification passed all three ACP Stop/continue/pause scenarios, including a rerun after merging the upstream waiting behavior: `PAPERCLIP_E2E_PORT=3397 pnpm test:e2e tests/e2e/acp-stop-continuation.spec.ts`. The interrupted-write case verifies that follow-up completes without a repeated write. - Final-head [CI run 34625037394](https://github.com/paperclipai/paperclip/actions/runs/34625037394) passed on `06ac4bd9d150f8b209a96e5fd609c696958794a0`: all 31 reported checks are green, including server/workspace suites, all browser shards, native runner verification, build, typecheck, release dry run, and aggregate gates. The two conditional Storybook checks were skipped. Greptile reviewed this exact commit at 5/5; all review threads are resolved. ## Risks - A new model turn can choose to repeat an action. Paperclip does not replay recorded tool calls and does not certify unknown action outcomes. - Conversation adapters now stop after their retry budget instead of requiring action reconciliation. Explicit Stop, pause, dependency, approval, budget, and ownership gates remain in force. - No schema migration or dependency changes. Historical holds are folded without changing task status or waking work. ## Model Used OpenAI GPT-6 through Codex, with reasoning, repository tools, code execution, and test execution. The session does not expose a more specific model build ID or context-window size. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
42a4f5b15b
commit
b1efd65edc
32 files changed
+696
-168
No files matched your search
@@ -22,6 +22,7 @@ import {
|
||||
createWakeAdmissionWriter,
|
||||
} from "./postgres.js";
|
||||
import type { WakeQueuePostgresAdapterDeps } from "./postgres.js";
|
||||
import { createReleaseIssueExecution } from "../application/use-cases.js";
|
||||
import type { TransactionScope } from "../application/ports.js";
|
||||
|
||||
// Proves the atomicity and company-scope properties the security review
|
||||
@@ -163,6 +164,38 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => {
|
||||
return id;
|
||||
}
|
||||
|
||||
for (const hasDeferredMessage of [false, true]) {
|
||||
it(`plans conversation recovery during owner cleanup without draining messages (queued=${hasDeferredMessage})`, async () => {
|
||||
const companyId = await seedCompany();
|
||||
const agentId = await seedAgent({ companyId });
|
||||
const issueId = await seedIssue({ companyId, assigneeAgentId: agentId });
|
||||
const runId = await seedRun({ companyId, agentId, status: "failed", contextSnapshot: { issueId } });
|
||||
await db.update(heartbeatRuns).set({
|
||||
processPid: process.pid,
|
||||
resultJson: { conversationContinuation: "continue_conversation_v1" },
|
||||
}).where(eq(heartbeatRuns.id, runId));
|
||||
const wakeId = hasDeferredMessage ? await seedDeferredWake({ companyId, agentId, issueId }) : null;
|
||||
const release = createReleaseIssueExecution({
|
||||
issueLock: createPostgresWakeQueueAdapter(db, stubDeps),
|
||||
recovery: {
|
||||
escalateStrandedAssignedIssue: async () => { throw new Error("unexpected escalation"); },
|
||||
escalateStrandedRecoveryIssueInPlace: async () => { throw new Error("unexpected escalation"); },
|
||||
},
|
||||
});
|
||||
const result = await release({ companyId, runId, now: new Date() });
|
||||
expect(result.outcome.kind).toBe("released");
|
||||
expect(result.postCommitEffects.every((effect) => effect.kind === "conversation_retry_requested")).toBe(true);
|
||||
if (!wakeId) expect(result.postCommitEffects).toEqual([
|
||||
{ kind: "conversation_retry_requested", companyId, runId, reviewParticipant: false },
|
||||
]);
|
||||
if (wakeId) {
|
||||
const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wakeId));
|
||||
expect(wake.status).toBe("deferred_issue_execution");
|
||||
}
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId))).toHaveLength(1);
|
||||
});
|
||||
}
|
||||
|
||||
it("leaves deferred work untouched until the effective execution hold clears", async () => {
|
||||
const companyId = await seedCompany();
|
||||
const agentId = await seedAgent({ companyId });
|
||||
|
||||
@@ -12,6 +12,7 @@ import {
|
||||
issues,
|
||||
nativeRunFinalizations,
|
||||
} from "@paperclipai/db";
|
||||
import { hasConversationContinuationPolicy } from "../../../services/conversation-continuation.js";
|
||||
import { legacyExecutionNeedsReconciliation } from "../../../services/legacy-execution-recovery.js";
|
||||
import {
|
||||
authorizeFailedChatRunRetryWake,
|
||||
@@ -85,6 +86,7 @@ function toRunSnapshot(row: HeartbeatRunRow): RunSnapshot {
|
||||
agentId: row.agentId,
|
||||
status: row.status,
|
||||
runtimeMode: row.runtimeMode,
|
||||
conversationContinuation: row.runtimeMode === "legacy" && hasConversationContinuationPolicy(row.resultJson),
|
||||
errorCode: row.errorCode,
|
||||
responsibleUserId: row.responsibleUserId,
|
||||
contextSnapshot: parseObject(row.contextSnapshot),
|
||||
@@ -1061,12 +1063,18 @@ export function createPostgresWakeQueueAdapter(db: Db, deps: WakeQueuePostgresAd
|
||||
return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot };
|
||||
}
|
||||
|
||||
// A release must leave deferred messages intact while replay is held.
|
||||
if (await getExecutionBlocker(tx, issueRow.companyId, issueRow.id)) {
|
||||
// A release must leave deferred messages intact while execution is held.
|
||||
// The finishing conversation may still own its lease until finally cleanup.
|
||||
// Allow only bounded retry planning in that case; admission stays gated.
|
||||
const executionBlocker = await getExecutionBlocker(tx, issueRow.companyId, issueRow.id);
|
||||
const recoveryOnly = Boolean(executionBlocker &&
|
||||
executionBlocker.cause === "execution_owner_active" && executionBlocker.runId === run.id &&
|
||||
runSnapshot.conversationContinuation && ["failed", "timed_out", "interrupted"].includes(run.status));
|
||||
if (executionBlocker && !recoveryOnly) {
|
||||
return { outcome: { kind: "released" }, postCommitEffects: [], run: runSnapshot };
|
||||
}
|
||||
|
||||
const locked: LockedIssueExecution = { primaryIssue: toIssueSnapshot(issueRow), run: runSnapshot };
|
||||
const locked: LockedIssueExecution = { primaryIssue: toIssueSnapshot(issueRow), run: runSnapshot, recoveryOnly };
|
||||
const result = await fn(locked, { host: buildHost(tx, deps), transaction: buildTransaction(tx, deps, db, run) });
|
||||
return { ...result, run: runSnapshot };
|
||||
});
|
||||
|
||||
@@ -12,6 +12,8 @@ export type { InvokableAgentSnapshot, IssueSnapshot, ReleaseRecoveryBlockedNotic
|
||||
|
||||
/** The primary issue a locked release resolves to, plus the finishing run the lock step already loaded. */
|
||||
export type LockedIssueExecution = {
|
||||
/** Plan bounded recovery without draining messages while the finishing owner cleans up. */
|
||||
recoveryOnly?: boolean;
|
||||
primaryIssue: IssueSnapshot;
|
||||
run: RunSnapshot;
|
||||
};
|
||||
|
||||
@@ -21,6 +21,8 @@ export type RunSnapshot = {
|
||||
contextSnapshot: Record<string, unknown>;
|
||||
/** `resultJson.configurationIncomplete`, already parsed; non-null only on a configuration-incomplete failed run. */
|
||||
configurationIncompletePayload: Record<string, unknown> | null;
|
||||
/** The host schedules failed conversation turns with its durable retry budget. */
|
||||
conversationContinuation?: boolean;
|
||||
};
|
||||
|
||||
export type IssueSnapshot = {
|
||||
@@ -67,7 +69,12 @@ export type IssueReopenedEffect = {
|
||||
};
|
||||
|
||||
/** Explicit post-commit work a caller applies only after the release transaction commits. */
|
||||
export type PostCommitEffect = RunQueuedEffect | IssueReopenedEffect;
|
||||
export type PostCommitEffect = RunQueuedEffect | IssueReopenedEffect | {
|
||||
kind: "conversation_retry_requested";
|
||||
companyId: string;
|
||||
runId: string;
|
||||
reviewParticipant: boolean;
|
||||
};
|
||||
|
||||
export type ReleaseOutcome =
|
||||
| { kind: "released" }
|
||||
|
||||
@@ -139,6 +139,10 @@ async function runReleaseDrain(
|
||||
const issue = locked.primaryIssue;
|
||||
const postCommitEffects: PostCommitEffect[] = [];
|
||||
|
||||
if (locked.recoveryOnly) {
|
||||
return runReleaseRecoveryTail(issue, run, ports.host, ports.transaction, input, postCommitEffects);
|
||||
}
|
||||
|
||||
// Each `continue` path below leaves the wake row off the
|
||||
// `deferred_issue_execution` status, so the next queue read cannot
|
||||
// return that same row again. That invariant is what ends this loop.
|
||||
@@ -554,6 +558,19 @@ async function runReleaseRecoveryTail(
|
||||
"wake-queue: queued a recovery run with no invokable recovery agent",
|
||||
);
|
||||
|
||||
if (run.conversationContinuation && ["failed", "timed_out", "interrupted"].includes(run.status)) {
|
||||
// Do not create an uncounted immediate successor inside the issue lock.
|
||||
// The host's idempotent scheduler claims it after commit with the same
|
||||
// retry counter used by restart and process-loss recovery.
|
||||
postCommitEffects.push({
|
||||
kind: "conversation_retry_requested",
|
||||
companyId: run.companyId,
|
||||
runId: run.id,
|
||||
reviewParticipant: decision.kind === "queue_review_participant_recovery",
|
||||
});
|
||||
return { outcome: { kind: "released" }, postCommitEffects };
|
||||
}
|
||||
|
||||
const sessionBefore = await host.resolveSessionBeforeForWakeup({
|
||||
companyId: issue.companyId,
|
||||
agentId: recoveryAgent.id,
|
||||
|
||||
Reference in new issue
Block a user