fix(chat): resume task Retry after retained-source containment (#17317)

## Thinking Path

> - Paperclip manages agent work through task and execution ownership.
> - A task Retry must retain the user's request while the system
verifies an old executor.
> - Connected conversations also retain their external principal and
accepted source batch.
> - Chat Retry admission required containment before it could save a
wait for that recovery.
> - A saved chat receipt could also remain deferred after containment
became available.
> - This pull request lets the exact recovery-bound request wait and
resume through its chat owner.
> - Normal admission still checks containment, current access and policy
before one successor starts.

## Linked Issues or Issue Description

**What happened?**

An ordinary task Retry on a connected conversation failed with
`failed_chat_retry_not_authorized` while its retained remote executor
still needed verification. The new recovery generation existed, but the
chat action failed before it could save a dispatch wait. This occurred
after the repairs in #17267 were confirmed live.

The local regression found a second gap. A processed chat action with a
deferred wake was not reconsidered after containment. Re-entering normal
admission also tried to insert its existing durable receipt again. A
historical hold for the same failed run could also hide the current
Retry's workspace strategy, so the request still waited after the old
executor was contained. Preparation could then read that sibling instead
of the recovery record that owned the fresh successor, losing the
fresh-workspace decision.

**Expected behavior**

Save the exact authorized Retry while its matching recovery generation
verifies the executor. Once containment is recorded, reconsider the same
chat receipt through normal admission. Start one successor with the
original source batch. Keep publication, principal, pause and budget
checks.

**Steps to reproduce**

1. Admit a GitHub or Telegram request and record a failed native run
with a retained remote lease.
2. Leave the executor's termination unconfirmed and exhaust its prior
recovery generation.
3. Use ordinary task Retry to create a new recovery generation.
4. Process the connected Retry. The old admission code fails instead of
retaining a wait.
5. With a deferred receipt, record exact containment and reconcile
again. The old dispatch path leaves it stranded.

**Paperclip version or commit**

Reproduced from `412914cfa78d5abefcc34d3eb8d4bb86e461caac`. The branch
is rebased onto current master.

**Deployment mode**

Hosted tenant for the reported failure. Isolated local PostgreSQL and
the assembled heartbeat/chat services for the regression.

Refs #17267 and #16965. This change does not add a Cloud dependency or
change the absence protocol.

## What Changed

- Permit a saved admission wait only for the exact current task Retry
and active source recovery generation.
- Keep promotion and execution dependent on independent containment
proof.
- Bind a Retry to active automatic recovery without replacing its
generation, attempt count or deadline.
- Share the current exact Retry strategy across historical holds for the
same run. Retain each hold's stop and effect checks.
- Select the canonical recovery record that proves the committed fresh
successor during workspace preparation.
- Let task receipt reconciliation return the same deferred action to its
chat owner.
- Recheck current board and external authority, then update the existing
wake atomically with its successor.
- Preserve the exact chat source batch. Keep generic board-comment
filtering unchanged.
- Add GitHub and Telegram cases for waiting, containment, one successor,
stale generations, source edits, endpoint pause and supersession.
- Update recovery documentation and capability maps.

## Verification

- Red: real PostgreSQL tests reproduced the admission rejection for
GitHub and Telegram. Six active-recovery cases then failed on the
generation-ID assumption. Two same-run historical-hold cases remained
deferred after containment on the old strategy lookup. A preparation
regression then returned no readiness when the historical hold was
stored first.
- Green: 73 selected chat integration tests and 394 continuation, task
Retry, workspace recovery, recovery-state and workspace-preparation
tests passed. The chat tests use the real chat admission and heartbeat
queue. Unrelated capacity stays occupied so no real provider starts.
They assert no executor before containment, one successor afterward,
unchanged source leases while waiting, preserved recovery budgets, and
denial after authority changes. A separate source hold still blocks
work. Both sibling insertion orders pass through the committed-readiness
lookup and real workspace preparation: the new directory has independent
Git storage and files, and unpublished old work remains intact.
- `pnpm -r typecheck`, `node scripts/check-maps.mjs` and `git diff
--check` passed after the review fixes. The source build passed on
`f6660efa48e37a80cd9857217a537606de85b5c8`.
- The local host reached its PostgreSQL semaphore limit. Successful runs
used a test-only loader to reduce disposable cluster connection and
worker limits. No repository database helper or application setting
changed.
- The full local `pnpm test:run` began before the final review edits,
failed in its first server phase and was stopped after those failures.
Later phases did not run. This is not a full local pass. The transient
Retry-notice case passes after fixture isolation; the four Discord
close-owned publication cases also pass in isolation. Other broad-run
failure causes are not established.
- Exact-head CI and security checks pass on
`f6660efa48e37a80cd9857217a537606de85b5c8`; Greptile is 5/5 with zero
unresolved review threads. Optional Paperclip Review completed neutral
with a 5/5 static assessment of all 11 changed files and no findings; it
did not run tests or hosted acceptance. All 57 current-head checks are
terminal: 54 success, two skipped and the optional neutral review, with
no failures or pending checks.
- The real hosted workflow confirmed that the deployed absence repair
recorded containment. A later ordinary UI Retry was accepted and saved a
durable request, but it remains behind the historical source hold. The
useful existing review remains unchanged. This follow-up is not
deployed; a live successful successor and final task disposition remain
unverified. No missing files are claimed recovered.

## Risks

- The existing task reconciliation sweep can restage a deferred chat
action. This reuses its receipt and generation; it does not grant new
execution or publication authority.
- Dispatch must keep the board Retry identity separate from the
connected request's principal. The regression checks these bindings and
preserves current chat authorization at each phase.
- Historical holds reuse a strategy only from the locked same-run
recovery that names the current Retry intent. A different run, malformed
recovery record or independent source hold remains fenced.
- Unknown executors without a matching active recovery still fail
admission. Promotion and execution require containment even when
admission saved a wait.
- No schema or API shape changes. No credential changes, manual
deployment or service restart.

## Model Used

OpenAI GPT-6 through Codex, with repository tools, code execution and
browser verification. The exact served model variant and context window
are not exposed in this session.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed 2026-10-23 07:51:34 -07:00
1 parent 3bd709df8b
commit 3f1ad6b25b
11 files changed
+398 -35

No files matched your search

+14
View File
@@ -18,6 +18,20 @@ Connected conversations use their existing inbound authorization owner. A board
Retry changes the workspace strategy, not the external principal, source batch,
conversation access or publication authority.
A connected-conversation Retry can save its exact request while the matching
system recovery generation verifies the old executor. Saving that wait grants
no execution authority. The chat owner rechecks the same board intent, external
principal and source batch when it resumes the deferred receipt. Containment,
current access, task eligibility, pause and budget gates must still pass before
one successor starts. Restarts reuse the receipt; they do not create another
Retry generation or another external publication.
Several retained recovery holds can describe the same failed execution. A
historical hold without a system recovery record uses the current exact Retry's
workspace strategy for that run. It does not borrow another run's strategy or
skip any hold's independent containment and effect checks. Attaching a Retry to
active automatic recovery preserves its generation, attempt count and deadline.
Agent Chat additionally requires the current conversation owner and enabled
Agent Chat. Admission and dispatch both recheck these conditions under the
issue lock. An unadmitted receipt from a former owner cannot own the current
+5 -1
View File
@@ -69,7 +69,11 @@ board task-mutation authority, not agent retry authority. The only request
fields are a UUID idempotency key and observed source run (or null). It accepts
no paths, stop attestations or client-selected strategy. The source-scoped
exact-run and maintenance APIs retain their existing contracts. Connected-chat
Retry reuses chat admission and its external identity. See the
Retry reuses chat admission and its external identity. It can persist a wait
only for the exact current system recovery generation. The chat owner resumes
that same deferred receipt through normal admission after containment. The
board intent and external principal remain separate authorities; current task,
connection, pause and budget gates still apply. See the
[durable contract](../../doc/task-recovery.md). Agent Chat also requires its
current conversation owner and feature enablement at admission and locked
dispatch; generic board mutation permission does not replace chat ownership.
+10
View File
@@ -67,6 +67,16 @@ Use the task's Retry action. It can create an isolated attempt after old executi
is contained even when old-file preservation remains unresolved. Approval,
credentials, budget and deliberate pause use their existing specific controls.
For a connected conversation, click Retry while an exact retained source still
needs containment. Observe a saved wait with no new executor. After verified
containment, the same chat receipt must admit one successor with the original
source batch. Reload and repeat reconciliation. Check that a paused endpoint,
changed source, superseded request or revoked access cannot start that work.
The connected-chat database regression in
[chat channel integration](../../server/src/__tests__/chat-channels.integration.test.ts)
exercises admission and queue continuation. It does not establish hosted
provider recovery or successful delivery.
### `post-stop-message`
After a stop, send a new instruction on the same task or inspect one already
@@ -1,5 +1,8 @@
import { idleWorkSnapshot } from "../services/task-admission.js";
import { readFreshWorkspaceReadiness } from "../services/fresh-workspace-continuation.js";
import { requestTaskRetry } from "../services/task-retry.js";
import { createSystemRecovery } from "../services/system-recovery-state.js";
import { remoteTerminationReceipt } from "../services/remote-execution-termination.js";
import { stageReceiptReactionRemovals } from "../services/chat-receipt-reactions.js";
import type { Db } from "@paperclipai/db";
import * as activityLogModule from "../services/activity-log.js";
@@ -66,6 +69,7 @@ import {
eq,
inArray,
isNotNull,
isNull,
like,
notInArray,
or,
@@ -60945,6 +60949,196 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
},
);
it.each((["github", "telegram"] as const).flatMap(provider =>
(["pending", "historical_sibling", "foreign_source_hold", "automatic_scheduled", "automatic_running", "automatic_awaiting_evidence", "exhausted", "wrong_generation", "paused_endpoint", "paused_after_wait", "paused_agent_after_wait", "edited_after_wait", "superseded_after_wait"] as const).map(gate => ({ provider, gate }))))(
"holds an ordinary $provider Retry behind its exact remote recovery: $gate", async ({ provider, gate }) => {
const context = await failedChatRetryFixture(provider, undefined, true);
try {
const companyId = context.fixture.companyId, issueId = context.issue.id;
const leaseId = randomUUID();
await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issueId,
nativePhase: "terminal_failure", runnerProfileJson: {},
resultJson: { workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: {
schema: "paperclip.workspace-restore-recovery.v1", runId: context.runId, leaseIds: [leaseId],
} },
}).where(eq(heartbeatRuns.id, context.runId));
await db.insert(nativeRunFinalizations).values({ companyId, issueId,
runId: context.runId, phase: "terminal_failure" });
await db.insert(environmentLeases).values({ id: leaseId, companyId, issueId,
heartbeatRunId: context.runId, provider: "daytona", providerLeaseId: randomUUID(),
status: "pending_cleanup", cleanupStatus: "failed", leasePolicy: "retain_on_failure",
metadata: { remoteCwd: "/home/daytona/paperclip-workspace", pendingCleanupInFlight: false },
});
const priorRecovery = { ...createSystemRecovery(context.runId),
state: gate === "automatic_scheduled" ? "scheduled" as const
: gate === "automatic_running" ? "running" as const
: gate === "automatic_awaiting_evidence" ? "awaiting_evidence" as const : "exhausted" as const,
attempts: 2, retriesUsed: 1, attemptId: randomUUID(),
firstFailureAt: Date.now() - 1000, deadlineAt: Date.now() + 60_000,
preservationOperationIds: [randomUUID()],
};
const canResume = gate === "pending" || gate === "historical_sibling" || gate.startsWith("automatic_");
await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId,
kind: "active_run_watchdog", status: "resolved", ownerType: "system",
cause: "native_continuation_requires_reconciliation", fingerprint: randomUUID(),
nextAction: "Verify and preserve the exact retained source before continuing.",
evidence: { runId: context.runId, automaticRecovery: { replay: "blocked" },
workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [leaseId] },
historicalWorkspaceSource: { schema: "paperclip.historical-workspace-source.v1", runId: context.runId, leaseIds: [leaseId] },
systemRecovery: priorRecovery,
},
});
const receipt = await requestTaskRetry({ db, companyId, issueId, actorId: "owner-user",
actorSource: "cloud_tenant", observedRunId: context.runId, idempotencyKey: randomUUID(),
prepareChatRetry: context.service.prepareFailedChatRunRetry });
const [recoveryAction] = await db.select().from(issueRecoveryActions).where(and(
eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.sourceIssueId, issueId),
sql`${issueRecoveryActions.evidence}->>'runId' = ${context.runId}`,
));
if (gate === "historical_sibling") await db.insert(issueRecoveryActions).values({
companyId, sourceIssueId: issueId, kind: "active_run_watchdog", status: "resolved", ownerType: "system",
cause: "native_session_retry_exhausted", fingerprint: randomUUID(),
nextAction: "Preserve the historical execution outcome.",
evidence: { runId: context.runId, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } },
});
if (gate.startsWith("automatic_")) expect(recoveryAction.evidence.systemRecovery).toMatchObject({
generationId: priorRecovery.generationId, intentId: receipt.requestId,
attempts: priorRecovery.attempts, retriesUsed: priorRecovery.retriesUsed,
attemptId: priorRecovery.attemptId, deadlineAt: priorRecovery.deadlineAt,
preservationOperationIds: priorRecovery.preservationOperationIds,
});
if (gate === "exhausted" || gate === "wrong_generation") await db.update(issueRecoveryActions).set({
evidence: { ...recoveryAction.evidence, systemRecovery: {
...recoveryAction.evidence.systemRecovery as Record<string, unknown>,
...(gate === "exhausted" ? { state: "exhausted" } : { intentId: randomUUID() }),
} },
}).where(eq(issueRecoveryActions.id, recoveryAction.id));
if (gate === "paused_endpoint") await db.update(chatEndpoints).set({ status: "paused" })
.where(eq(chatEndpoints.id, context.endpoint.id));
// Keep capacity occupied so a regression cannot start a real adapter.
await db.update(agents).set({ adapterType: "paperclip_runner", status: "running",
runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } } })
.where(eq(agents.id, context.fixture.assignedAgentId));
await db.insert(heartbeatRuns).values({ companyId,
agentId: context.fixture.assignedAgentId, status: "running", startedAt: new Date() });
context.wakeup.mockImplementation(heartbeatService(db).wakeup);
const outcome = await context.service.processFailedChatRunRetry(receipt.chatActionId!);
if (["exhausted", "wrong_generation", "paused_endpoint"].includes(gate)) {
expect(outcome).toMatchObject({ status: "failed" });
expect(await readAgentWakeupRequestsById(receipt.chatActionId!)).toHaveLength(0);
expect((await readHeartbeatRunsByCompanyId(companyId))
.filter(run => run.contextSnapshot?.taskRetryRequestId === receipt.requestId)).toHaveLength(0);
return;
}
expect(outcome).toMatchObject({ status: "deferred" });
const [wake] = await readAgentWakeupRequestsById(receipt.chatActionId!);
expect(wake).toMatchObject({ status: "deferred_issue_execution", runId: null,
payload: { taskRetryRequestId: receipt.requestId, retryOfRunId: context.runId } });
const [intent] = await readAgentWakeupRequestsById(receipt.requestId!);
expect(intent).toMatchObject({ status: "deferred_issue_execution", runId: null });
expect((await readHeartbeatRunsByCompanyId(companyId))
.filter(run => run.contextSnapshot?.taskRetryRequestId === receipt.requestId)).toHaveLength(0);
expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseId)))[0])
.toMatchObject({ status: "pending_cleanup", cleanupStatus: "failed", metadata: { pendingCleanupInFlight: false } });
const retryContext = context.wakeup.mock.calls.at(-1)![1].contextSnapshot!;
for (const phase of ["promotion", "execution"] as const) {
await expect(authorizeFailedChatRunRetryWake(db, db, { phase,
wakeupRequestId: receipt.chatActionId!, companyId, issueId,
agentId: context.fixture.assignedAgentId, contextSnapshot: retryContext,
})).rejects.toMatchObject({ details: { code: "chat_failed_run_retry_not_authorized" } });
}
const heartbeat = heartbeatService(db);
await heartbeat.resumeQueuedRuns();
await context.service.processFailedChatRunRetries();
expect((await readAgentWakeupRequestsById(receipt.chatActionId!))[0])
.toMatchObject({ status: "deferred_issue_execution", runId: null });
// Model the recovery worker's validated absence receipt; this test
// exercises queue continuation, not the provider's absence signature.
const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseId));
await db.update(environmentLeases).set({ status: "released", cleanupStatus: "success", releasedAt: new Date(),
failureReason: "workspace_source_unavailable", metadata: { ...lease.metadata,
remoteExecutionTermination: remoteTerminationReceipt(lease, { providerLeaseId: lease.providerLeaseId, state: "destroyed" }),
} }).where(eq(environmentLeases.id, leaseId));
await db.update(issueRecoveryActions).set({ evidence: { ...recoveryAction.evidence,
systemRecovery: { ...recoveryAction.evidence.systemRecovery as Record<string, unknown>, state: "exhausted",
failure: { code: "source_unavailable", retryClass: "service_required", evidenceFingerprint: "missing-source" } },
} }).where(eq(issueRecoveryActions.id, recoveryAction.id));
if (gate === "foreign_source_hold") {
const foreignRunId = randomUUID();
await db.insert(heartbeatRuns).values({ id: foreignRunId, companyId, issueId, nativeIssueId: issueId,
agentId: context.fixture.assignedAgentId, runtimeMode: "native", nativePhase: "terminal_failure",
status: "failed", createdAt: new Date(Date.now() - 120_000), finishedAt: new Date(Date.now() - 60_000),
resultJson: { workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: {
schema: "paperclip.workspace-restore-recovery.v1", runId: foreignRunId, leaseIds: [] } },
});
await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId,
kind: "active_run_watchdog", status: "resolved", ownerType: "system",
cause: "native_session_retry_exhausted", fingerprint: randomUUID(), nextAction: "An independent source still needs recovery.",
evidence: { runId: foreignRunId, automaticRecovery: { replay: "blocked" } },
});
}
if (gate === "paused_after_wait") await db.update(chatEndpoints).set({ status: "paused" })
.where(eq(chatEndpoints.id, context.endpoint.id));
if (gate === "paused_agent_after_wait") await db.update(agents).set({ status: "paused" })
.where(eq(agents.id, context.fixture.assignedAgentId));
if (gate === "edited_after_wait") await db.update(issueComments).set({ body: "Changed after waiting", updatedAt: new Date() })
.where(eq(issueComments.id, String(context.action.payload.commentId)));
if (gate === "superseded_after_wait") {
const replacement = await requestTaskRetry({ db, companyId, issueId, actorId: "owner-user",
actorSource: "cloud_tenant", observedRunId: context.runId, idempotencyKey: randomUUID(),
prepareChatRetry: context.service.prepareFailedChatRunRetry });
expect(replacement.requestId).not.toBe(receipt.requestId);
// Reconsider only the old action, including a delayed worker after
// supersession. The replacement remains owned by its own request.
await context.service.processFailedChatRunRetry(receipt.chatActionId!);
} else {
await heartbeat.resumeQueuedRuns();
await context.service.processFailedChatRunRetries();
}
if (!canResume) {
if (gate === "foreign_source_hold") {
expect((await readAgentWakeupRequestsById(receipt.chatActionId!))[0]).toMatchObject({
status: "deferred_issue_execution", payload: { executionWait: { reason: "workspace_repair_required" } },
});
} else if (gate !== "superseded_after_wait") {
await heartbeat.resumeQueuedRuns();
expect((await db.select().from(chatActions).where(eq(chatActions.id, receipt.chatActionId!)))[0])
.toMatchObject({ status: "failed" });
expect((await readAgentWakeupRequestsById(receipt.requestId!))[0]).toMatchObject({ status: "failed" });
}
expect((await readAgentWakeupRequestsById(receipt.chatActionId!))[0]).toMatchObject({ runId: null });
expect((await readHeartbeatRunsByCompanyId(companyId))
.filter(run => run.contextSnapshot?.taskRetryRequestId === receipt.requestId)).toHaveLength(0);
return;
}
const [admitted] = await readAgentWakeupRequestsById(receipt.chatActionId!);
expect(admitted).toMatchObject({ status: "queued" });
expect(admitted.runId).toBeTruthy();
const [successor] = await readHeartbeatRunsById(admitted.runId!);
expect(successor).toMatchObject({ status: "queued", retryOfRunId: context.runId,
contextSnapshot: { wakeCommentIds: [context.action.payload.commentId], taskRetryRequestId: receipt.requestId } });
expect(await readFreshWorkspaceReadiness(db, successor)).toMatchObject({
mode: "fresh_isolated_workspace", retryIntentId: receipt.requestId, sourceRunId: context.runId,
retainedSourceIds: [leaseId],
});
await heartbeat.resumeQueuedRuns();
await context.service.processFailedChatRunRetries();
expect((await readHeartbeatRunsByCompanyId(companyId))
.filter(run => run.contextSnapshot?.taskRetryRequestId === receipt.requestId)).toHaveLength(1);
} finally {
await context.service.shutdown();
// Remove this case's unadmitted synthetic Retry receipts. Global
// notice scans must not consume another case's injected read failure.
await db.delete(agentWakeupRequests).where(and(
eq(agentWakeupRequests.companyId, context.fixture.companyId),
inArray(agentWakeupRequests.reason, ["task_retry", "retry_failed_run"]),
isNull(agentWakeupRequests.runId),
));
}
},
);
it.each(["stopped", "real_queue", "unknown_executor", "source_edited", "cancelled_stopped", "cancelled_unknown"] as const)(
"admits ordinary chat task Retry with a missing checkpoint only after containment: %s", async kind => {
const context = await failedChatRetryFixture("telegram", undefined, kind === "real_queue");
@@ -1,5 +1,5 @@
import { continuingTeamsFilePublication, type ContinuingTeamsFileTransfer } from "./microsoft-teams/file-transfer-source.js";
import { readTaskChatRetryIntent, taskChatRetryExecutionStopped } from "./task-retry-authority.js";
import { readTaskChatRetryIntent, taskChatRetryCanAwaitRecovery, taskChatRetryExecutionStopped } from "./task-retry-authority.js";
import { notifyChatActionWork, notifyChatPublicationWork } from "../chat-work-notifications.js";
import { FAILED_RUN_RETRY_STALE_MS, dueChatAction } from "../chat-action-work.js";
@@ -549,6 +549,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
source: Pick<FailedChatRetrySource, "failedRunId" | "issueId" | "agentId">,
companyId: string,
taskRetryIntentId?: string,
allowPendingRecovery = false,
) {
const [run] = await tx
.select()
@@ -566,6 +567,9 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
if (!await readTaskChatRetryIntent(tx as unknown as Db, { companyId, issueId: source.issueId,
agentId: source.agentId, sourceRunId: run.id, intentId: taskRetryIntentId })) throw failedChatRetryDenied();
if (await taskChatRetryExecutionStopped(tx as unknown as Db, run)) return;
if (allowPendingRecovery && await taskChatRetryCanAwaitRecovery(
tx as unknown as Db, run, source.issueId, taskRetryIntentId,
)) return;
// A still-live/unknown executor remains held. The exact pre-provider
// proof below can establish a never-started native attempt.
if (run.runtimeMode !== "native") throw failedChatRetryDenied();
@@ -1437,7 +1441,8 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
retryActionId: input.wakeupRequestId,
});
await assertFailedNativeRetryState(tx, source, input.companyId,
typeof source.intent.payload.taskRetryIntentId === "string" ? source.intent.payload.taskRetryIntentId : undefined);
typeof source.intent.payload.taskRetryIntentId === "string" ? source.intent.payload.taskRetryIntentId : undefined,
input.phase === "admission");
const context = input.contextSnapshot;
const hint = context.chatFailedRunRetry as
| Record<string, unknown>
@@ -1587,10 +1592,13 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
authorizeFailedChatRunRetry(tx, scope),
);
const existing = await db
.select({ id: agentWakeupRequests.id })
.select()
.from(agentWakeupRequests)
.where(eq(agentWakeupRequests.id, claimed.id));
if (!existing.length) {
const resumeTaskRetry = typeof claimed.payload.taskRetryIntentId === "string" &&
existing[0]?.status === "deferred_issue_execution" && !existing[0].runId &&
existing[0].payload?.taskRetryRequestId === claimed.payload.taskRetryIntentId;
if (!existing.length || resumeTaskRetry) {
const context = {
issueId: source.issueId,
...(typeof claimed.payload.taskRetryIntentId === "string" ? {
@@ -1641,6 +1649,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
current,
claimed.companyId,
typeof claimed.payload.taskRetryIntentId === "string" ? claimed.payload.taskRetryIntentId : undefined,
true, // The durable wake callback validates admission; later phases recheck containment.
);
},
}),
@@ -1662,11 +1671,6 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
.select()
.from(agentWakeupRequests)
.where(eq(agentWakeupRequests.id, claimed.id));
if (receipt)
return settle("processed", {
code: "failed_chat_retry_durable",
attemptCount,
});
const denied =
error &&
typeof error === "object" &&
@@ -1678,6 +1682,24 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
"chat_failed_run_retry_not_authorized",
"chat_action_authorization_changed",
].includes(String(error.details.code));
if (receipt) {
if (denied && typeof claimed.payload.taskRetryIntentId === "string" &&
receipt.status === "deferred_issue_execution" && !receipt.runId &&
receipt.payload?.taskRetryRequestId === claimed.payload.taskRetryIntentId) {
// Receipt existence proves the earlier wait, not current authority.
// Retire only this unadmitted wake; an admitted run keeps its history.
await db.update(agentWakeupRequests).set({
status: "failed", error: "chat_failed_run_retry_not_authorized",
finishedAt: new Date(), updatedAt: new Date(),
}).where(and(eq(agentWakeupRequests.id, receipt.id),
eq(agentWakeupRequests.companyId, claimed.companyId),
eq(agentWakeupRequests.status, "deferred_issue_execution"), isNull(agentWakeupRequests.runId)));
return settle("failed", { code: "failed_chat_retry_not_authorized", attemptCount });
}
return settle(["skipped", "failed", "cancelled"].includes(receipt.status) ? "failed" : "processed", {
code: "failed_chat_retry_durable", attemptCount,
});
}
const retryable = !denied && attemptCount < 8;
return settle(retryable ? "issued" : "failed", {
code: retryable
@@ -1,9 +1,11 @@
import { and, eq, sql } from "drizzle-orm";
import { agentWakeupRequests, environmentLeases, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db";
import { agentWakeupRequests, environmentLeases, heartbeatRuns, issueRecoveryActions, nativeRunFinalizations, type Db } from "@paperclipai/db";
import { readTaskRetryIntent } from "../task-retry.js";
import { hasNativeLocalProcessStop, hasHistoricalSuspendedNativeSession, recordedLocalExecutionStopped } from "../native-local-process-stop.js";
import { hasRemoteTerminationReceipt } from "../remote-execution-termination.js";
import { adapterExecutionControls } from "../adapter-execution-control.js";
import { getExecutionBlocker } from "../execution-blocker.js";
import { readSystemRecovery } from "../system-recovery-state.js";
/** This receipt changes the workspace strategy, never the external principal,
* accepted inbound batch, publication scope, or current conversation access. */
@@ -43,3 +45,25 @@ export async function taskChatRetryExecutionStopped(db: Db, run: typeof heartbea
await hasNativeLocalProcessStop(db, run.companyId, run.id) ||
await hasHistoricalSuspendedNativeSession(db, run);
}
/** Admission may persist a wait for the exact recovery this user requested.
* This grants no execution authority: promotion and execution still require
* containment, and the queue retains the source's current execution blocker. */
export async function taskChatRetryCanAwaitRecovery(db: Db, run: typeof heartbeatRuns.$inferSelect,
issueId: string, intentId: string) {
const retry = await readTaskChatRetryIntent(db, { companyId: run.companyId, issueId,
agentId: run.agentId, sourceRunId: run.id, intentId });
if (!retry || retry.wake.status !== "deferred_issue_execution" || retry.wake.runId) return false;
const blocker = await getExecutionBlocker(db, run.companyId, issueId);
if (blocker?.runId !== run.id || !blocker.recoveryActionId) return false;
const [action] = await db.select({ evidence: issueRecoveryActions.evidence }).from(issueRecoveryActions).where(and(
eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.sourceIssueId, issueId),
eq(issueRecoveryActions.id, blocker.recoveryActionId),
)).for("share", { noWait: true });
const recovery = readSystemRecovery(action?.evidence);
// Attaching Retry to active automatic recovery preserves that recovery's
// generation and budget. Its current intentId binds this user request; the
// Retry receipt's generation is a different identity.
return Boolean(recovery && recovery.sourceRunId === run.id && recovery.intentId === intentId &&
["scheduled", "running", "awaiting_evidence"].includes(recovery.state));
}
@@ -194,8 +194,19 @@ export async function admitExplicitNativeContinuation(input: {
const unusedAdmission = run.status === "cancelled" && !run.startedAt &&
run.errorCode === "execution_reconciliation_required" &&
!run.processPid && !run.processGroupId && !run.nativeSessionId;
// Several retained holds may describe this same failed execution. Reuse
// its current explicit Retry strategy for a historical hold that predates
// system recovery. Every hold still passes the stop/effect checks below.
const sourceRunId = run.id;
const strategyEvidence = input.taskRetryRequestId && !Object.hasOwn(action.evidence, "systemRecovery")
? actions.find(candidate => {
const recovery = readSystemRecovery(candidate.evidence);
return candidate.evidence.runId === sourceRunId && recovery?.sourceRunId === sourceRunId &&
recovery.intentId === input.taskRetryRequestId;
})?.evidence ?? action.evidence
: action.evidence;
const freshIntent = await canSelectFreshWorkspace({ db, companyId, issueId, agentId, actorId,
sourceRunId: run.id, evidence: action.evidence, taskRetryRequestId: input.taskRetryRequestId,
sourceRunId: run.id, evidence: strategyEvidence, taskRetryRequestId: input.taskRetryRequestId,
queuedRequestId: input.queuedCommentRequestId });
const legacyUserTurn = run.runtimeMode === "legacy" &&
action.cause === "legacy_execution_requires_reconciliation" &&
@@ -56,6 +56,10 @@ export async function readFreshWorkspaceReadiness(db: Db, run: Pick<typeof heart
eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.sourceIssueId, issueId),
sql`${issueRecoveryActions.evidence}#>>'{explicitUserContinuation,runId}' = ${run.id}`,
sql`${issueRecoveryActions.evidence}#>>'{explicitUserContinuation,readiness,mode}' = 'fresh_isolated_workspace'`,
// Historical sibling holds also record the continuation, but only the
// recovery generation owns its committed fresh attempt. Select that proof.
sql`${issueRecoveryActions.evidence}#>>'{systemRecovery,freshAttemptRunId}' = ${run.id}`,
sql`${issueRecoveryActions.evidence}#>>'{systemRecovery,sourceRunId}' = ${issueRecoveryActions.evidence}#>>'{explicitUserContinuation,readiness,sourceRunId}'`,
sql`${issueRecoveryActions.evidence}->>'runId' = ${issueRecoveryActions.evidence}#>>'{explicitUserContinuation,readiness,sourceRunId}'`,
)).limit(1);
if (!action) {
@@ -113,8 +117,9 @@ export async function readFreshWorkspaceReadiness(db: Db, run: Pick<typeof heart
const parsed = freshReadinessSchema.safeParse(authorization?.readiness);
if (!parsed.success) return null;
const readiness = parsed.data;
if (action?.evidence.runId !== readiness.sourceRunId ||
readSystemRecovery(action.evidence)?.freshAttemptRunId !== run.id) return null;
const recovery = readSystemRecovery(action.evidence);
if (action.evidence.runId !== readiness.sourceRunId || recovery?.sourceRunId !== readiness.sourceRunId ||
recovery.freshAttemptRunId !== run.id) return null;
const [wake] = await db.select({ id: agentWakeupRequests.id }).from(agentWakeupRequests).where(and(
eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.id, readiness.retryIntentId),
eq(agentWakeupRequests.agentId, run.agentId), eq(agentWakeupRequests.runId, run.id),
+43 -20
View File
@@ -2359,6 +2359,15 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
receipt.payload?.museBindingGeneration !== museRequest.bindingGeneration || receipt.payload?.museRequestDigest !== museRequest.requestDigest)) {
throw conflict("Request ID reused with changed input.");
}
if (receipt && durableRequest?.failedRunRetry && opts.taskRetryRequestId &&
receipt.status === "deferred_issue_execution" && !receipt.runId &&
receipt.payload?.taskRetryRequestId === opts.taskRetryRequestId) {
// A bound ordinary Retry may reconsider its same deferred chat receipt.
// The issue lock, live chat authority and execution holds below still
// decide whether it can dispatch; this never creates another wake.
executionWaitRequestId = receipt.id;
return null;
}
return receipt;
};
const priorReceipt = await existingDurableReceipt(db);
@@ -2798,7 +2807,9 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
// A user message can join a queue originally created by a
// system wake. Admission validates the saved user comment or board click.
(opts.queuedCommentInterruptId ?? opts.queuedCommentRequestId) === executionWaitRequestId
? undefined : eq(agentWakeupRequests.requestedByActorType, "user"),
? undefined : eq(agentWakeupRequests.requestedByActorType,
durableRequest?.failedRunRetry && opts.taskRetryRequestId
? opts.requestedByActorType ?? "" : "user"),
opts.queuedCommentInterruptId === executionWaitRequestId
? sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt'->>'actorId' = ${opts.requestedByActorId ?? ""}`
: opts.queuedCommentRequestId === executionWaitRequestId ? undefined
@@ -2812,7 +2823,12 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
agent.companyId, issueId, pending.payload)))) {
return { kind: "deferred" as const };
}
if (opts.taskRetryRequestId) {
if (opts.taskRetryRequestId && durableRequest?.failedRunRetry) {
if (pending.id !== durableRequest.id || pending.payload?.taskRetryRequestId !== opts.taskRetryRequestId)
return { kind: "deferred" as const };
// The parent board Retry is checked separately below after the
// connected conversation reauthorizes this exact child receipt.
} else if (opts.taskRetryRequestId) {
const retry = readTaskRetryIntent(pending.payload);
if (pending.id !== opts.taskRetryRequestId || retry?.generationId !== pending.id || retry.chatActionId ||
retry.actorId !== opts.requestedByActorId || retry.sourceRunId !== enrichedContextSnapshot.previousRunId)
@@ -4148,23 +4164,30 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
enrichedContextSnapshot.explicitUserContinuation = explicitContinuation;
}
const wakeupRequest = await tx
.insert(agentWakeupRequests)
.values({
...durableReceiptFields,
companyId: agent.companyId,
agentId,
source,
triggerDetail,
reason,
payload,
status: "queued",
requestedByActorType: opts.requestedByActorType ?? null,
requestedByActorId: opts.requestedByActorId ?? null,
idempotencyKey: opts.idempotencyKey ?? null,
})
.returning()
.then((rows) => rows[0]);
const wakeupValues = {
...durableReceiptFields,
companyId: agent.companyId,
agentId,
source,
triggerDetail,
reason,
payload,
status: "queued" as const,
requestedByActorType: opts.requestedByActorType ?? null,
requestedByActorId: opts.requestedByActorId ?? null,
idempotencyKey: opts.idempotencyKey ?? null,
};
// A deferred durable chat wake already owns this ID and source.
// Promote that receipt in the same transaction as its successor.
const wakeupRequest = durableRequest?.failedRunRetry && executionWaitRequestId === durableRequest.id
? await tx.update(agentWakeupRequests).set({
payload, status: "queued", error: null, finishedAt: null, updatedAt: new Date(),
}).where(and(eq(agentWakeupRequests.id, durableRequest.id),
eq(agentWakeupRequests.companyId, agent.companyId),
eq(agentWakeupRequests.status, "deferred_issue_execution"), isNull(agentWakeupRequests.runId)))
.returning().then(rows => rows[0])
: await tx.insert(agentWakeupRequests).values(wakeupValues).returning().then(rows => rows[0]);
if (!wakeupRequest) return { kind: "deferred" as const };
// A handoff changes the executor, not the owner of saved user input.
// Validate its exact stopped source while the issue row is locked;
@@ -4228,7 +4251,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
...queuedCommentIdsFromRunContext(enrichedContextSnapshot),
]),
];
if (opts.queuedCommentRequestId || opts.taskRetryRequestId || handoffSource) {
if (opts.queuedCommentRequestId || (opts.taskRetryRequestId && !failedChatRetry) || handoffSource) {
adoptedCommentIds = await undeliveredLegacyUserCommentIds(tx as unknown as Db,
agent.companyId, issueId, agentId, adoptedCommentIds);
}
@@ -1,4 +1,5 @@
import { execFile as execFileCallback } from "node:child_process";
import { randomUUID } from "node:crypto";
import { mkdtemp, mkdir, readFile, readdir, rm, writeFile, access, rename } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
@@ -6,7 +7,7 @@ import { pathToFileURL } from "node:url";
import { promisify } from "node:util";
import { captureDirectorySnapshot, directorySnapshotSha256, serializeDirectorySnapshot } from "@paperclipai/adapter-utils/workspace-restore-merge";
import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/execution-target";
import { agents, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, environmentLeases, executionWorkspaces, executionWorkspaceRepositoryAccess, heartbeatRuns, issues, issueAccessGrants, plugins, projects, projectWorkspaces, type Db } from "@paperclipai/db";
import { agentWakeupRequests, issueRecoveryActions, agents, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, environmentLeases, executionWorkspaces, executionWorkspaceRepositoryAccess, heartbeatRuns, issues, issueAccessGrants, plugins, projects, projectWorkspaces, type Db } from "@paperclipai/db";
import { eq, sql } from "drizzle-orm";
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
@@ -21,6 +22,8 @@ import { prepareHeartbeatWorkspace, type HeartbeatWorkspacePreparationInput } fr
import { buildNativeExecutionInput } from "../native-runtime/native-execution-input.js";
import { nativeRuntimeContextFixture } from "../native-runtime/runtime-context.test-fixture.js";
import * as managedRepositoryCheckout from "../managed-repository-checkout.js";
import { readFreshWorkspaceReadiness } from "../fresh-workspace-continuation.js";
import { createSystemRecovery } from "../system-recovery-state.js";
import { isolatedRemoteSourcePath } from "../isolated-remote-workspace-source.js";
import { cleanupExecutionWorkspaceArtifacts } from "../workspace-runtime.js";
import { buildProjectMentionHref } from "@paperclipai/shared";
@@ -262,6 +265,45 @@ describe.skipIf(!support.supported)("heartbeat workspace preparation boundary",
expect((await readIssue(issue.id)).executionWorkspaceId).toBe(replacement.persistedExecutionWorkspace!.id);
}, 30_000);
it.each(["historical_first", "recovery_first"] as const)(
"prepares the admitted fresh directory with sibling holds in either order: %s", async order => {
const { input, source, issue } = await fixture(true);
const original = await prepareHeartbeatWorkspace(db, input);
await writeFile(join(original.executionWorkspace.cwd, "unpublished.txt"), "keep the old work\n");
const sourceRunId = input.run.id, retryIntentId = randomUUID();
const [nextRun] = await db.insert(heartbeatRuns).values({ companyId: input.run.companyId,
agentId: input.agent.id, issueId: issue.id, scopeKind: "issue", status: "running", runtimeMode: "legacy",
contextSnapshot: { issueId: issue.id, taskRetryRequestId: retryIntentId } }).returning();
await db.insert(agentWakeupRequests).values({ id: retryIntentId, companyId: input.run.companyId,
agentId: input.agent.id, reason: "task_retry", source: "on_demand", status: "coalesced", runId: nextRun.id,
payload: { issueId: issue.id } });
const readiness = { mode: "fresh_isolated_workspace" as const, retryIntentId, sourceRunId, retainedSourceIds: [] };
const evidence = { runId: sourceRunId, explicitUserContinuation: { runId: nextRun.id, readiness } };
const hold = { companyId: input.run.companyId, sourceIssueId: issue.id, kind: "active_run_watchdog",
status: "resolved", ownerType: "system", nextAction: "Preserve the original source outcome." };
const historical = { ...hold, cause: "native_session_retry_exhausted", fingerprint: randomUUID(), evidence };
const recovery = { ...hold, cause: "native_continuation_requires_reconciliation", fingerprint: randomUUID(),
evidence: { ...evidence, systemRecovery: { ...createSystemRecovery(sourceRunId, retryIntentId),
state: "exhausted", freshAttemptRunId: nextRun.id } } };
await db.insert(issueRecoveryActions).values(order === "historical_first" ? [historical, recovery] : [recovery, historical]);
// Use the same committed-receipt lookup as run preparation, not a
// hand-built freshWorkspace input. The historical hold has no generation.
input.task.freshWorkspace = await readFreshWorkspaceReadiness(db, nextRun);
expect(input.task.freshWorkspace).toEqual(readiness);
input.run = nextRun;
input.task.issueRef = { ...issue, executionWorkspaceId: null, executionWorkspacePreference: "create_new" };
const replacement = await prepareHeartbeatWorkspace(db, input);
expect(replacement.executionWorkspace.cwd).not.toBe(original.executionWorkspace.cwd);
expect(replacement.executionWorkspace.cwd).not.toBe(source);
expect(await readFile(join(replacement.executionWorkspace.cwd, "README.md"), "utf8")).toBe("source content\n");
await expect(access(join(replacement.executionWorkspace.cwd, "unpublished.txt"))).rejects.toThrow();
expect(await readFile(join(original.executionWorkspace.cwd, "unpublished.txt"), "utf8")).toBe("keep the old work\n");
const common = await execFile("git", ["-C", replacement.executionWorkspace.cwd, "rev-parse", "--path-format=absolute", "--git-common-dir"]);
expect(common.stdout.trim()).toBe(join(replacement.executionWorkspace.cwd, ".git"));
expect((await readIssue(issue.id)).executionWorkspaceId).toBe(replacement.persistedExecutionWorkspace!.id);
}, 30_000,
);
it("starts a fresh clean task directory without treating it as a Git source", async () => {
const { input, source, issue } = await fixture();
input.policy.useIsolatedTaskDirectory = true;
+14
View File
@@ -7,6 +7,7 @@ import { persistActivity } from "./activity-log.js";
import { createSystemRecovery, readSystemRecovery } from "./system-recovery-state.js";
import { updateSystemRecovery } from "./system-recovery.js";
import { instanceSettingsService } from "./instance-settings.js";
import { notifyChatActionWork } from "./chat-work-notifications.js";
export async function assertTaskRetryConversationAllowed(
db: Db,
@@ -182,6 +183,19 @@ export async function reconcileTaskChatRetryReceipts(db: Db) {
const [receipt] = chat ? await db.select().from(agentWakeupRequests).where(and(
eq(agentWakeupRequests.companyId, wake.companyId), eq(agentWakeupRequests.id, chat.id),
eq(agentWakeupRequests.agentId, wake.agentId))) : [];
// A deferred chat receipt is admitted input, not a completed dispatch.
// Re-stage that exact action through chat's owner; it rechecks authority
// and normal queue policy before updating the existing receipt.
if (chat?.status === "processed" && chat.payload.taskRetryIntentId === wake.id &&
chat.payload.failedRunId === intent.sourceRunId && receipt?.status === "deferred_issue_execution" &&
!receipt.runId && receipt.payload?.taskRetryRequestId === wake.id) {
await db.transaction(async tx => {
await notifyChatActionWork(tx, "failed_run_retry");
await tx.update(chatActions).set({ status: "issued", updatedAt: new Date() })
.where(and(eq(chatActions.id, chat.id), eq(chatActions.companyId, wake.companyId),
eq(chatActions.status, "processed"), eq(chatActions.updatedAt, chat.updatedAt)));
});
}
const failed = chat && ["failed", "cancelled"].includes(chat.status);
await db.update(agentWakeupRequests).set({
...(receipt?.runId ? { status: "coalesced", runId: receipt.runId, finishedAt: new Date() }