mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-08 22:13:59 +01:00
fix(recovery): preserve Retry authority and reopen cross-task cases (#18997)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Tasks keep execution identity and saved workspace files separate from control-plane updates. > - A scanner can have permission to reopen a case while its run cannot restore that case's files. > - Retry can also follow a preparation failure that has no recovery-action record. > - Both paths can accept valid work but leave the target unable to run. > - This change restores files under the target executor and saves admitted Retry authority on the canonical receipt. > - Existing workspace, user, source and successor checks still apply. ## Linked Issues or Issue Description **What happened?** An authorized agent updates another closed task. The route tries to restore the target workspace under the caller's run identity and returns `403 Task workspace run identity is unavailable`. A separate failure occurs when a normal board Retry follows a pre-provider preparation failure. Admission has no recovery-action row on which to save continuation authority. Dispatch then fails with `continuation_user_authorization_missing`. **Expected behavior** Accept the authorized task update. Restore its terminal workspace when the target's admitted executor starts. An admitted Retry must retain its exact dispatch authority even when the predecessor has no recovery-action record. **Steps to reproduce** 1. Complete a task with an isolated workspace. From a different running task, use an authorized comment or PATCH with `resume: true` on that closed task. 2. Separately, fail native preparation before provider work starts, with no recovery-action row. Submit board Retry, admit its successor, and build the execution continuation. 3. Before this change, the first path returns 403 and the second rejects its admitted authority. **Paperclip version or commit** Reproduced against `e968131efa7e63d321f3bcca8d47a62513c2efb4`. Rebased onto `afbc5a885b5`. **Deployment mode** Server-managed native task execution. Tests use real PostgreSQL and temporary workspace files. Related PRs: #16927 covers terminal workspace races. #17916 retains exact native export sources. Neither covers these route and Retry receipt failures. A GitHub search found no duplicate. ROADMAP.md was checked; this is a recovery bug fix. ## What Changed - Defer workspace restoration only for authorized cross-task comments and updates from a live run. Checkout keeps its existing workspace identity checks. - Restore the target's selected terminal workspace under its own admitted run. Preserve its saved files and existing source checks. Leave manually archived workspaces unchanged. - Store server-created continuation authority on the canonical admitted Retry receipt when no source recovery-action record exists. Existing action records remain authoritative. Read the fallback only after matching the company, task, agent, user, source, generation and successor. - Add route and PostgreSQL regression tests. Document both paths. ## Verification - Red: both cross-task route tests returned the observed 403 before the route fix. - Red: the target executor could not reuse its archived terminal workspace before the restoration fix. - Red: an admitted preparation Retry with no recovery-action row failed with `continuation_user_authorization_missing` before the receipt fix. - Green: 419 recovery and continuation tests pass, plus 183 focused route, environment and Retry tests. The assembled heartbeat test starts the agent in the same terminal worktree and verifies both its unpushed commit and uncommitted file bytes. It fails before the restoration fix. - Negative cases reject wrong Retry runs, users, source runs, generations, unadmitted receipts and missing authority. The original failed run stays unchanged. Workspace tests verify saved file bytes. - Full `pnpm build`, `pnpm -r typecheck`, `pnpm check:module-boundaries` and `git diff --check` pass. - The full suite found an existing assembled chat authority regression: clearing a recovery-action record could be bypassed by the new fallback. The fix now writes the fallback only when that source has no action record. That assembled regression passes with the original test unchanged. - Full `pnpm test:run` is running again. Its first attempt also hit an unrelated ancestor skill-root lookup for a missing Slack skill. The rerun uses a temporary untracked link to the repository's own bundled skills. This link is excluded from the PR. Full-suite completion and hosted CI are pending. - Hosted CI has multiple runner interruptions. E2e shard 1/8, server shards 3/12, 8/12 and 12/12, and workspaces-a shard 1/2 show runner shutdown signals. E2e shard 5/8 and server shards 1/12 and 2/12 have GitHub annotations reporting lost runner communication; their logs return 404. The `ci / verify` assertion failed because the general-test matrix was cancelled. GitHub rejected job and failed-workflow reruns while this workflow is still running. Rerun the interrupted checks after it settles. - Greptile reviewed exact head `a1e3929aedb1ee3bfcd001f4ebb1510106bf4070` at 2026-11-04 17:37 UTC: 5/5, with no new blocking findings and no unresolved review threads. ## Risks - A deferred workspace restore can still fail at target admission. The run keeps the existing workspace failure path; this change does not bypass file recovery or provider stop checks. - Retry receipts now store server-created authority. Existing action receipts remain valid. No schema migration is required. - Local regression tests establish the software behavior. The new revision has not been deployed to a hosted instance. ## Model Used OpenAI `gpt-6.1-sol` through the Codex harness in T3 Code, with xhigh reasoning, repository inspection, tool use and code execution. The context-window size is not exposed by the runtime. ## 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 - [ ] 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 - [ ] 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:
Devin FoleyandPaperclip authored and GitHub committed
1 parent
039e0dad2c
commit
edb9d29942
9 files changed
+192
-21
No files matched your search
@@ -276,6 +276,14 @@ still checks the exact current binding and snapshot after provider settlement.
|
||||
|
||||
The work is complete and terminal.
|
||||
|
||||
An authorized agent working on another task may add evidence or request a
|
||||
follow-up on a completed task. That control-plane request does not restore the
|
||||
completed task's closed workspace under the caller's run identity. When the
|
||||
follow-up is admitted, its own executor restores the task's selected workspace
|
||||
if task completion closed it. Workspace execution, company, project, and source
|
||||
access checks still apply under that executor. An archive for another reason
|
||||
does not become eligible for restoration through this path.
|
||||
|
||||
### `cancelled`
|
||||
|
||||
The work will not continue and is terminal.
|
||||
@@ -2212,6 +2220,11 @@ unknown, and they do not gain automatic replay eligibility.
|
||||
Admission validates the persisted comment's author, task, and time against every
|
||||
held predecessor. Retry validates the selected failed run's company, task, and
|
||||
agent and preserves that run's identity through admission and history loading.
|
||||
When preparation failed before a source recovery-action row was created, the
|
||||
admitted Retry receipt saves its continuation authorization. Existing action
|
||||
records remain the authority when present. Dispatch checks the exact receipt,
|
||||
successor, source run, and user. A caller-provided continuation object does not
|
||||
establish authority.
|
||||
Duplicate Retry requests adopt the same successor. Only a validated, unconsumed human queue receipt can authorize saved-message continuation after a native failure; arbitrary historical input, an agent-authored comment, or a generic
|
||||
system wake cannot release a hold. The source task keeps its assignee. Process
|
||||
ownership, active controllers, cleanup leases, pause, approval, budget, and normal
|
||||
|
||||
@@ -68,7 +68,7 @@ describe.skipIf(!support.supported)("heartbeat resume of an idle-expired workspa
|
||||
await fs.rm(root, { recursive: true, force: true });
|
||||
}, 60_000);
|
||||
|
||||
async function seed(options: { closedParent?: boolean } = {}) {
|
||||
async function seed(options: { closedParent?: boolean; terminalArchive?: boolean } = {}) {
|
||||
const home = path.join(root, randomUUID());
|
||||
vi.stubEnv("PAPERCLIP_HOME", home);
|
||||
vi.stubEnv("PAPERCLIP_INSTANCE_ID", "default");
|
||||
@@ -120,7 +120,11 @@ describe.skipIf(!support.supported)("heartbeat resume of an idle-expired workspa
|
||||
const unpushed = git(realized.cwd, "rev-parse", "HEAD");
|
||||
await fs.writeFile(path.join(realized.cwd, "notes.md"), "uncommitted notes\n");
|
||||
const workspace = () => db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspaceId)).then(rows => rows[0]!);
|
||||
const expiration = await expireReconstructableGitWorkspace(db, await workspace(), {
|
||||
const expiration = options.terminalArchive ? await (async () => {
|
||||
await db.update(executionWorkspaces).set({ status: "archived", cleanupReason: "issue_terminal",
|
||||
closedAt: new Date("2020-01-01T00:00:00Z") }).where(eq(executionWorkspaces.id, workspaceId));
|
||||
return null;
|
||||
})() : await expireReconstructableGitWorkspace(db, await workspace(), {
|
||||
dryRun: false, expireBase: false, now: new Date(), idleCutoff: options.closedParent ? undefined : new Date(Date.now() - 7 * DAY_MS),
|
||||
restorePoint: { storage: getStorageService(), notBefore: new Date(Date.now() - DAY_MS), keyDurable: true },
|
||||
});
|
||||
@@ -167,6 +171,28 @@ describe.skipIf(!support.supported)("heartbeat resume of an idle-expired workspa
|
||||
expect(issue.executionWorkspaceId).toBe(seeded.workspaceId);
|
||||
});
|
||||
|
||||
it("starts a reopened task in its terminal workspace with the saved files intact", async () => {
|
||||
const seeded = await seed({ terminalArchive: true });
|
||||
const archived = await seeded.workspace();
|
||||
expect(archived).toMatchObject({ status: "archived", cleanupReason: "issue_terminal" });
|
||||
expect(archived.metadata?.gitCacheExpiration).toBeUndefined();
|
||||
await seeded.queueRun();
|
||||
const heartbeat = heartbeatService(db);
|
||||
await heartbeat.resumeQueuedRuns();
|
||||
const run = await waitForRun(heartbeat, seeded.runId);
|
||||
await heartbeat.waitForRunExecutionDrain(seeded.runId);
|
||||
|
||||
expect(run).toMatchObject({ status: "succeeded" });
|
||||
expect(adapterExecute).toHaveBeenCalledTimes(1);
|
||||
expect(adapterExecute.mock.calls[0]?.[0]).toMatchObject({ context: { executionWorkspaceId: seeded.workspaceId,
|
||||
paperclipWorkspace: { cwd: seeded.cwd } } });
|
||||
expect(git(seeded.cwd, "rev-parse", "HEAD")).toBe(seeded.unpushed);
|
||||
expect(await fs.readFile(path.join(seeded.cwd, "notes.md"), "utf8")).toBe("uncommitted notes\n");
|
||||
expect(await seeded.workspace()).toMatchObject({ status: "active", closedAt: null });
|
||||
const [issue] = await db.select().from(issues).where(eq(issues.id, seeded.issueId));
|
||||
expect(issue.executionWorkspaceId).toBe(seeded.workspaceId);
|
||||
});
|
||||
|
||||
it("restores a closed parent's expired workspace when an open child binds to it", async () => {
|
||||
const seeded = await seed({ closedParent: true });
|
||||
expect(seeded.expiration).toMatchObject({ expired: true, restorePoint: true });
|
||||
|
||||
@@ -42,6 +42,16 @@ const mockHeartbeatService = vi.hoisted(() => ({
|
||||
cancelRun: vi.fn(async (): Promise<unknown> => null),
|
||||
}));
|
||||
|
||||
const mockExecutionWorkspaceService = vi.hoisted(() => ({
|
||||
getStoredById: vi.fn(async (): Promise<unknown> => null),
|
||||
reopenClosedIsolatedExecutionWorkspaceForIssue: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock("../services/execution-workspaces.js", async importOriginal => ({
|
||||
...await importOriginal<typeof import("../services/execution-workspaces.js")>(),
|
||||
executionWorkspaceService: () => mockExecutionWorkspaceService,
|
||||
}));
|
||||
|
||||
const mockAgentService = vi.hoisted(() => ({
|
||||
getById: vi.fn(),
|
||||
list: vi.fn(),
|
||||
@@ -197,7 +207,7 @@ vi.mock("../services/index.js", () => ({
|
||||
remapOpenThreadsForDocument: async () => [],
|
||||
}),
|
||||
documentService: () => ({}),
|
||||
executionWorkspaceService: () => ({}),
|
||||
executionWorkspaceService: () => mockExecutionWorkspaceService,
|
||||
feedbackService: () => mockFeedbackService,
|
||||
goalService: () => ({}),
|
||||
heartbeatService: () => mockHeartbeatService,
|
||||
@@ -405,6 +415,8 @@ describe("issue comment reopen routes", () => {
|
||||
mockHeartbeatService.getRun.mockReset();
|
||||
mockHeartbeatService.getActiveRunForAgent.mockReset();
|
||||
mockHeartbeatService.cancelRun.mockReset();
|
||||
mockExecutionWorkspaceService.getStoredById.mockReset().mockResolvedValue(null);
|
||||
mockExecutionWorkspaceService.reopenClosedIsolatedExecutionWorkspaceForIssue.mockReset();
|
||||
mockAgentService.getById.mockReset();
|
||||
mockAgentService.list.mockReset();
|
||||
mockAgentService.resolveByReference.mockReset();
|
||||
@@ -587,6 +599,33 @@ describe("issue comment reopen routes", () => {
|
||||
);
|
||||
});
|
||||
|
||||
it.each(["comment", "update"])("accepts an authorized cross-task %s without restoring the target's files", async operation => {
|
||||
const issue = { ...makeIssue("done"), executionWorkspaceId: "33333333-3333-4333-8333-333333333333" };
|
||||
const actor = { ...agentActor(), source: "agent_jwt", runId: "66666666-6666-4666-8666-666666666666" };
|
||||
mockIssueService.getById.mockResolvedValue(issue);
|
||||
mockIssueService.update.mockResolvedValue({ ...issue, status: "todo" });
|
||||
mockHeartbeatService.getRun.mockResolvedValue({ id: actor.runId, agentId: actor.agentId,
|
||||
companyId: issue.companyId, status: "running", runtimeMode: "native",
|
||||
nativeIssueId: "55555555-5555-4555-8555-555555555555", contextSnapshot: {} });
|
||||
mockExecutionWorkspaceService.getStoredById.mockResolvedValue({ id: issue.executionWorkspaceId,
|
||||
companyId: issue.companyId, sourceIssueId: issue.id, mode: "isolated_workspace", status: "archived",
|
||||
closedAt: new Date("2020-01-01T00:00:00Z"), cleanupReason: "issue_terminal" });
|
||||
// A scanner's run may change this task, but cannot execute in its workspace.
|
||||
mockExecutionWorkspaceService.reopenClosedIsolatedExecutionWorkspaceForIssue.mockRejectedValue(
|
||||
new HttpError(403, "Task workspace run identity is unavailable"));
|
||||
const app = request(await installActor(createApp(), actor));
|
||||
const response = operation === "comment"
|
||||
? await app.post(`/api/issues/${issue.id}/comments`).send({ body: "New regression evidence", resume: true })
|
||||
: await app.patch(`/api/issues/${issue.id}`).send({ comment: "New regression evidence", resume: true });
|
||||
|
||||
expect(response.status, JSON.stringify(response.body)).toBe(operation === "comment" ? 201 : 200);
|
||||
expect(mockIssueService.update.mock.calls[0]?.slice(0, 2))
|
||||
.toEqual([issue.id, expect.objectContaining({ status: "todo" })]);
|
||||
expect(mockExecutionWorkspaceService.reopenClosedIsolatedExecutionWorkspaceForIssue).not.toHaveBeenCalled();
|
||||
await waitForWakeup(() => expect(mockHeartbeatService.wakeup).toHaveBeenCalledWith(issue.assigneeAgentId,
|
||||
expect.objectContaining({ payload: expect.objectContaining({ issueId: issue.id }) })));
|
||||
});
|
||||
|
||||
it("finishes missing reference and activity writes for an accepted comment", async () => {
|
||||
const issue = makeIssue("done");
|
||||
mockIssueService.getById.mockResolvedValue(issue);
|
||||
|
||||
@@ -16,6 +16,7 @@ import { issueSummariesQuerySchema, type IssueSummary } from "@paperclipai/share
|
||||
import { listIssueSummaries } from "../services/issue-summaries.js";
|
||||
import { assertNoCaseHandoffMutation } from "../services/heartbeat/case-handoff-lifecycle.js";
|
||||
import { assertLegacyRunTaskOwnership } from "../legacy-run-task-ownership.js";
|
||||
import { runRetainsIssueIdentity } from "../services/task-repository-acquisition.js";
|
||||
import { withConversationReadState } from "../services/agent-chat-read-state.js";
|
||||
import { markIssueReadSchema } from "@paperclipai/shared";
|
||||
import { enqueueFastResponse } from "../services/fast-responses.js";
|
||||
@@ -7867,6 +7868,8 @@ export function issueRoutes(
|
||||
// this request did not set the flag. The caller continues but
|
||||
// must not install the guard, or it can clear the flag that
|
||||
// the other request still owns.
|
||||
// "deferred" - an authorized cross-task comment/update leaves restoration
|
||||
// to the target's executor. No workspace flag was changed.
|
||||
// null - this function sent an error response, so the caller stops.
|
||||
// it runs only after the route already authorized the request on the issue.
|
||||
async function reopenClosedIssueExecutionWorkspaceOrRespond(
|
||||
@@ -7874,11 +7877,22 @@ export function issueRoutes(
|
||||
res: Response,
|
||||
issue: { id: string; companyId: string; projectId?: string | null },
|
||||
workspace: Pick<ExecutionWorkspace, "id">,
|
||||
options: { deferForOtherTask?: boolean } = {},
|
||||
): Promise<{
|
||||
outcome: "reopened" | "already-open";
|
||||
generation: number;
|
||||
outcome: "reopened" | "already-open" | "deferred";
|
||||
generation: number | null;
|
||||
} | null> {
|
||||
const actor = getActorInfo(req);
|
||||
if (options.deferForOtherTask && req.actor.type === "agent" && req.actor.runId) {
|
||||
const run = await heartbeat.getRun(req.actor.runId);
|
||||
if (run?.status === "running" && run.companyId === issue.companyId && run.agentId === req.actor.agentId
|
||||
&& !runRetainsIssueIdentity(run, issue.id)) {
|
||||
// Task authority permits a scanner's comment or follow-up. Its run
|
||||
// cannot restore another task's files. The target's admitted executor
|
||||
// restores the workspace with its own run identity when work resumes.
|
||||
return { outcome: "deferred", generation: null };
|
||||
}
|
||||
}
|
||||
const result =
|
||||
await executionWorkspacesSvc.reopenClosedIsolatedExecutionWorkspaceForIssue(
|
||||
{
|
||||
@@ -14990,6 +15004,7 @@ export function issueRoutes(
|
||||
res,
|
||||
existing,
|
||||
closedExecutionWorkspace,
|
||||
{ deferForOtherTask: true },
|
||||
);
|
||||
if (reopenOutcome === null) {
|
||||
return;
|
||||
@@ -18997,6 +19012,7 @@ export function issueRoutes(
|
||||
res,
|
||||
issue,
|
||||
closedExecutionWorkspace,
|
||||
{ deferForOtherTask: true },
|
||||
);
|
||||
if (reopenOutcome === null) {
|
||||
return;
|
||||
|
||||
@@ -18,6 +18,7 @@ import { queuedInteractionId } from "./queued-interaction-response.js";
|
||||
import { childReviewOutcomes } from "./native-runtime/child-review-outcomes.js";
|
||||
import { isCompletedOnboardingHandoffWake } from "./chat-completion-delivery.js";
|
||||
import { githubRecoveryReceipts } from "./chat-channels/github/recovery-receipts.js";
|
||||
import { readTaskRetryIntent } from "./task-retry.js";
|
||||
|
||||
export class StaleExecutionContinuationError extends Error {
|
||||
constructor(readonly code: "continuation_task_ownership_changed") {
|
||||
@@ -424,8 +425,22 @@ export async function buildExecutionContinuation(input: {
|
||||
eq(agentWakeupRequests.agentId, input.agentId), eq(agentWakeupRequests.status, "coalesced"),
|
||||
eq(agentWakeupRequests.runId, input.runId), sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
|
||||
)) : [];
|
||||
const taskRetryId = string(explicitContinuation.taskRetryRequestId);
|
||||
const [taskRetryWake] = taskRetryId && z.string().guid().safeParse(taskRetryId).success
|
||||
? await db.select().from(agentWakeupRequests).where(and(
|
||||
eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.id, taskRetryId),
|
||||
eq(agentWakeupRequests.agentId, input.agentId), eq(agentWakeupRequests.reason, "task_retry"),
|
||||
eq(agentWakeupRequests.requestedByActorType, "user"), eq(agentWakeupRequests.status, "coalesced"),
|
||||
sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
|
||||
input.runId ? eq(agentWakeupRequests.runId, input.runId) : undefined,
|
||||
)) : [];
|
||||
const taskRetry = readTaskRetryIntent(taskRetryWake?.payload);
|
||||
const retryAuthorization = taskRetry && taskRetry.generationId === taskRetryWake?.id
|
||||
&& taskRetry.actorId === taskRetryWake.requestedByActorId
|
||||
? object(object(taskRetryWake.payload?.taskRetry).continuation) : {};
|
||||
const continuationAuthorizations = [...reconciliations.map(row => object(row.evidence.explicitUserContinuation)),
|
||||
...(recordedQueue ? [object(object(recordedQueue.payload?.queuedCommentInterrupt).continuation)] : [])]
|
||||
...(recordedQueue ? [object(object(recordedQueue.payload?.queuedCommentInterrupt).continuation)] : []),
|
||||
retryAuthorization]
|
||||
.filter(value => value.previousRunId === explicitUserSource &&
|
||||
(!input.runId || value.runId === input.runId) &&
|
||||
value.commentId === explicitContinuation.commentId &&
|
||||
@@ -443,15 +458,6 @@ export async function buildExecutionContinuation(input: {
|
||||
sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt' is not null`,
|
||||
input.runId ? eq(agentWakeupRequests.runId, input.runId) : undefined,
|
||||
)) : [];
|
||||
const taskRetryId = string(explicitContinuation.taskRetryRequestId);
|
||||
const [taskRetryWake] = taskRetryId && z.string().guid().safeParse(taskRetryId).success
|
||||
? await db.select().from(agentWakeupRequests).where(and(
|
||||
eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.id, taskRetryId),
|
||||
eq(agentWakeupRequests.agentId, input.agentId), eq(agentWakeupRequests.reason, "task_retry"),
|
||||
eq(agentWakeupRequests.requestedByActorType, "user"), eq(agentWakeupRequests.status, "coalesced"),
|
||||
sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
|
||||
input.runId ? eq(agentWakeupRequests.runId, input.runId) : undefined,
|
||||
)) : [];
|
||||
const authorization = continuationAuthorizations.find(value => taskRetryId
|
||||
? value.taskRetryRequestId === taskRetryId && taskRetryWake?.runId === value.runId &&
|
||||
taskRetryWake.requestedByActorId === value.actorId &&
|
||||
|
||||
@@ -540,10 +540,16 @@ export async function admitExplicitNativeContinuation(input: {
|
||||
if (readiness) await persistActivity(db, { companyId, actorType: "system", actorId: "task-recovery",
|
||||
action: "issue.fresh_workspace_selected", entityType: "issue", entityId: issueId,
|
||||
details: { ...readiness, runId: input.successorRunId } });
|
||||
if (input.taskRetryRequestId) await db.update(agentWakeupRequests).set({
|
||||
payload: sql`jsonb_set(${agentWakeupRequests.payload}, '{taskRetry,strategy}',
|
||||
${JSON.stringify(readiness?.mode === "fresh_isolated_workspace" ? "fresh_isolated_workspace" : "continue")}::jsonb)`,
|
||||
}).where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.id, input.taskRetryRequestId)));
|
||||
if (input.taskRetryRequestId) {
|
||||
const payload = sql`jsonb_set(${agentWakeupRequests.payload} #- '{taskRetry,continuation}', '{taskRetry,strategy}',
|
||||
${JSON.stringify(readiness?.mode === "fresh_isolated_workspace" ? "fresh_isolated_workspace" : "continue")}::jsonb)`;
|
||||
// Preserve the action ledger as the authority when it exists. Preparation
|
||||
// can fail before that ledger exists, so only that path needs a fallback.
|
||||
const hasActionAuthorization = actions.some(action => (action.evidence.runId ?? action.evidence.sourceRunId) === previous.id);
|
||||
await db.update(agentWakeupRequests).set({ payload: hasActionAuthorization ? payload
|
||||
: sql`jsonb_set(${payload}, '{taskRetry,continuation}', ${JSON.stringify(authorization)}::jsonb)`,
|
||||
}).where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.id, input.taskRetryRequestId)));
|
||||
}
|
||||
return continuation;
|
||||
}
|
||||
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { mkdtemp, mkdir, rm } from "node:fs/promises";
|
||||
import { mkdtemp, mkdir, rm, writeFile, readFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { agents, companies, createDb, environments, heartbeatRuns, issues, projects, projectWorkspaces, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, type Db } from "@paperclipai/db";
|
||||
@@ -137,6 +137,25 @@ describe.skipIf(!support.supported)("heartbeat environment selection boundary",
|
||||
expect(result.requestedReusableExecutionWorkspaceConfig).toBeNull();
|
||||
});
|
||||
|
||||
it("restores a reopened task's terminal workspace under its own admitted run", async () => {
|
||||
const { input, issue, workspace } = await fixture();
|
||||
const saved = await workspace();
|
||||
await writeFile(join(home, "saved-work.txt"), "Retain the completed investigation\n");
|
||||
await executionWorkspaceService(db).update(saved.id, { status: "archived", strategyType: "project_primary",
|
||||
cleanupReason: "issue_terminal", closedAt: new Date("2020-01-01T00:00:00Z") });
|
||||
await db.update(issues).set({ executionWorkspaceId: saved.id, executionWorkspacePreference: "reuse_existing" })
|
||||
.where(eq(issues.id, issue.id));
|
||||
input.task.issueRef!.executionWorkspaceId = saved.id;
|
||||
input.task.issueRef!.executionWorkspacePreference = "reuse_existing";
|
||||
|
||||
const result = await selectHeartbeatEnvironment(db, input);
|
||||
|
||||
expect(result.reusableExistingExecutionWorkspace).toMatchObject({ id: saved.id, status: "active" });
|
||||
expect(await readFile(join(home, "saved-work.txt"), "utf8")).toBe("Retain the completed investigation\n");
|
||||
expect(await executionWorkspaceService(db).getStoredById(input.agent.companyId, saved.id))
|
||||
.toMatchObject({ status: "active", closedAt: null, sourceIssueId: issue.id });
|
||||
});
|
||||
|
||||
describe("workspaces removed by git cache expiration", () => {
|
||||
async function expired() {
|
||||
const f = await fixture();
|
||||
|
||||
@@ -122,7 +122,9 @@ export async function selectHeartbeatEnvironment(db: Db, input: HeartbeatEnviron
|
||||
&& existingExecutionWorkspace.id === readNonEmptyString(issueRef?.executionWorkspaceId)
|
||||
&& (existingExecutionWorkspace.status === "archived" || existingExecutionWorkspace.status === "cleanup_failed")
|
||||
&& (readGitCacheExpiration(existingExecutionWorkspace as Parameters<typeof readGitCacheExpiration>[0])?.workspaceRemoved
|
||||
|| readTaskDirectoryRetention(existingExecutionWorkspace) !== null)) {
|
||||
|| readTaskDirectoryRetention(existingExecutionWorkspace) !== null
|
||||
|| (existingExecutionWorkspace.sourceIssueId === issueId
|
||||
&& existingExecutionWorkspace.cleanupReason?.startsWith("issue_terminal") === true))) {
|
||||
const [boundIssue] = await db.select({ id: issues.id, companyId: issues.companyId, projectId: issues.projectId })
|
||||
.from(issues).where(and(eq(issues.id, issueId), eq(issues.companyId, agent.companyId)));
|
||||
if (!boundIssue) throw new Error("Execution workspace issue is unavailable");
|
||||
|
||||
@@ -14,6 +14,7 @@ import { instanceSettingsService } from "./instance-settings.js";
|
||||
import { heartbeatService } from "./heartbeat.js";
|
||||
import { taskChatRetryCanAwaitRecovery, taskChatRetryExecutionStopped } from "./chat-channels/task-retry-authority.js";
|
||||
import { admitExplicitNativeContinuation } from "./explicit-native-continuation.js";
|
||||
import { buildExecutionContinuation } from "./execution-continuation.js";
|
||||
|
||||
describe("durable task Retry admission", () => {
|
||||
let postgres: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
@@ -38,6 +39,49 @@ describe("durable task Retry admission", () => {
|
||||
contextSnapshot: { issueId }, finishedAt: new Date() });
|
||||
return { db, companyId, issueId, actorId: "board", observedRunId: runId, idempotencyKey: randomUUID() };
|
||||
}
|
||||
it.each(["valid", "wrong_run", "wrong_actor", "wrong_source", "wrong_generation", "unadmitted", "missing_authorization"])(
|
||||
"dispatches a preparation Retry with no recovery-action row only from its exact admitted receipt: %s", async state => {
|
||||
const input = await seed();
|
||||
await db.update(issues).set({ status: "todo" }).where(eq(issues.id, input.issueId));
|
||||
const [source] = await db.update(heartbeatRuns).set({ runtimeMode: "legacy", errorCode: "setup_failed",
|
||||
executionStage: "preparing", runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } },
|
||||
resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } },
|
||||
}).where(eq(heartbeatRuns.id, input.observedRunId)).returning();
|
||||
const receipt = await requestTaskRetry(input), successorRunId = randomUUID();
|
||||
const continuation = await db.transaction(async tx => {
|
||||
await tx.select().from(issues).where(eq(issues.id, input.issueId)).for("update");
|
||||
const result = await admitExplicitNativeContinuation({ db: tx as unknown as Db,
|
||||
companyId: input.companyId, issueId: input.issueId, agentId: source.agentId,
|
||||
actorId: input.actorId, actorType: "user", reason: "task_retry", commentId: null,
|
||||
successorRunId, taskRetryRequestId: receipt.requestId!,
|
||||
});
|
||||
await tx.insert(heartbeatRuns).values({ id: successorRunId, companyId: input.companyId, agentId: source.agentId,
|
||||
issueId: input.issueId, status: "queued", contextSnapshot: { issueId: input.issueId } });
|
||||
await tx.update(agentWakeupRequests).set({ status: "coalesced", runId: successorRunId })
|
||||
.where(eq(agentWakeupRequests.id, receipt.requestId!));
|
||||
return result;
|
||||
});
|
||||
expect(continuation).toMatchObject({ previousRunId: source.id, taskRetryRequestId: receipt.requestId });
|
||||
expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, input.issueId))).toHaveLength(0);
|
||||
const [admitted] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.requestId!));
|
||||
const taskRetry = admitted.payload!.taskRetry as Record<string, unknown>;
|
||||
if (state !== "valid") await db.update(agentWakeupRequests).set({
|
||||
...(state === "wrong_run" ? { runId: source.id } : {}),
|
||||
...(state === "wrong_actor" ? { requestedByActorId: "other-user" } : {}),
|
||||
...(state === "unadmitted" ? { status: "deferred_issue_execution" } : {}),
|
||||
...(state === "wrong_source" ? { payload: { ...admitted.payload, taskRetry: { ...taskRetry, sourceRunId: randomUUID() } } } : {}),
|
||||
...(state === "wrong_generation" ? { payload: { ...admitted.payload, taskRetry: { ...taskRetry, generationId: randomUUID() } } } : {}),
|
||||
...(state === "missing_authorization" ? { payload: { ...admitted.payload,
|
||||
taskRetry: Object.fromEntries(Object.entries(taskRetry).filter(([key]) => key !== "continuation")) } } : {}),
|
||||
}).where(eq(agentWakeupRequests.id, admitted.id));
|
||||
const result = buildExecutionContinuation({ db, companyId: input.companyId, issueId: input.issueId,
|
||||
agentId: source.agentId, runId: successorRunId,
|
||||
context: { issueId: input.issueId, explicitUserContinuation: continuation }, summary: null, exposeLowTrustRaw: false });
|
||||
if (state === "valid") await expect(result).resolves.toMatchObject({ interruptedRunId: source.id });
|
||||
else await expect(result).rejects.toThrow("continuation_user_authorization_missing");
|
||||
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, source.id)))[0])
|
||||
.toMatchObject({ status: "failed", resultJson: source.resultJson });
|
||||
});
|
||||
it("serializes concurrent clicks and remembers each key after completion", async () => {
|
||||
const input = await seed();
|
||||
const second = { ...input, idempotencyKey: randomUUID() };
|
||||
|
||||
Reference in new issue
Block a user