mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-09 07:10:34 +01:00
fix(runner): retry quarantined tasks in their saved workspace (#19340)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Native sessions retain execution ownership and action history across turns. > - Task Retry must prove that old execution has stopped before it starts another turn. > - An attempt can inherit its predecessor's identity and fail before it launches a provider. > - The cleanup proof searched under that failed attempt and compared unrelated event sequences. > - This pull request verifies the retained owner and starts one fresh authorized turn. > - Saved results and uncertain action outcomes remain intact. ## Linked Issues or Issue Description **What happened?** Task Retry accepted repeated clicks but kept the canonical request deferred with `executionWait.reason: local_cleanup`. It created no successor, although the retained local worker and provider had stopped. **Expected behavior:** After exact cleanup proof, admit one fresh turn. Preserve the old journal and result. Keep unresolved action outcomes unknown. **Steps to reproduce:** 1. Finish a local native turn with retained cleanup state and a saved result. 2. Let the next attempt inherit its identity and fail with `native_session_cleanup_quarantined` before any launch. 3. Upgrade across a tool-contract change, then click task Retry. 4. The old code requires the failed attempt's stop receipt and journal identity. It also requires the current tool fingerprint and equal sequence numbers across raw and normalized events. Those requirements cannot match the retained owner. **Paperclip version or commit:** Reproduced on `31bf6e4482`. **Deployment mode / installation method:** Authenticated, self-hosted source checkout. **Agent adapter(s) involved:** Native `paperclip_runner`, local Codex app-server. The shared physical-stop proof also covers ACPX. **Database mode:** Embedded PostgreSQL 18.1. **Access context:** Board task Retry. **Node.js version / operating system:** Incident: Node 24.20.0, Linux ARM64. Local verification: macOS ARM64. **Relevant logs or output:** Accepted Retry stays in `deferred_issue_execution`; no new run or provider launch occurs. **Relevant config:** Per-turn local native execution. The old tool contract differs from the current contract. **Additional context:** Related: #19311 prevents the oversized tool result that can lead to retained cleanup. #18217 covers broader cleanup isolation and retained-result maintenance. #19134 admits stopped Runner sources under the current run lease. This patch fixes explicit continuation admission on current master. ## What Changed - Resolve the retained journal's owner under the same company, task, agent, workspace, native session, and runner instance. - Require terminal ownership, settled controllers, released local leases, and exact inherited process birth. Refuse any launch evidence on the failed attempt. - Accept a current host's verified process absence from its exact launch receipt when an older server lacks a stop receipt. A later launch invalidates that observation. - Validate normalized receipts against their own IDs, sequences, and digests. Match ordered provider identities to the independent raw journal. - Keep tool compatibility at the session reuse boundary. Explicit continuation always uses a fresh session. - Bind the stop proof to its exact successor and canonical Retry receipt. Reuse the proven local workspace through normal preparation. Refuse changed bindings, closed or remote workspaces, pending operations, and other live leases. Projectless run-ID placeholders keep the existing fresh-workspace fallback. - Retain journals, results, files, and uncertain action outcomes. Add regression and opt-in live acceptance coverage. - Make the split-handoff browser test wait for both child-completion notices and all run settlement before teardown. Assert the final task set stays unchanged. ## Verification - Red: canonical Retry on the original code accepts one saved intent but creates zero successors and records `local_cleanup`. - Red: distinct raw and normalized event sequences reproduce that wait after owner resolution alone. After admission was repaired, live acceptance then exposed the empty-attempt workspace problem. A PostgreSQL regression also reproduced that wrong workspace selection. - Green at `6684df5e82`: **479 targeted cases**, including 26 new PostgreSQL cases for duplicate clicks, exact process and workspace ownership, controller/lease fences, changed state, and independent receipt integrity. The complete 1,150-test native executor suite passed earlier. - `pnpm -r typecheck` and `pnpm build` pass again at final head `432a52bea8`, including its build stamp. Token gates pass. Master now contains the unrelated directory-fixture type correction found during the first typecheck. - Real authenticated HTTP Retry with Codex `gpt-6.1-sol`, high effort, and per-turn execution passes again at final head `432a52bea8` (93.3 seconds). It preserves the first file line, appends the second once, starts a fresh session in the existing workspace, keeps the retained journal unchanged, commits finalization, and clears task locks. - The strengthened split-handoff browser scenario passes **three consecutive local runs**. Its original CI teardown and one same-head rerun failed when cleanup cancelled an active legacy completion-notification run and hit a PostgreSQL row-lock timeout. The scenario now waits for the complete handoff before cleanup. [Initial failure](https://github.com/paperclipai/paperclip/actions/runs/42106466518/job/127523693634), [same-head rerun](https://github.com/paperclipai/paperclip/actions/runs/42106466518/job/127531698321). These failures remain part of the verification record. - The local full `pnpm test:run` passed all six chat integration shards. Its obsolete pre-rebase attempt was stopped during the general server phase; no complete local-suite pass is claimed. Full CI, including both required gates, passed at `6684df5e82`. At final head `432a52bea8`, all initial CI groups passed except the unchanged status-card deadline test, whose post-wakeup clock assertion missed its bound by 10 ms. That exact case passed three unchanged local repetitions (16.8/27.0/18.3 seconds); the single isolated shard rerun passed all **2,218 tests** (123 files), with no source changes. [Initial timing failure](https://github.com/paperclipai/paperclip/actions/runs/42114473652/job/127550722187). - Greptile reviewed `6684df5e82` at **5/5**, with no actionable findings. A second reviewer identified the supported projectless placeholder case. A new real PostgreSQL regression reproduced its workspace-binding failure before the one-line fallback correction; all **45 focused cases** now pass, including default environment selection and changed binding/cwd refusal. Greptile reviewed final head `432a52bea8` at **5/5**, with no blocking findings. The second reviewer’s thread is resolved. - The patched Linux build verified the actual retained worker/provider ownership and process absence in a transaction that was rolled back. No retained state was retired or provider started during that pre-deployment check. ## Risks - This changes an execution admission proof. Incorrect ownership could permit a second executor. The proof retains process, controller, lease, scope, journal, and receipt fences, and duplicate admission is tested through the real queue. - Missing or conflicting evidence remains deferred. Windows, remote execution, automatic replacement, and session reuse retain their existing policies. - No database migration or UI change. The original affected task still needs deployment of the fix before recovery can be claimed. ## Model Used - OpenAI GPT-6 through Codex, with reasoning, repository tools, code execution, SSH diagnostics, and live provider testing. This session does not expose the exact backend model build or context-window size. The live test agent uses OpenAI `gpt-6.1-sol` at high effort. ## 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
1 parent
15249a689d
commit
789a71bc17
12 files changed
+598
-34
No files matched your search
@@ -1221,6 +1221,38 @@ Newer saved input keeps its own receipt while the existing Retry explanation sta
|
||||
visible. Terminal waits direct the user to send new input and hide the Retry action
|
||||
that cannot continue that attempt.
|
||||
|
||||
When a native attempt fails before launch because its predecessor's session is
|
||||
quarantined, explicit task Retry verifies the retained journal's actual owner.
|
||||
That owner must belong to the same company, task, agent, workspace, native session,
|
||||
and runner instance. Its controller must have settled, its local leases must be
|
||||
released, and its recorded worker, provider, and tool processes must be absent.
|
||||
A host-bound launch receipt with the exact process birth can establish the stop
|
||||
when an older server did not write a stop receipt. A later launch invalidates it.
|
||||
The failed attempt must have no launch evidence and must retain that same process
|
||||
identity. Missing or conflicting evidence keeps Retry deferred.
|
||||
|
||||
This proof retires cleanup ownership only. It preserves the old journal, saved
|
||||
result, and uncertain action outcomes. The admitted turn uses a fresh native
|
||||
session under current authority. The old tool-contract fingerprint therefore
|
||||
does not gate physical cleanup; existing compatibility checks still govern
|
||||
session reuse and automatic replacement.
|
||||
Raw runner journal events and normalized run-log events have independent sequence
|
||||
numbers. Cleanup validates each normalized receipt's own sequence, event ID, and
|
||||
digest, then matches the ordered provider lifecycle identities across the two
|
||||
streams. It does not compare their sequence numbers to one another.
|
||||
|
||||
The committed stop proof also names its successor, canonical Retry receipt, and
|
||||
local execution workspace. Workspace preparation uses `resume_stopped_native_workspace`
|
||||
only for that exact admitted successor and unchanged active task binding. It
|
||||
rechecks the directory and refuses pending workspace operations or another live
|
||||
lease. Environment selection pins that workspace again before preparation. This
|
||||
keeps saved files in place while the provider session starts fresh. An unrelated
|
||||
Retry or changed binding cannot borrow the proof or silently replace the files
|
||||
with an empty attempt directory.
|
||||
Projectless runs use their run ID as a transient workspace placeholder. That
|
||||
placeholder cannot authorize managed workspace reuse; those successors keep the
|
||||
existing fresh-workspace preparation path while retaining the old directory.
|
||||
|
||||
A continuation cancelled by the queued-run gate while waiting for review did not start a provider. Its matching dispatch-gate receipt and unclaimed execution fields keep this deliberate wait out of unknown-action reconciliation. Pending review and decision gates still apply. If an older server already created a recovery hold for this exact state, Retry or a new user message can request one fresh turn after the server verifies there is no provider, coordinator, or unfinished cleanup. Saved user input is reconsidered through the same admission path; the cancelled turn is not replayed.
|
||||
|
||||
Other legacy queued-run gate cancellations also stay out of unknown-action reconciliation when the terminal run has no retained execution authority, session, workspace recovery, resources, or execution events. The same proof removes the effect of an old false recovery hold only for its exact company, task, and run; the audit row remains. Started or conflicting work retains its hold. Review waits retain their guarded Retry path. New work still passes the current approval, dependency, budget, pause, and ownership gates.
|
||||
|
||||
@@ -371,6 +371,16 @@ server-authored event among these three types; provider
|
||||
source events cannot supply stop authority. These records stay in the local run
|
||||
log and do not add Telemetry or OpenTelemetry data.
|
||||
|
||||
Explicit continuation writes `native.stopped_conversation_verified` after it
|
||||
verifies the retained worker and provider processes stopped. Its payload binds
|
||||
the requesting `runId`, retained `ownerRunId`, native session, runner instance,
|
||||
process IDs, retained-state fingerprint, successor run, Retry receipt, and local
|
||||
workspace identity. The owner can be a preceding terminal
|
||||
run when the requesting attempt failed before launch with the same process
|
||||
identity. The server checks host-bound process birth and independent provider
|
||||
receipts before retiring cleanup ownership. The event does not settle uncertain
|
||||
actions or authorize replay of the old session.
|
||||
|
||||
## Verified Local Codex Replacement Evidence
|
||||
|
||||
The server writes `native.stopped_text_turn_verified` in the same transaction
|
||||
|
||||
@@ -496,7 +496,8 @@ export async function admitExplicitNativeContinuation(input: {
|
||||
companyId, runId: String(stopped.evidence.runId), agentId,
|
||||
eventType: "native.stopped_conversation_verified", stream: "system", level: "info",
|
||||
message: "The old runner and provider stopped. New user input will use a fresh session; prior action outcomes remain recorded.",
|
||||
payload: stopped.evidence,
|
||||
payload: { ...stopped.evidence, successorRunId: input.successorRunId,
|
||||
...(input.taskRetryRequestId ? { taskRetryRequestId: input.taskRetryRequestId } : {}) },
|
||||
});
|
||||
}
|
||||
const authorization = { ...continuation, actorId, commentId, ...(response ? { interactionId: response.source.interactionId } : {}), ...(retry ? { failedRunId: input.failedRunId } : {}),
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { agentWakeupRequests, environmentLeases, executionWorkspaces, heartbeatRuns, issues, issueRecoveryActions, nativeRunFinalizations, nativeRunResults, workspaceOperations, type Db } from "@paperclipai/db";
|
||||
import { and, eq, inArray, isNull, ne, or, sql } from "drizzle-orm";
|
||||
import { agentWakeupRequests, environmentLeases, executionWorkspaces, heartbeatRunEvents, heartbeatRuns, issues, issueRecoveryActions, nativeRunFinalizations, nativeRunResults, workspaceOperations, type Db } from "@paperclipai/db";
|
||||
import { verifiedWorkspaceRepairReceipt } from "./workspace-recovery-receipt.js";
|
||||
import { persistActivity } from "./activity-log.js";
|
||||
import { readSystemRecovery } from "./system-recovery-state.js";
|
||||
@@ -16,6 +16,7 @@ import { taskRetryRunAncestry } from "./recovery-run-ancestry.js";
|
||||
export type ContinuationReadiness =
|
||||
| { mode: "resume_preserved_workspace"; preservationReceiptId: string }
|
||||
| { mode: "reuse_current_workspace"; retryIntentId: string; sourceRunId: string; workspaceId: string; cwd: string }
|
||||
| { mode: "resume_stopped_native_workspace"; retryIntentId: string; sourceRunId: string; workspaceId: string; cwd: string }
|
||||
| { mode: "fresh_isolated_workspace"; retryIntentId: string; sourceRunId: string; retainedSourceIds: string[] };
|
||||
const freshReadinessSchema = z.object({
|
||||
mode: z.literal("fresh_isolated_workspace"), retryIntentId: z.string().uuid(),
|
||||
@@ -108,6 +109,47 @@ async function reusableDirectoryAfterPreparationFailure(db: Db, source: typeof h
|
||||
return operation ? null : workspace;
|
||||
}
|
||||
|
||||
async function stoppedNativeContinuationWorkspace(db: Db, source: typeof heartbeatRuns.$inferSelect,
|
||||
input: { runId: string; issueId: string; retryIntentId: string }) {
|
||||
if (source.runtimeMode !== "native" || source.status !== "failed" ||
|
||||
source.errorCode !== "native_session_cleanup_quarantined" || !source.nativeSessionId || !source.runnerInstanceId) return null;
|
||||
const [proof] = await db.select({ payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents).where(and(
|
||||
eq(heartbeatRunEvents.companyId, source.companyId), eq(heartbeatRunEvents.runId, source.id),
|
||||
eq(heartbeatRunEvents.agentId, source.agentId), eq(heartbeatRunEvents.eventType, "native.stopped_conversation_verified"),
|
||||
isNull(heartbeatRunEvents.sourceEventId),
|
||||
sql`${heartbeatRunEvents.payload}->>'successorRunId' = ${input.runId}`,
|
||||
sql`${heartbeatRunEvents.payload}->>'taskRetryRequestId' = ${input.retryIntentId}`,
|
||||
)).limit(1);
|
||||
if (!proof) return null;
|
||||
const parsed = z.object({ schema: z.literal("paperclip.stopped_native_conversation.v1"),
|
||||
runId: z.literal(source.id), nativeSessionId: z.literal(source.nativeSessionId), runnerInstanceId: z.literal(source.runnerInstanceId),
|
||||
executionWorkspaceId: z.string().uuid(), workspaceCwd: z.string().min(1),
|
||||
}).safeParse(proof.payload);
|
||||
if (!parsed.success) throw new Error("Verified native continuation workspace is unavailable");
|
||||
if (parsed.data.executionWorkspaceId === source.id) return null;
|
||||
const [bound] = await db.select({ id: executionWorkspaces.id, cwd: executionWorkspaces.cwd }).from(issues).innerJoin(executionWorkspaces, and(
|
||||
eq(executionWorkspaces.companyId, issues.companyId), eq(executionWorkspaces.id, issues.executionWorkspaceId),
|
||||
)).where(and(eq(issues.companyId, source.companyId), eq(issues.id, input.issueId),
|
||||
eq(issues.assigneeAgentId, source.agentId), eq(executionWorkspaces.id, parsed.data.executionWorkspaceId),
|
||||
eq(executionWorkspaces.cwd, parsed.data.workspaceCwd), eq(executionWorkspaces.status, "active"),
|
||||
eq(executionWorkspaces.providerType, "local_fs"),
|
||||
or(isNull(executionWorkspaces.sourceIssueId), eq(executionWorkspaces.sourceIssueId, input.issueId)),
|
||||
)).limit(1);
|
||||
if (!bound) throw new Error("Verified native continuation workspace binding changed");
|
||||
const [operation] = await db.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and(
|
||||
eq(workspaceOperations.companyId, source.companyId), eq(workspaceOperations.executionWorkspaceId, bound.id),
|
||||
eq(workspaceOperations.status, "running"), or(isNull(workspaceOperations.heartbeatRunId), ne(workspaceOperations.heartbeatRunId, input.runId)),
|
||||
)).limit(1);
|
||||
const [lease] = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and(
|
||||
eq(environmentLeases.companyId, source.companyId), eq(environmentLeases.executionWorkspaceId, bound.id),
|
||||
or(isNull(environmentLeases.heartbeatRunId), ne(environmentLeases.heartbeatRunId, input.runId)),
|
||||
or(isNull(environmentLeases.releasedAt), inArray(environmentLeases.status, ["active", "pending_cleanup"])),
|
||||
)).limit(1);
|
||||
if (operation || lease) throw new Error("Verified native continuation workspace cleanup is pending");
|
||||
return { mode: "resume_stopped_native_workspace" as const, retryIntentId: input.retryIntentId,
|
||||
sourceRunId: source.id, workspaceId: bound.id, cwd: parsed.data.workspaceCwd };
|
||||
}
|
||||
|
||||
/** Preparation trusts the committed admission receipt, not a raw wake payload. */
|
||||
export async function readFreshWorkspaceReadiness(db: Db, run: Pick<typeof heartbeatRuns.$inferSelect,
|
||||
"id" | "companyId" | "agentId" | "issueId" | "nativeIssueId" | "contextSnapshot">): Promise<ContinuationReadiness | null> {
|
||||
@@ -151,6 +193,8 @@ export async function readFreshWorkspaceReadiness(db: Db, run: Pick<typeof heart
|
||||
)).limit(1);
|
||||
if (!source) throw new Error("Fresh retry source changed");
|
||||
if (intent.strategy !== "fresh_isolated_workspace") {
|
||||
const stoppedWorkspace = await stoppedNativeContinuationWorkspace(db, source, { runId: run.id, issueId, retryIntentId: wake.id });
|
||||
if (stoppedWorkspace) return stoppedWorkspace;
|
||||
const preservationReceiptId = await verifiedWorkspaceRepairReceipt(db, source);
|
||||
if (preservationReceiptId) return { mode: "resume_preserved_workspace", preservationReceiptId };
|
||||
const [bound] = await db.select({ id: executionWorkspaces.id, metadata: executionWorkspaces.metadata }).from(issues)
|
||||
|
||||
@@ -125,6 +125,26 @@ describe.skipIf(!support.supported)("heartbeat environment selection boundary",
|
||||
expect(result.persistedNativeExecutionInput?.binding.executionWorkspaceId).toBe(input.run.id);
|
||||
});
|
||||
|
||||
it.each(["preserved", "changed_binding", "changed_cwd"])("pins the verified Retry workspace during environment selection (%s)", async mode => {
|
||||
const { input, issue, workspace } = await fixture();
|
||||
const saved = await workspace();
|
||||
await db.update(issues).set({ executionWorkspaceId: saved.id }).where(eq(issues.id, issue.id));
|
||||
input.task.issueRef!.executionWorkspaceId = saved.id;
|
||||
input.task.issueRef!.executionWorkspacePreference = null;
|
||||
input.task.freshWorkspace = { mode: "resume_stopped_native_workspace", retryIntentId: randomUUID(),
|
||||
sourceRunId: randomUUID(), workspaceId: saved.id, cwd: home };
|
||||
if (mode === "changed_binding") input.task.issueRef!.executionWorkspaceId = (await workspace()).id;
|
||||
if (mode === "changed_cwd") await executionWorkspaceService(db).update(saved.id, { cwd: join(home, "changed") });
|
||||
if (mode !== "preserved") {
|
||||
await expect(selectHeartbeatEnvironment(db, input)).rejects.toThrow("The saved Retry workspace binding changed.");
|
||||
return;
|
||||
}
|
||||
const result = await selectHeartbeatEnvironment(db, input);
|
||||
expect(result.requestedShouldReuseExisting).toBe(true);
|
||||
expect(result.reusableExistingExecutionWorkspace).toMatchObject({ id: saved.id, cwd: home });
|
||||
expect(result.nativeRecoveryExecutionWorkspaceId).toBeNull();
|
||||
});
|
||||
|
||||
it("does not make an archived issue workspace available for reuse", async () => {
|
||||
const { input, workspace } = await fixture();
|
||||
const saved = await workspace();
|
||||
|
||||
@@ -145,10 +145,11 @@ export async function selectHeartbeatEnvironment(db: Db, input: HeartbeatEnviron
|
||||
existingExecutionWorkspace = await executionWorkspacesSvc.getStoredById(agent.companyId, existingExecutionWorkspace.id);
|
||||
}
|
||||
const retainedDirectory = input.task.freshWorkspace?.mode === "reuse_current_workspace" ? input.task.freshWorkspace : null;
|
||||
if (retainedDirectory && (issueRef?.executionWorkspaceId !== retainedDirectory.workspaceId ||
|
||||
existingExecutionWorkspace?.id !== retainedDirectory.workspaceId || existingExecutionWorkspace.cwd !== retainedDirectory.cwd)) {
|
||||
const retainedWorkspace = input.task.freshWorkspace?.mode === "resume_stopped_native_workspace" ? input.task.freshWorkspace : retainedDirectory;
|
||||
if (retainedWorkspace && (issueRef?.executionWorkspaceId !== retainedWorkspace.workspaceId ||
|
||||
existingExecutionWorkspace?.id !== retainedWorkspace.workspaceId || existingExecutionWorkspace.cwd !== retainedWorkspace.cwd)) {
|
||||
throw new WorkspaceValidationFailure("The saved Retry workspace binding changed.", {
|
||||
workspaceValidation: { reason: "isolated_task_directory_binding_mismatch", issueId },
|
||||
workspaceValidation: { reason: retainedDirectory ? "isolated_task_directory_binding_mismatch" : "stopped_native_workspace_binding_mismatch", issueId },
|
||||
});
|
||||
}
|
||||
const nativeRecoveryExecutionWorkspaceId =
|
||||
@@ -159,7 +160,7 @@ export async function selectHeartbeatEnvironment(db: Db, input: HeartbeatEnviron
|
||||
const workspaceReuseRequest =
|
||||
resolveExecutionWorkspaceReuseRequestForIssue({
|
||||
issueExecutionWorkspaceId: requestedExecutionWorkspaceId,
|
||||
issueExecutionWorkspacePreference: controllerOnly ? null : nativeRecoveryExecutionWorkspaceId
|
||||
issueExecutionWorkspacePreference: controllerOnly ? null : nativeRecoveryExecutionWorkspaceId || retainedWorkspace
|
||||
? "reuse_existing"
|
||||
: (issueRef?.executionWorkspacePreference ?? null),
|
||||
existingExecutionWorkspaceStatus:
|
||||
|
||||
@@ -26,12 +26,13 @@ function absent(pid: number): boolean {
|
||||
* the original host, process birth, and launch event before accepting it. */
|
||||
export async function recordedLocalExecutionStopped(db: Db, run: Pick<typeof heartbeatRuns.$inferSelect,
|
||||
"id" | "companyId" | "processPid" | "processGroupId" | "processStartedAt">) {
|
||||
const [identity] = await db.select({ payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents).where(and(
|
||||
const [identity] = await db.select({ eventType: heartbeatRunEvents.eventType, payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents).where(and(
|
||||
eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id),
|
||||
eq(heartbeatRunEvents.eventType, PROCESS_IDENTITY_RECORDED), isNull(heartbeatRunEvents.sourceEventId),
|
||||
inArray(heartbeatRunEvents.eventType, [PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED]),
|
||||
isNull(heartbeatRunEvents.sourceEventId),
|
||||
)).orderBy(desc(heartbeatRunEvents.seq)).limit(1);
|
||||
const recorded = identity?.payload;
|
||||
if (recorded?.targetKind !== "local" || recorded.hostname !== os.hostname() ||
|
||||
if (identity?.eventType !== PROCESS_IDENTITY_RECORDED || recorded?.targetKind !== "local" || recorded.hostname !== os.hostname() ||
|
||||
recorded.processPid !== run.processPid || recorded.processGroupId !== run.processGroupId ||
|
||||
!run.processStartedAt || recorded.processStartedAt !== run.processStartedAt.toISOString()) return false;
|
||||
return Boolean(run.processPid && absent(run.processPid) &&
|
||||
|
||||
@@ -5972,15 +5972,20 @@ describe("stopped native conversation physical cleanup", () => {
|
||||
...(mode === "startup_intent" ? { startupAttempt: { phase: "intent" } } : {}),
|
||||
...(mode === "pending_identity" ? { pendingEvents: [{ eventType: "session.started" }] } : {}),
|
||||
};
|
||||
const normalizedEvent = mode === "normalized_session_receipt" ? { ...event, turnId: undefined, itemId: undefined } : event;
|
||||
const normalizedEvent = { ...event, sourceEventId: `${identity.runnerInstanceId}:${identity.runId}:1`,
|
||||
...(mode === "normalized_session_receipt" ? { turnId: undefined, itemId: undefined } : {}) };
|
||||
const receipt = { sourceEventId: `${identity.runnerInstanceId}:${identity.runId}:1`,
|
||||
sourceSeq: 1, sourceInstanceId: identity.runnerInstanceId, eventType: normalizedEvent.eventType,
|
||||
sourcePayloadSha256: mode === "wrong_receipt" ? "wrong" : hash(normalizedEvent), payload: { prpEvent: normalizedEvent } };
|
||||
const normalizedReplacement = { ...replacement, sourceEventId: `${identity.runnerInstanceId}:${identity.runId}:2` };
|
||||
const stop = { eventType: mode === "new_launch" ? "native.process_start_requested" : "native.local_process_stopped",
|
||||
payload: { processPid: mode === "worker_alive" ? process.pid : 99_999_999, processGroupId: mode === "worker_alive" ? process.pid : 99_999_999 } };
|
||||
let queryIndex = 0;
|
||||
const db = { select() { const rows = [
|
||||
[{ provider: mode === "remote" ? "daytona" : "local", releasedAt: mode === "unreleased" ? null : new Date() }],
|
||||
[stop], mode === "missing_receipt" ? [] : [receipt, ...(mode.startsWith("replacement") ? [{ ...receipt, sourceEventId: `${identity.runnerInstanceId}:${identity.runId}:2`, sourcePayloadSha256: hash(replacement), payload: { prpEvent: replacement } }] : [])],
|
||||
[stop], mode === "missing_receipt" ? [] : [receipt, ...(mode.startsWith("replacement") ? [{ ...receipt,
|
||||
sourceEventId: normalizedReplacement.sourceEventId, sourceSeq: 2, eventType: normalizedReplacement.eventType,
|
||||
sourcePayloadSha256: hash(normalizedReplacement), payload: { prpEvent: normalizedReplacement } }] : [])],
|
||||
][queryIndex++] ?? [];
|
||||
const q: any = { from: () => q, where: () => q, orderBy: () => q, limit: () => q,
|
||||
then: Promise.resolve(rows).then.bind(Promise.resolve(rows)) }; return q;
|
||||
|
||||
@@ -95,7 +95,7 @@ import { stoppedCodexTurnIsTextOnly } from "./stopped-codex-turn.js";
|
||||
import { prepareVerifiedRemoteProviderPack, useVerifiedPreinstalledRemoteProviderPack, shouldUseBundledAcpxImageAssets } from "./remote-provider-pack.js";
|
||||
import { uploadVerifiedProviderPack } from "./provider-pack-upload.js";
|
||||
import { selectRemotePiCompanion } from "./remote-pi-companion.js";
|
||||
import { readNativeLocalProcessStop, PROCESS_START_REQUESTED, withReleasedNativeRunProcess } from "../native-local-process-stop.js";
|
||||
import { readNativeLocalProcessStop, recordedLocalExecutionStopped, PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED, withReleasedNativeRunProcess } from "../native-local-process-stop.js";
|
||||
import { remoteLeaseCleanupScope } from "../remote-execution-termination.js";
|
||||
import { connectorToolDefinitionsForSkills } from "../connector-runtime.js";
|
||||
import { setNativeExecutionActive } from "./native-active-runs.js";
|
||||
@@ -3015,7 +3015,8 @@ export function rebaseRetainedNativeCleanupProviderHome(
|
||||
export async function verifyStoppedNativeSessionForContinuation(
|
||||
db: Db,
|
||||
run: Pick<typeof heartbeatRuns.$inferSelect, "id" | "companyId" | "agentId" | "runtimeMode" |
|
||||
"status" | "finishedAt" | "nativeIssueId" | "nativeSessionId" | "runnerInstanceId" | "runnerProfileJson">,
|
||||
"status" | "finishedAt" | "nativeIssueId" | "nativeSessionId" | "runnerInstanceId" | "runnerProfileJson" |
|
||||
"errorCode" | "createdAt" | "processPid" | "processGroupId" | "processStartedAt">,
|
||||
): Promise<{ evidence: Record<string, unknown>; retire: () => boolean } | null> {
|
||||
try {
|
||||
if (run.runtimeMode !== "native" || !["failed", "cancelled", "interrupted", "timed_out"].includes(run.status) ||
|
||||
@@ -3024,22 +3025,66 @@ export async function verifyStoppedNativeSessionForContinuation(
|
||||
if (!(["codex", "acpx"] as string[]).includes(execution.provider.kind) ||
|
||||
execution.binding.runId !== run.id || execution.binding.companyId !== run.companyId ||
|
||||
execution.binding.agentId !== run.agentId || execution.binding.issueId !== run.nativeIssueId ||
|
||||
nativeSessionKey(execution) !== run.nativeSessionId ||
|
||||
record(run.runnerProfileJson).nativeToolContractFingerprint !== nativeToolContractFingerprintForTarget("local")) return null;
|
||||
nativeSessionKey(execution) !== run.nativeSessionId) return null;
|
||||
const scope = nativeSessionScopeKey(execution);
|
||||
const idle = () => !activeNativeSessions.has(run.id) && !executingRunnerdSessionScopes.has(scope) &&
|
||||
!initializingSessionToolAuthorities.has(scope) && !warmNativeSessions.has(scope);
|
||||
if (!idle()) return null;
|
||||
const leases = await db.select().from(environmentLeases).where(and(
|
||||
eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id)));
|
||||
if (leases.some(lease => lease.provider !== "local" || !lease.releasedAt || lease.cleanupStatus === "failed")) return null;
|
||||
const stopped = await readNativeLocalProcessStop(db, run.companyId, run.id);
|
||||
if (!stopped) return null;
|
||||
if (leases.some(lease => lease.provider !== "local" || !lease.releasedAt ||
|
||||
lease.status === "pending_cleanup" || lease.cleanupStatus === "failed")) return null;
|
||||
const root = scopedRunnerdStateRoot(execution);
|
||||
const providerFile = runnerProviderStateFilename(execution);
|
||||
const snapshot = cleanupStateSnapshot(root, providerFile);
|
||||
const identity = record(snapshot.control.identity);
|
||||
if (!durableIdentityMatchesExecution(identity, execution) || identity.runnerInstanceId !== run.runnerInstanceId ||
|
||||
let owner = run;
|
||||
let ownerExecution = execution;
|
||||
if (identity.runId !== run.id) {
|
||||
if (run.errorCode !== "native_session_cleanup_quarantined" || typeof identity.runId !== "string") return null;
|
||||
const [started] = await db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and(
|
||||
eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id),
|
||||
inArray(heartbeatRunEvents.eventType, [PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED,
|
||||
"session.started", "session.resumed", "session.reconciled"]))).limit(1);
|
||||
if (started) return null;
|
||||
const [retainedRow] = await db.select().from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.id, identity.runId),
|
||||
eq(heartbeatRuns.agentId, run.agentId), eq(heartbeatRuns.nativeIssueId, run.nativeIssueId),
|
||||
eq(heartbeatRuns.nativeSessionId, run.nativeSessionId), eq(heartbeatRuns.runnerInstanceId, run.runnerInstanceId),
|
||||
)).for("update", { noWait: true }).limit(1);
|
||||
const retained = retainedRow ? await withReleasedNativeRunProcess(db, retainedRow) : null;
|
||||
if (!retained || retained.runtimeMode !== "native" ||
|
||||
!["succeeded", "failed", "cancelled", "interrupted", "timed_out"].includes(retained.status) ||
|
||||
!retained.finishedAt || retained.finishedAt > run.createdAt ||
|
||||
retained.controllerLeaseExpiresAt && retained.controllerLeaseExpiresAt > new Date() ||
|
||||
!retained.processStartedAt || !run.processStartedAt ||
|
||||
retained.processStartedAt.getTime() !== run.processStartedAt.getTime() ||
|
||||
retained.processPid !== run.processPid || retained.processGroupId !== run.processGroupId) return null;
|
||||
const [coordinator] = await db.select().from(nativeRunFinalizations).where(and(
|
||||
eq(nativeRunFinalizations.companyId, run.companyId), eq(nativeRunFinalizations.issueId, run.nativeIssueId),
|
||||
eq(nativeRunFinalizations.runId, retained.id),
|
||||
)).for("update", { noWait: true }).limit(1);
|
||||
if (!coordinator || !["committed", "terminal_failure"].includes(coordinator.phase) || coordinator.leaseOwner ||
|
||||
coordinator.nextAttemptAt || coordinator.recoveryState) return null;
|
||||
ownerExecution = parseNativeExecutionInput(record(retained.runnerProfileJson).nativeExecutionInput);
|
||||
if (ownerExecution.binding.runId !== retained.id || ownerExecution.binding.companyId !== run.companyId ||
|
||||
ownerExecution.binding.agentId !== run.agentId || ownerExecution.binding.issueId !== run.nativeIssueId ||
|
||||
nativeSessionScopeKey(ownerExecution) !== scope || ownerExecution.workspace.cwd !== execution.workspace.cwd) return null;
|
||||
const ownerLeases = await db.select().from(environmentLeases).where(and(
|
||||
eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, retained.id)));
|
||||
if (ownerLeases.some(lease => lease.provider !== "local" || !lease.releasedAt ||
|
||||
lease.status === "pending_cleanup" || lease.cleanupStatus === "failed")) return null;
|
||||
owner = retained;
|
||||
}
|
||||
let stop = await readNativeLocalProcessStop(db, owner.companyId, owner.id);
|
||||
if (!stop && owner.processPid && owner.processPid > 1 && owner.processGroupId === owner.processPid &&
|
||||
await recordedLocalExecutionStopped(db, owner)) {
|
||||
stop = { processPid: owner.processPid, processGroupId: owner.processPid,
|
||||
...(owner.processStartedAt ? { processStartedAt: owner.processStartedAt.toISOString() } : {}) };
|
||||
}
|
||||
const stopped = stop;
|
||||
if (!stopped || activeNativeSessions.has(owner.id)) return null;
|
||||
if (!durableIdentityMatchesExecution(identity, ownerExecution) || identity.runnerInstanceId !== owner.runnerInstanceId ||
|
||||
!["runnerInstanceId", "environmentLeaseId", "runId", "normalizedSessionId", "turnId", "itemId"].every(key =>
|
||||
typeof identity[key] === "string" && identity[key] && snapshot.runner[key] === identity[key]) ||
|
||||
!Array.isArray(snapshot.control.committedEvents)) return null;
|
||||
@@ -3048,7 +3093,7 @@ export async function verifyStoppedNativeSessionForContinuation(
|
||||
if (!RUNNERD_CONTROL_PLANE_STATE_SCHEMAS.has(String(snapshot.control.schema)) ||
|
||||
snapshot.runner.schema !== RUNNERD_STATE_SCHEMA ||
|
||||
snapshot.provider.schema !== (execution.provider.kind === "codex" ? "paperclip.runner.codex-provider-state.v1" : ACPX_PROVIDER_STATE_SCHEMA) ||
|
||||
!["turn_active", "prepared", "suspended"].includes(String(snapshot.provider.lifecycle)) ||
|
||||
!["turn_active", "prepared", "suspended", "session_open"].includes(String(snapshot.provider.lifecycle)) ||
|
||||
snapshot.provider.startupAttempt != null) return null;
|
||||
const pending = JSON.stringify([snapshot.runner.outbox, snapshot.provider.pendingEvents, snapshot.provider.queuedEvents]);
|
||||
if (/session\.(started|resumed|reconciled)/.test(pending)) return null;
|
||||
@@ -3056,36 +3101,42 @@ export async function verifyStoppedNativeSessionForContinuation(
|
||||
const providerEvents = events.filter(event => ["session.started", "session.resumed", "session.reconciled"].includes(String(event.eventType)));
|
||||
if (!providerEvents.length) return null;
|
||||
const receipts = await db.select().from(heartbeatRunEvents).where(and(
|
||||
eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id),
|
||||
eq(heartbeatRunEvents.sourceInstanceId, run.runnerInstanceId),
|
||||
eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, owner.id),
|
||||
eq(heartbeatRunEvents.sourceInstanceId, owner.runnerInstanceId!),
|
||||
inArray(heartbeatRunEvents.eventType, ["session.started", "session.resumed", "session.reconciled", "harness.diagnostic"])));
|
||||
const providerReceipts = receipts.filter(row => row.eventType !== "harness.diagnostic")
|
||||
.sort((a, b) => Number(a.sourceSeq) - Number(b.sourceSeq));
|
||||
if (providerReceipts.length !== providerEvents.length) return null;
|
||||
const providerPids = new Set<number>();
|
||||
for (const event of providerEvents) {
|
||||
for (const [index, event] of providerEvents.entries()) {
|
||||
const provider = record(event.payload);
|
||||
if (!validatePrpEvent(event).ok || event.sourceKind !== "runner" || event.sourceInstanceId !== run.runnerInstanceId ||
|
||||
event.runId !== run.id || event.normalizedSessionId !== run.nativeSessionId ||
|
||||
if (!validatePrpEvent(event).ok || event.sourceKind !== "runner" || event.sourceInstanceId !== owner.runnerInstanceId ||
|
||||
event.runId !== owner.id || event.normalizedSessionId !== run.nativeSessionId ||
|
||||
typeof provider.providerSessionId !== "string" || !provider.providerSessionId ||
|
||||
!cleanupProcessAbsent(provider.processId) || provider.processId === stopped.processPid) return null;
|
||||
if (event.eventType === "session.reconciled" &&
|
||||
(!providerPids.has(Number(provider.previousProcessId)) || !cleanupProcessAbsent(provider.previousProcessId))) return null;
|
||||
const receipt = receipts.find(row => row.sourceEventId === `${run.runnerInstanceId}:${run.id}:${event.sourceSeq}`);
|
||||
const receipt = providerReceipts[index];
|
||||
const durable = record(record(receipt?.payload).prpEvent);
|
||||
const durableProvider = record(durable.payload);
|
||||
if (!receipt || !validatePrpEvent(durable).ok || durable.sourceInstanceId !== event.sourceInstanceId ||
|
||||
durable.runId !== run.id || durable.normalizedSessionId !== run.nativeSessionId ||
|
||||
durable.runId !== owner.id || durable.normalizedSessionId !== run.nativeSessionId ||
|
||||
// Normalized session-open receipts omit turn/item IDs. Those belong
|
||||
// to the durable runner identity checked above, not the open event.
|
||||
durable.sourceSeq !== event.sourceSeq || durable.eventType !== event.eventType ||
|
||||
durable.sourceSeq !== receipt.sourceSeq || durable.eventType !== event.eventType ||
|
||||
receipt.eventType !== durable.eventType || receipt.sourceInstanceId !== durable.sourceInstanceId ||
|
||||
receipt.sourceEventId !== `${owner.runnerInstanceId}:${owner.id}:${durable.sourceSeq}` ||
|
||||
durable.sourceEventId !== receipt.sourceEventId ||
|
||||
receipt.sourcePayloadSha256 !== nativeSha256(durable) ||
|
||||
(durableProvider.processId !== undefined && durableProvider.processId !== provider.processId) ||
|
||||
(durableProvider.driverSessionId ?? durableProvider.providerSessionId) !== provider.providerSessionId) return null;
|
||||
providerPids.add(provider.processId);
|
||||
}
|
||||
if (receipts.filter(row => record(record(row.payload).prpEvent).eventType !== "harness.diagnostic").length !== providerEvents.length) return null;
|
||||
// ACPX's sidecar is not its agent process. Include separately recorded
|
||||
// owners from both the durable journal and the independent DB/checkpoint.
|
||||
// Extra evidence can only add a stop requirement, never waive one.
|
||||
const owners: Record<string, unknown>[] = [record(record(run.runnerProfileJson?.sessionCheckpoint).process),
|
||||
record(record(owner.runnerProfileJson?.sessionCheckpoint).process),
|
||||
snapshot.provider, record(snapshot.provider.identity), record(snapshot.provider.descriptor)];
|
||||
for (const event of [...events, ...receipts.map(row => record(record(row.payload).prpEvent))]) {
|
||||
const payload = record(event.payload);
|
||||
@@ -3103,16 +3154,18 @@ export async function verifyStoppedNativeSessionForContinuation(
|
||||
if (!cleanupProcessAbsent(pid)) return null;
|
||||
providerPids.add(pid);
|
||||
}
|
||||
const unchanged = () => idle() && cleanupProcessAbsent(stopped.processPid) &&
|
||||
const unchanged = () => idle() && !activeNativeSessions.has(owner.id) && cleanupProcessAbsent(stopped.processPid) &&
|
||||
[...providerPids].every(cleanupProcessAbsent) && cleanupStateSnapshot(root, providerFile).fingerprint === snapshot.fingerprint;
|
||||
return {
|
||||
evidence: { schema: "paperclip.stopped_native_conversation.v1", runId: run.id,
|
||||
ownerRunId: owner.id,
|
||||
executionWorkspaceId: execution.binding.executionWorkspaceId, workspaceCwd: execution.workspace.cwd,
|
||||
nativeSessionId: run.nativeSessionId, runnerInstanceId: run.runnerInstanceId,
|
||||
processPid: stopped.processPid, providerPids: [...providerPids], stateFingerprint: snapshot.fingerprint },
|
||||
retire: () => {
|
||||
try { return unchanged() && completeTerminatedLocalNativeSessionCleanup({
|
||||
companyId: run.companyId, runId: run.id, runnerInstanceId: run.runnerInstanceId!,
|
||||
}); } catch { return false; }
|
||||
try { return unchanged() && [owner.id, run.id].every(runId => completeTerminatedLocalNativeSessionCleanup({
|
||||
companyId: run.companyId, runId, runnerInstanceId: run.runnerInstanceId!,
|
||||
})); } catch { return false; }
|
||||
},
|
||||
};
|
||||
} catch { return null; }
|
||||
|
||||
@@ -0,0 +1,141 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { copyFile, mkdir, readFile, writeFile } from "node:fs/promises";
|
||||
import { homedir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { agents, agentWakeupRequests, companies, heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations } from "@paperclipai/db";
|
||||
import { expect, it } from "vitest";
|
||||
import { startRunnerApiTestServer } from "../../__tests__/helpers/runner-api-server.js";
|
||||
import { createIsolatedTestHome } from "../../__tests__/helpers/isolated-test-home.js";
|
||||
import { heartbeatService } from "../heartbeat.js";
|
||||
import { issueService } from "../issues.js";
|
||||
import { instanceSettingsService } from "../instance-settings.js";
|
||||
import { withReleasedNativeRunProcess } from "../native-local-process-stop.js";
|
||||
import { nativeSha256 } from "./canonical.js";
|
||||
import { closeIdleWarmNativeSessionsForRestart } from "./native-session-executor.js";
|
||||
import { parseNativeExecutionInput } from "../../vendor/paperclip-runner/index.js";
|
||||
|
||||
it.skipIf(process.env.PAPERCLIP_LIVE_QUARANTINED_RETRY !== "true")("starts a fresh real Codex turn through task Retry after inherited quarantine", async () => {
|
||||
const home = createIsolatedTestHome();
|
||||
const keys = ["PAPERCLIP_HOME", "CODEX_HOME", "PAPERCLIP_API_URL", "PAPERCLIP_AGENT_JWT_SECRET", "PAPERCLIP_TELEMETRY_DISABLED", "PAPERCLIP_RUNNER_STATE_DIR"];
|
||||
const saved = Object.fromEntries(keys.map(key => [key, process.env[key]]));
|
||||
let server: Awaited<ReturnType<typeof startRunnerApiTestServer>> | undefined;
|
||||
let heartbeat: ReturnType<typeof heartbeatService> | undefined;
|
||||
let companyId: string | undefined;
|
||||
try {
|
||||
const codexHome = join(home.home, "codex"), stateBase = join(home.home, "runner-state");
|
||||
await mkdir(codexHome, { mode: 0o700 });
|
||||
await copyFile(join(process.env.PAPERCLIP_LIVE_CODEX_HOME ?? join(homedir(), ".codex"), "auth.json"), join(codexHome, "auth.json"));
|
||||
process.env.CODEX_HOME = codexHome;
|
||||
process.env.PAPERCLIP_HOME = join(home.home, "instance");
|
||||
process.env.PAPERCLIP_RUNNER_STATE_DIR = stateBase;
|
||||
process.env.PAPERCLIP_AGENT_JWT_SECRET = randomUUID();
|
||||
process.env.PAPERCLIP_TELEMETRY_DISABLED = "1";
|
||||
server = await startRunnerApiTestServer({ deploymentMode: "authenticated", humanSessions: true });
|
||||
process.env.PAPERCLIP_API_URL = server.apiUrl;
|
||||
const f = await server.fixture({ connectionScenario: "fresh" }), db = server.db;
|
||||
companyId = f.companyId;
|
||||
const markers = [randomUUID(), randomUUID()];
|
||||
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, f.runId));
|
||||
await db.update(issues).set({ status: "todo", executionRunId: null, executionWorkspaceSettings: { mode: "shared_workspace" },
|
||||
title: "Preserve work through task Retry", description: "Follow the latest human comment. Use the task workspace. Complete only that requested file step and finish with paperclip_finish, reportedWorkDisposition done. This is a local file task; no external writes are needed.",
|
||||
}).where(eq(issues.id, f.issueId));
|
||||
await db.update(companies).set({ defaultResponsibleUserId: f.userId }).where(eq(companies.id, f.companyId));
|
||||
await db.update(agents).set({ adapterConfig: { provider: "codex", model: "gpt-6.1-sol", modelReasoningEffort: "high",
|
||||
cwd: f.workspace, workspaceStrategy: { type: "project_primary" }, env: { CODEX_HOME: codexHome }, timeoutSec: 180, lifecycleMode: "per_turn" },
|
||||
runtimeConfig: { heartbeat: { enabled: false, wakeOnDemand: true, maxConcurrentRuns: 1 } },
|
||||
}).where(eq(agents.id, f.agentId));
|
||||
await instanceSettingsService(db).updateExperimental({ enableNativeRunner: true, enableIsolatedWorkspaces: true });
|
||||
heartbeat = heartbeatService(db);
|
||||
const service = issueService(db);
|
||||
const first = await service.addComment(f.issueId, `Write exactly ${markers[0]} and a newline to retry-proof.txt. Read it, then finish.`, { userId: f.userId! });
|
||||
const run = await heartbeat.wakeup(f.agentId, { source: "automation", triggerDetail: "system", reason: "issue_commented",
|
||||
requestedByActorType: "user", requestedByActorId: f.userId,
|
||||
payload: { issueId: f.issueId, commentId: first.id },
|
||||
contextSnapshot: { issueId: f.issueId, taskId: f.issueId, wakeCommentId: first.id, skipIssueComment: true },
|
||||
});
|
||||
expect(run?.id).toBeTruthy();
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
await closeIdleWarmNativeSessionsForRestart();
|
||||
const [original] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run!.id));
|
||||
expect({ status: original.status, error: original.error }).toMatchObject({ status: "succeeded", error: null });
|
||||
expect(await readFile(join(f.workspace, "retry-proof.txt"), "utf8")).toBe(`${markers[0]}\n`);
|
||||
const owner = await withReleasedNativeRunProcess(db, original);
|
||||
expect(owner.processPid).toBeTruthy();
|
||||
const execution = parseNativeExecutionInput(owner.runnerProfileJson!.nativeExecutionInput);
|
||||
const root = join(stateBase, nativeSha256({ schema: "paperclip.native-session-scope.v2", companyId: f.companyId, agentId: f.agentId,
|
||||
workspace: { kind: "managed", executionWorkspaceId: execution.binding.executionWorkspaceId },
|
||||
provider: { driverKind: "codex_app_server", identity: { kind: "codex" } }, normalizedSessionId: owner.nativeSessionId }));
|
||||
const controlPath = join(root, "control-plane/control-plane-state.json"), providerPath = join(root, "runner/codex-provider-state.json");
|
||||
const control = JSON.parse(await readFile(controlPath, "utf8")), provider = JSON.parse(await readFile(providerPath, "utf8"));
|
||||
control.commands.push({ commandId: "unacknowledged-result", status: "pending", payload: { commandType: "tool.result", result: "Retain this uncertain command without replay." } });
|
||||
provider.lifecycle = "session_open";
|
||||
await writeFile(controlPath, JSON.stringify(control));
|
||||
await writeFile(providerPath, JSON.stringify(provider));
|
||||
const retainedBytes = await readFile(controlPath, "utf8"), sourceId = randomUUID();
|
||||
const retryComment = await service.addComment(f.issueId, `Read retry-proof.txt and preserve its existing line. Append exactly ${markers[1]} and a newline once. Never repeat the earlier file step. Read the complete file and finish.`, { userId: f.userId! });
|
||||
const oldFingerprint = `sha256:${"1".repeat(64)}`;
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { ...owner.runnerProfileJson, nativeToolContractFingerprint: oldFingerprint } }).where(eq(heartbeatRuns.id, owner.id));
|
||||
await db.insert(heartbeatRuns).values({ id: sourceId, companyId: f.companyId, agentId: f.agentId, issueId: f.issueId,
|
||||
nativeIssueId: f.issueId, runtimeMode: "native", nativeSessionId: owner.nativeSessionId, runnerInstanceId: owner.runnerInstanceId,
|
||||
processPid: owner.processPid, processGroupId: owner.processGroupId, processStartedAt: owner.processStartedAt,
|
||||
status: "failed", finishedAt: new Date(), errorCode: "native_session_cleanup_quarantined", responsibleUserId: f.userId,
|
||||
contextSnapshot: { issueId: f.issueId, commentId: retryComment.id, wakeCommentId: retryComment.id, wakeCommentIds: [retryComment.id] }, runnerProfileJson: { ...owner.runnerProfileJson,
|
||||
nativeExecutionInput: { ...execution, binding: { ...execution.binding, runId: sourceId },
|
||||
task: { ...execution.task, prompt: retryComment.body } }, nativeToolContractFingerprint: oldFingerprint },
|
||||
});
|
||||
await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: f.issueId, runId: sourceId, phase: "terminal_failure" });
|
||||
await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.issueId, kind: "active_run_watchdog",
|
||||
cause: "native_session_cleanup_quarantined", fingerprint: sourceId, status: "resolved", outcome: "blocked",
|
||||
nextAction: "Automatic recovery stopped.", evidence: { runId: sourceId, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } } });
|
||||
await db.update(issues).set({ status: "blocked", executionRunId: null, checkoutRunId: null }).where(eq(issues.id, f.issueId));
|
||||
const cookie = await server.humanSession(f.userId!);
|
||||
const response = await fetch(`${server.apiUrl}/api/issues/${f.issueId}/retry`, { method: "POST",
|
||||
headers: { "content-type": "application/json", cookie, origin: server.apiUrl },
|
||||
body: JSON.stringify({ observedRunId: sourceId, idempotencyKey: randomUUID() }),
|
||||
});
|
||||
const receipt = await response.json() as { requestId: string; accepted: boolean };
|
||||
expect(response.status, JSON.stringify(receipt)).toBe(202);
|
||||
expect(receipt.accepted).toBe(true);
|
||||
await expect.poll(async () => {
|
||||
const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.requestId));
|
||||
return wake.runId;
|
||||
}, { timeout: 30_000, interval: 250 }).toBeTruthy();
|
||||
const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.requestId));
|
||||
await expect.poll(async () => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, wake.runId!)))[0].status,
|
||||
{ timeout: 240_000, interval: 500 }).toBe("succeeded");
|
||||
const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, wake.runId!));
|
||||
expect(successor.nativeSessionId).not.toBe(owner.nativeSessionId);
|
||||
expect(successor.sessionIdAfter).not.toBe(owner.sessionIdAfter);
|
||||
expect(successor.contextSnapshot).toMatchObject({ previousRunId: sourceId, forceFreshSession: true });
|
||||
expect(await readFile(join(f.workspace, "retry-proof.txt"), "utf8")).toBe(markers.join("\n") + "\n");
|
||||
expect(await readFile(controlPath, "utf8")).toBe(retainedBytes);
|
||||
const [task] = await db.select().from(issues).where(eq(issues.id, f.issueId));
|
||||
expect(task).toMatchObject({ status: "done", executionRunId: null, checkoutRunId: null });
|
||||
const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, successor.id));
|
||||
expect(coordinator.phase).toBe("committed");
|
||||
const proof = await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, sourceId), eq(heartbeatRunEvents.eventType, "native.stopped_conversation_verified")));
|
||||
expect(proof).toHaveLength(1);
|
||||
const evidence = { passed: true, platform: `${process.platform}-${process.arch}`, model: "gpt-6.1-sol", effort: "high",
|
||||
ownerRunId: owner.id, sourceRunId: sourceId, successorRunId: successor.id, freshSession: true,
|
||||
priorJournalPreserved: true, fileLines: 2, taskStatus: task.status, locksCleared: true, finalization: coordinator.phase };
|
||||
if (process.env.PAPERCLIP_LIVE_EVIDENCE_PATH) await writeFile(process.env.PAPERCLIP_LIVE_EVIDENCE_PATH, JSON.stringify(evidence, null, 2) + "\n");
|
||||
console.log(JSON.stringify(evidence));
|
||||
} catch (error) {
|
||||
if (server && companyId && process.env.PAPERCLIP_LIVE_EVIDENCE_PATH) {
|
||||
const runs = await server.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId));
|
||||
await writeFile(`${process.env.PAPERCLIP_LIVE_EVIDENCE_PATH}.failure.json`, JSON.stringify({ error: String(error),
|
||||
runs: runs.map(run => ({ id: run.id, status: run.status, error: run.error, result: run.resultJson,
|
||||
execution: (run.runnerProfileJson?.nativeExecutionInput as { workspace?: unknown } | undefined)?.workspace,
|
||||
continuation: run.contextSnapshot?.executionContinuation })),
|
||||
}, null, 2));
|
||||
}
|
||||
throw error;
|
||||
} finally {
|
||||
await heartbeat?.drainActiveRunExecutions();
|
||||
await closeIdleWarmNativeSessionsForRestart();
|
||||
await server?.close();
|
||||
for (const key of keys) { if (saved[key] === undefined) delete process.env[key]; else process.env[key] = saved[key]; }
|
||||
home.cleanup();
|
||||
}
|
||||
}, 600_000);
|
||||
@@ -0,0 +1,243 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { mkdtemp, mkdir, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import { hostname, tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import {
|
||||
agents, agentWakeupRequests, companies, createDb, environmentLeases, executionWorkspaces, heartbeatRunEvents,
|
||||
heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, workspaceOperations,
|
||||
} from "@paperclipai/db";
|
||||
import { startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js";
|
||||
import { parseNativeExecutionInput } from "../vendor/paperclip-runner/index.js";
|
||||
import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js";
|
||||
import { heartbeatService } from "./heartbeat.js";
|
||||
import { getExecutionBlocker } from "./execution-blocker.js";
|
||||
import { readFreshWorkspaceReadiness } from "./fresh-workspace-continuation.js";
|
||||
import { instanceSettingsService } from "./instance-settings.js";
|
||||
import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED } from "./native-local-process-stop.js";
|
||||
import { nativeSha256 } from "./native-runtime/canonical.js";
|
||||
import { verifyStoppedNativeSessionForContinuation } from "./native-runtime/native-session-executor.js";
|
||||
import { requestTaskRetry } from "./task-retry.js";
|
||||
|
||||
describe("task Retry after inherited native cleanup quarantine", () => {
|
||||
let postgres: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
beforeAll(async () => {
|
||||
postgres = await startEmbeddedPostgresTestDatabase("quarantined-task-retry-");
|
||||
db = createDb(postgres.connectionString);
|
||||
await instanceSettingsService(db).updateExperimental({ enableNativeRunner: true });
|
||||
}, 90_000);
|
||||
afterAll(async () => { await db?.$client.end(); await postgres?.cleanup(); });
|
||||
|
||||
async function seed(base: string, transient = false) {
|
||||
const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID();
|
||||
const ownerId = randomUUID(), sourceId = randomUUID(), nativeSessionId = randomUUID(), runnerInstanceId = randomUUID();
|
||||
const workspaceId = randomUUID();
|
||||
const execution = (runId: string) => parseNativeExecutionInput({
|
||||
schema: "paperclip.native-execution-input.v1", provider: { kind: "codex", model: null },
|
||||
binding: { companyId, agentId, issueId, runId, executionWorkspaceId: transient ? runId : workspaceId },
|
||||
task: { identifier: "TEST", title: "Continue saved work", description: null, prompt: "Continue saved work", workMode: "standard" },
|
||||
workspace: { cwd: base, repoUrl: null, repoRef: null, branchName: null },
|
||||
session: { normalizedSessionId: nativeSessionId, driverKind: "codex_app_server", protocolVersion: 1,
|
||||
lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } },
|
||||
completionContract: { id: "contract", sha256: "sha", schemaVersion: "paperclip.completion-contract.v1",
|
||||
contract: { revision: "1", objective: "Continue saved work", criteria: [{ id: "done", requirement: "Save the result" }] } },
|
||||
interactionResponses: [], credentialBindings: [],
|
||||
});
|
||||
const processStartedAt = new Date(Date.now() - 120_000);
|
||||
const processIdentity = { processPid: 99_999_999, processGroupId: 99_999_999, processStartedAt };
|
||||
await db.insert(companies).values({ id: companyId, name: "Quarantined Retry", defaultResponsibleUserId: "board",
|
||||
issuePrefix: `Q${companyId.replaceAll("-", "")}` });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Native worker", adapterType: "paperclip_runner", status: "idle",
|
||||
runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } });
|
||||
await db.insert(issues).values({ id: issueId, companyId, title: "Continue saved work", status: "blocked", assigneeAgentId: agentId });
|
||||
if (!transient) {
|
||||
await db.insert(executionWorkspaces).values({ id: workspaceId, companyId, sourceIssueId: issueId,
|
||||
mode: "shared_workspace", strategyType: "project_primary", name: "Saved work", cwd: base });
|
||||
await db.update(issues).set({ executionWorkspaceId: workspaceId }).where(eq(issues.id, issueId));
|
||||
}
|
||||
const resultJson = { savedAnswer: "Existing work", recoveredExecutionFailure: { errorCode: "provider_transport_failed" } };
|
||||
for (const runId of [ownerId, sourceId]) {
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, issueId, nativeIssueId: issueId,
|
||||
runtimeMode: "native", nativeSessionId, runnerInstanceId, ...processIdentity,
|
||||
status: runId === ownerId ? "succeeded" : "failed", resultJson: runId === ownerId ? resultJson : null,
|
||||
errorCode: runId === sourceId ? "native_session_cleanup_quarantined" : null,
|
||||
createdAt: new Date(Date.now() - (runId === ownerId ? 110_000 : 60_000)),
|
||||
startedAt: new Date(Date.now() - (runId === ownerId ? 109_000 : 59_000)),
|
||||
finishedAt: new Date(Date.now() - (runId === ownerId ? 100_000 : 50_000)),
|
||||
contextSnapshot: { issueId }, runnerProfileJson: { nativeExecutionInput: execution(runId),
|
||||
nativeToolContractFingerprint: `sha256:${"1".repeat(64)}`, sessionCheckpoint: { sessionId: "retained-provider-thread" } },
|
||||
});
|
||||
await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId,
|
||||
phase: runId === ownerId ? "committed" : "terminal_failure" });
|
||||
await db.insert(environmentLeases).values({ companyId, issueId, heartbeatRunId: runId,
|
||||
provider: "local", status: "failed", leasePolicy: "ephemeral", releasedAt: new Date(Date.now() - 40_000) });
|
||||
}
|
||||
await appendHeartbeatRunEvent(db, { companyId, runId: ownerId, agentId, eventType: PROCESS_IDENTITY_RECORDED,
|
||||
payload: { ...processIdentity, processStartedAt: processStartedAt.toISOString(), targetKind: "local", hostname: hostname() } });
|
||||
await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId, kind: "active_run_watchdog",
|
||||
cause: "native_session_cleanup_quarantined", fingerprint: sourceId, status: "resolved", outcome: "blocked",
|
||||
nextAction: "Automatic recovery stopped.", evidence: { runId: sourceId,
|
||||
automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } } });
|
||||
const identity = { runId: ownerId, runnerInstanceId, normalizedSessionId: nativeSessionId,
|
||||
environmentLeaseId: transient ? ownerId : workspaceId, turnId: `turn-${ownerId}`, itemId: `item-${ownerId}` };
|
||||
const event = { schema: "paperclip.prp.event.v1", schemaVersion: 1, sourceKind: "runner", sourceSeq: 5,
|
||||
sourceEventId: "provider-open", sourceInstanceId: runnerInstanceId, runId: ownerId,
|
||||
normalizedSessionId: nativeSessionId, turnId: identity.turnId, itemId: identity.itemId,
|
||||
priority: 0, emittedAt: processStartedAt.toISOString(), eventType: "session.resumed",
|
||||
payload: { providerSessionId: "retained-provider-thread", processId: 99_999_998 } };
|
||||
const normalized = { ...event, sourceSeq: 1, sourceEventId: `${runnerInstanceId}:${ownerId}:1`, turnId: undefined, itemId: undefined,
|
||||
payload: { providerSessionId: "retained-provider-thread", driverSessionId: "retained-provider-thread" } };
|
||||
await appendHeartbeatRunEvent(db, { companyId, runId: ownerId, agentId, eventType: event.eventType,
|
||||
payload: { prpEvent: normalized }, nativeSource: { sourceInstanceId: runnerInstanceId,
|
||||
sourceEventId: normalized.sourceEventId, sourceSeq: 1, protocolSchemaVersion: 1, canonicalPayload: normalized } });
|
||||
const root = join(base, nativeSha256({ schema: "paperclip.native-session-scope.v2", companyId, agentId,
|
||||
workspace: transient ? { kind: "transient", cwd: base, repoUrl: null, repoRef: null, branchName: null }
|
||||
: { kind: "managed", executionWorkspaceId: workspaceId },
|
||||
provider: { driverKind: "codex_app_server", identity: { kind: "codex" } }, normalizedSessionId: nativeSessionId }));
|
||||
const files = {
|
||||
"control-plane/control-plane-state.json": { schema: "paperclip.runner.durable.control-plane-state.v1", identity,
|
||||
committedEvents: [{ envelope: { payload: event } }], commands: [{ status: "pending", payload: { commandType: "tool.result" } }] },
|
||||
"runner/runner-state.json": { schema: "paperclip.runner.durable.state.v1", ...identity, lifecycle: "suspended", outbox: [] },
|
||||
"runner/codex-provider-state.json": { schema: "paperclip.runner.codex-provider-state.v1", lifecycle: "session_open",
|
||||
threadId: "retained-provider-thread", startupAttempt: null, pendingEvents: [], queuedEvents: [] },
|
||||
};
|
||||
await mkdir(join(root, "control-plane"), { recursive: true });
|
||||
await mkdir(join(root, "runner"), { recursive: true });
|
||||
for (const [path, value] of Object.entries(files)) await writeFile(join(root, path), JSON.stringify(value));
|
||||
await writeFile(join(base, "saved-work.txt"), "Keep the prior result and uncertain action history.");
|
||||
return { companyId, agentId, issueId, workspaceId, ownerId, sourceId, runnerInstanceId, resultJson, root, files };
|
||||
}
|
||||
|
||||
it.each(["preserved", "projectless", "changed_binding", "changed_cwd", "pending_operation", "active_lease", "closed_workspace", "remote_workspace", "foreign_receipt"])(
|
||||
"admits one fresh successor through canonical Retry without replaying the retained session (%s)", async mode => {
|
||||
const base = await mkdtemp(join(tmpdir(), "quarantined-retry-"));
|
||||
const previous = process.env.PAPERCLIP_RUNNER_STATE_DIR;
|
||||
process.env.PAPERCLIP_RUNNER_STATE_DIR = base;
|
||||
try {
|
||||
const f = await seed(base, mode === "projectless");
|
||||
await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" });
|
||||
const input = { db, companyId: f.companyId, issueId: f.issueId, actorId: "board", actorSource: "local_implicit", observedRunId: f.sourceId };
|
||||
const receipt = await requestTaskRetry({ ...input, idempotencyKey: randomUUID() });
|
||||
const duplicate = await requestTaskRetry({ ...input, idempotencyKey: randomUUID() });
|
||||
expect(duplicate.requestId).toBe(receipt.requestId);
|
||||
const heartbeat = heartbeatService(db);
|
||||
await Promise.all([heartbeat.resumeTaskRetry(f.companyId, receipt.requestId!), heartbeat.resumeTaskRetry(f.companyId, receipt.requestId!)]);
|
||||
const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued")));
|
||||
const [state] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.requestId!));
|
||||
expect(successors, JSON.stringify(state.payload)).toHaveLength(1);
|
||||
expect(successors[0].contextSnapshot).toMatchObject({ previousRunId: f.sourceId, forceFreshSession: true, taskRetryRequestId: receipt.requestId });
|
||||
if (mode === "changed_binding") await db.update(issues).set({ executionWorkspaceId: null }).where(eq(issues.id, f.issueId));
|
||||
if (mode === "changed_cwd") await db.update(executionWorkspaces).set({ cwd: join(base, "another-directory") }).where(eq(executionWorkspaces.id, f.workspaceId));
|
||||
if (mode === "closed_workspace") await db.update(executionWorkspaces).set({ status: "closed" }).where(eq(executionWorkspaces.id, f.workspaceId));
|
||||
if (mode === "remote_workspace") await db.update(executionWorkspaces).set({ providerType: "plugin" }).where(eq(executionWorkspaces.id, f.workspaceId));
|
||||
if (mode === "pending_operation") await db.insert(workspaceOperations).values({ companyId: f.companyId,
|
||||
executionWorkspaceId: f.workspaceId, heartbeatRunId: f.ownerId, issueId: f.issueId, phase: "workspace_restore", status: "running" });
|
||||
if (mode === "active_lease") await db.insert(environmentLeases).values({ companyId: f.companyId,
|
||||
executionWorkspaceId: f.workspaceId, heartbeatRunId: f.ownerId, issueId: f.issueId, provider: "local", status: "active", leasePolicy: "ephemeral" });
|
||||
if (mode === "foreign_receipt") {
|
||||
const [proof] = await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, f.sourceId),
|
||||
eq(heartbeatRunEvents.eventType, "native.stopped_conversation_verified")));
|
||||
await db.update(heartbeatRunEvents).set({ payload: { ...proof.payload, successorRunId: randomUUID() } })
|
||||
.where(eq(heartbeatRunEvents.id, proof.id));
|
||||
expect(await readFreshWorkspaceReadiness(db, successors[0])).toMatchObject({ mode: "fresh_isolated_workspace" });
|
||||
} else if (mode === "projectless") {
|
||||
expect(await readFreshWorkspaceReadiness(db, successors[0])).toMatchObject({ mode: "fresh_isolated_workspace",
|
||||
retryIntentId: receipt.requestId, sourceRunId: f.sourceId });
|
||||
} else if (mode === "preserved") {
|
||||
expect(await readFreshWorkspaceReadiness(db, successors[0])).toMatchObject({ mode: "resume_stopped_native_workspace",
|
||||
retryIntentId: receipt.requestId, sourceRunId: f.sourceId, workspaceId: f.workspaceId, cwd: base });
|
||||
} else await expect(readFreshWorkspaceReadiness(db, successors[0])).rejects.toThrow(/Verified native continuation workspace/);
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
|
||||
const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.requestId!));
|
||||
expect(wake).toMatchObject({ status: "coalesced", runId: successors[0].id });
|
||||
const [owner] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.ownerId));
|
||||
expect(owner.resultJson).toEqual(f.resultJson);
|
||||
const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
|
||||
expect(action.evidence.automaticRecovery).toMatchObject({ actionOutcome: "unknown", replay: "explicit_user_continuation" });
|
||||
const proof = await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, f.sourceId), eq(heartbeatRunEvents.eventType, "native.stopped_conversation_verified")));
|
||||
expect(proof).toHaveLength(1);
|
||||
expect(proof[0].payload).toMatchObject({ runId: f.sourceId, ownerRunId: f.ownerId });
|
||||
for (const [path, value] of Object.entries(f.files)) expect(await readFile(join(f.root, path), "utf8")).toBe(JSON.stringify(value));
|
||||
expect(await readFile(join(base, "saved-work.txt"), "utf8")).toContain("Keep the prior result");
|
||||
} finally {
|
||||
if (previous === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; else process.env.PAPERCLIP_RUNNER_STATE_DIR = previous;
|
||||
await rm(base, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it.each(["provider_alive", "worker_alive", "owner_running", "controller_owned", "owner_remote", "owner_unreleased",
|
||||
"owner_new_launch", "source_new_launch", "foreign_host", "foreign_workspace", "foreign_cwd", "foreign_runner", "missing_provider_receipt",
|
||||
"pending_provider_launch", "changed_process_birth", "changed_journal", "normalized_receipt_identity", "provider_identity_conflict"])("keeps quarantine for %s", async mode => {
|
||||
const base = await mkdtemp(join(tmpdir(), "quarantined-retry-fence-"));
|
||||
const previous = process.env.PAPERCLIP_RUNNER_STATE_DIR;
|
||||
process.env.PAPERCLIP_RUNNER_STATE_DIR = base;
|
||||
try {
|
||||
const f = await seed(base);
|
||||
if (mode === "provider_alive" || mode === "pending_provider_launch") {
|
||||
await writeFile(join(f.root, "runner/codex-provider-state.json"), JSON.stringify({
|
||||
...f.files["runner/codex-provider-state.json"],
|
||||
...(mode === "provider_alive" ? { processId: process.pid } : { startupAttempt: { phase: "intent" } }),
|
||||
}));
|
||||
}
|
||||
if (mode === "worker_alive") {
|
||||
await db.update(heartbeatRuns).set({ processPid: process.pid, processGroupId: process.pid })
|
||||
.where(eq(heartbeatRuns.companyId, f.companyId));
|
||||
const [owner] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.ownerId));
|
||||
await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: f.ownerId, agentId: f.agentId,
|
||||
eventType: PROCESS_IDENTITY_RECORDED, payload: { targetKind: "local", hostname: hostname(),
|
||||
processPid: process.pid, processGroupId: process.pid, processStartedAt: owner.processStartedAt!.toISOString() } });
|
||||
}
|
||||
if (mode === "owner_running") await db.update(heartbeatRuns).set({ status: "running", finishedAt: null }).where(eq(heartbeatRuns.id, f.ownerId));
|
||||
if (mode === "controller_owned") await db.update(nativeRunFinalizations).set({ leaseOwner: "other-controller" }).where(eq(nativeRunFinalizations.runId, f.ownerId));
|
||||
if (mode === "owner_remote" || mode === "owner_unreleased") await db.update(environmentLeases).set(
|
||||
mode === "owner_remote" ? { provider: "daytona" } : { releasedAt: null },
|
||||
).where(eq(environmentLeases.heartbeatRunId, f.ownerId));
|
||||
if (mode === "owner_new_launch" || mode === "source_new_launch") await appendHeartbeatRunEvent(db, {
|
||||
companyId: f.companyId, runId: mode === "owner_new_launch" ? f.ownerId : f.sourceId, agentId: f.agentId, eventType: PROCESS_START_REQUESTED,
|
||||
});
|
||||
if (mode === "foreign_host") {
|
||||
const [identity] = await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, f.ownerId),
|
||||
eq(heartbeatRunEvents.eventType, PROCESS_IDENTITY_RECORDED)));
|
||||
await db.update(heartbeatRunEvents).set({ payload: { ...identity.payload, hostname: "another-host" } })
|
||||
.where(eq(heartbeatRunEvents.id, identity.id));
|
||||
}
|
||||
if (mode === "changed_process_birth") await db.update(heartbeatRuns).set({ processStartedAt: new Date() }).where(eq(heartbeatRuns.id, f.sourceId));
|
||||
if (mode === "foreign_runner") await db.update(heartbeatRuns).set({ runnerInstanceId: randomUUID() }).where(eq(heartbeatRuns.id, f.ownerId));
|
||||
if (mode === "foreign_workspace" || mode === "foreign_cwd") {
|
||||
const [owner] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.ownerId));
|
||||
const input = parseNativeExecutionInput(owner.runnerProfileJson!.nativeExecutionInput);
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { ...owner.runnerProfileJson,
|
||||
nativeExecutionInput: { ...input,
|
||||
...(mode === "foreign_workspace" ? { binding: { ...input.binding, executionWorkspaceId: "another-workspace" } }
|
||||
: { workspace: { ...input.workspace, cwd: join(base, "another-directory") } }),
|
||||
},
|
||||
} }).where(eq(heartbeatRuns.id, f.ownerId));
|
||||
}
|
||||
if (mode === "missing_provider_receipt") await db.delete(heartbeatRunEvents).where(and(
|
||||
eq(heartbeatRunEvents.runId, f.ownerId), eq(heartbeatRunEvents.eventType, "session.resumed")));
|
||||
if (mode === "normalized_receipt_identity") await db.update(heartbeatRunEvents).set({ sourceSeq: 2 }).where(and(
|
||||
eq(heartbeatRunEvents.runId, f.ownerId), eq(heartbeatRunEvents.eventType, "session.resumed")));
|
||||
if (mode === "provider_identity_conflict") {
|
||||
const [receipt] = await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, f.ownerId),
|
||||
eq(heartbeatRunEvents.eventType, "session.resumed")));
|
||||
const event = receipt.payload!.prpEvent as Record<string, unknown>;
|
||||
const changed = { ...event, payload: { providerSessionId: "another-thread", driverSessionId: "another-thread" } };
|
||||
await db.update(heartbeatRunEvents).set({ payload: { prpEvent: changed }, sourcePayloadSha256: nativeSha256(changed) })
|
||||
.where(eq(heartbeatRunEvents.id, receipt.id));
|
||||
}
|
||||
const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceId));
|
||||
const proof = await verifyStoppedNativeSessionForContinuation(db, source);
|
||||
if (mode === "changed_journal") {
|
||||
expect(proof).not.toBeNull();
|
||||
await writeFile(join(f.root, "runner/runner-state.json"), "{}");
|
||||
expect(proof!.retire()).toBe(false);
|
||||
} else expect(proof).toBeNull();
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull();
|
||||
} finally {
|
||||
if (previous === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; else process.env.PAPERCLIP_RUNNER_STATE_DIR = previous;
|
||||
await rm(base, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -195,6 +195,19 @@ test("split handoff commits relevant plans before execution and never creates ch
|
||||
})
|
||||
).status(),
|
||||
).toBe(422);
|
||||
await expect.poll(async () => {
|
||||
const comments = await json(await request.get(`/api/issues/${chat.id}/comments`));
|
||||
const completed = comments.filter((comment: any) => comment.authorAgentId === f.agent.id &&
|
||||
comment.body.startsWith("Delegated work completed:"));
|
||||
return tasks.every((task: any) => completed.some((comment: any) => comment.body.includes(task.identifier)));
|
||||
}, { timeout: 30_000 }).toBe(true);
|
||||
await expect.poll(async () => await json(await request.get(`/api/companies/${f.company.id}/live-runs`)),
|
||||
{ timeout: 30_000 }).toEqual([]);
|
||||
const settledTasks = await json(await request.get(`/api/companies/${f.company.id}/issues`));
|
||||
expect(settledTasks.map((task: any) => ({ id: task.id, status: task.status, parentId: task.parentId }))
|
||||
.sort((a: any, b: any) => a.id.localeCompare(b.id))).toEqual(tasks.map((task: any) => ({
|
||||
id: task.id, status: "done", parentId: null,
|
||||
})).sort((a: any, b: any) => a.id.localeCompare(b.id)));
|
||||
} finally {
|
||||
await f.restore();
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user