mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
fix(recovery): re-queue native runs that fail on a transient database fault before provider work (#20561)
## Thinking Path
> - Paperclip is the open source app people use to manage AI agents for
work.
> - Native (Paperclip Runner) runs prepare a task, pass a final dispatch
gate, and only then start a provider.
> - A native run can fail before any provider work starts. Examples are
a Postgres serialization or deadlock error (`40001`, `40P01`) at budget
admission, or a transient database error during setup.
> - Today such a run ends `adapter_failed` or `setup_failed`. `failRun`
writes the "no provider work" record only for legacy runs, and the
bounded retry paths do not retry these codes for native runs. The
release then opens the native hold
(`native_continuation_requires_reconciliation`), and a person must click
Retry.
> - The goal is that 0% of tasks need a Retry click after a restart or a
storage/DB fault. Duplicate provider work is worse than a Retry click,
so the fix must only act when the server can prove no provider work
started.
> - This pull request records that proof for native runs and sends a
transient database failure to the existing uncharged, bounded
`interrupted_preparation` retry lane.
> - The benefit is that these runs re-queue by themselves, and runs with
any sign of provider work keep today's hold.
## Linked Issues or Issue Description
**What happened?**
In restart-chaos waves with database pressure, a native run failed at
budget admission with exhausted serialization retries (`40001`/`40P01`).
The run event history had no `native.process_start_requested` event.
About 15 s later the server recorded "Automatic recovery stopped" with
cause `native_continuation_requires_reconciliation`. The task stayed
blocked until a person clicked Retry.
**Expected behavior**
A native run that failed before any provider work, on a fault that
clears by itself, re-queues automatically. The attempt is bounded and
does not spend the failure budget. A persistent failure still ends
failed, with a working Retry.
**Steps to reproduce**
1. Start a native (Paperclip Runner) task.
2. Make budget admission or a setup database write fail with SQLSTATE
`40001` or `40P01` until its bounded retries are exhausted.
3. See the run end `adapter_failed`/`setup_failed`, the task go to
`blocked`, and a `native_continuation_requires_reconciliation` recovery
action that waits for Retry.
**Paperclip version or commit**
`master` at 902f376796.
## What Changed
- `run-failure-cause.ts`: new durable record
`resultJson.nativePreDispatchFailure = { version: 1,
providerWorkStarted: false }`, and
`isNativePreDispatchTransientFailure`. The predicate needs a native
failed run, a generic code (`adapter_failed`/`setup_failed`), the
record, and a transient database `failureCause` (`40001`, `40P01`,
`55P03`, `57P01`-`57P03`, `53300`, `08xxx`, postgres.js connection
codes).
- `run-completion.ts`: `failRun` and `failRunSetup` write the record for
a native run only when all of these are true:
- the caller reports `nativeDispatchMayHaveStarted: false` (an omitted
value means "may have started");
- the run was not stopped during the failure, and has no terminal native
coordinator failure;
- the run's `native_run_finalizations` coordinator, if any, is still the
unclaimed row that runtime selection creates (`observed`, attempt 0, no
lease owner, no controller, no result). Any executor claim, in any
generation, disqualifies;
- the run has no recorded process (`processPid`, `processGroupId`,
`processStartedAt`), no `sessionIdAfter`, and no `native_run_results`
row. Process fields copied from a warm predecessor (marked by
`runnerProfileJson.inheritedProcessIdentityRunId`) are not this run's
own process and do not disqualify; the run's own start events are still
checked;
- the run's whole event history has no execution-start evidence. This
uses the canonical `hasNativeExecutionStartEvidence` helper from restart
recovery (`native.process_start_requested`,
`native.process_identity_recorded`, `adapter.invoke`,
`tool.execution.started`, or any runner event with a source event ID).
The helper is unchanged; the earlier inline copy matched it exactly and
was removed.
- `run-completion.ts`: a recorded transient failure closes the unclaimed
coordinator as `terminal_failure` (`native_pre_dispatch_failure`),
because `hasPendingRunCleanup` otherwise holds the successor on it. It
then calls the existing `scheduleBoundedRetryForRun` with `retryReason:
"interrupted_preparation"`. An answered-interaction continuation keeps
its own retry lane. The storage-outage lane is unchanged.
- `heartbeat.ts`: passes `nativeDispatchMayHaveStarted`. It is true once
native dispatch started, once the dispatch gate entered the executor
(also when the gate then fails), while native ownership is held, or for
a restart-recovery or native-lease resume.
- No change to the native hold in
`modules/wake-queue/adapters/postgres.ts`. The release already returns
early when a queued/scheduled successor exists, so a scheduled retry
never opens the hold. See Risks for why the hold stays for the exhausted
and no-successor cases.
## Verification
- `heartbeat-native-predispatch-retry.test.ts` (assembled heartbeat,
mock native backend): budget admission fails once with `40001`. The run
records the proof, one uncharged `interrupted_preparation` successor is
scheduled, no recovery action opens, and the promoted retry succeeds
once. A failure after the dispatch gate entered the executor gets no
record and no such successor.
- `run-completion.test.ts` (embedded Postgres):
- native pre-dispatch `40P01` in `failRun` writes the record and
schedules `interrupted_preparation`; the release suppresses immediate
recovery;
- native `setup_failed` on `40001` does the same;
- a run with `native.process_start_requested` in its history gets no
record and no retry;
- `nativeDispatchMayHaveStarted: true` and an omitted value (older
caller) get no record and no retry;
- a claimed native coordinator gets no record and no retry; an unclaimed
one is closed and the run re-queues;
- a run with a recorded process or provider session gets no record and
no retry;
- a non-transient cause gets the record but no retry.
- `heartbeat-retry-scheduling.test.ts`: the real scheduler chains two
`interrupted_preparation` retries for a recorded native failure. The
first is due at once. `failureRetries` stays 0. A spent lane returns
`retry_exhausted`, queues nothing, and the run stays `failed`.
- `wake-queue/adapters/postgres.test.ts`: with a scheduled successor,
the release opens no recovery action and the task stays `in_progress`.
Without one, today's hold opens (`blocked`,
`native_continuation_requires_reconciliation`, Retry available).
- `run-failure-cause.test.ts`: the predicate is false for a row without
the record (older server), for `providerWorkStarted: true`, for a
non-transient cause, a non-generic code, a legacy run, and a cancelled
run.
- Command: `cd server && npx vitest run
src/__tests__/heartbeat-native-predispatch-retry.test.ts
src/services/heartbeat/run-failure-cause.test.ts
src/services/heartbeat/run-completion.test.ts
src/modules/wake-queue/adapters/postgres.test.ts
src/__tests__/heartbeat-retry-scheduling.test.ts` (all new cases pass).
`tsc --noEmit` for `server` passes. `pnpm verify` selected 2451 files
because `heartbeat.ts` is a hub; I stopped it and rely on CI for the
broad suite.
- Live path unproven: I did not run a restart-chaos wave for this
change.
## Risks
- A wrong "no provider work" record could cause duplicate work. The
record needs several independent facts (no gate entry, no resume claim,
no claimed coordinator, no process, session or result, no
execution-start evidence in the whole history). Any read error keeps the
hold.
- Rows saved by an older server have no record. They keep today's hold
and manual Retry. Nothing is migrated. A run resumed after a deploy is a
restart-recovery or lease resume, so it gets no record.
- The normal path gains no I/O. The evidence reads run only on the
failure path, and only when dispatch was never entered.
- The native hold is not skipped when no successor exists. Skipping it
would send the run to the generic immediate-recovery tail, whose new run
does not carry this lane's counter. That could loop without a bound. So
a spent lane, and a crash between the failure write and the retry
scheduling, still end in today's hold with a working Retry.
- The lane is bounded by `INTERRUPTED_PREPARATION_RETRY_MAX_ATTEMPTS`
(12) with backoff after the first attempt, and it never revives a chain
whose failure budget is already spent.
> For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and
discuss it in `#dev` before opening the PR. Feature PRs that overlap
with planned core work may need to be redirected — check the roadmap
first. See `CONTRIBUTING.md`.
## Model Used
- Anthropic Claude, model ID `claude-opus-5-5`, used through Claude Code
with tool use (file edits, shell, test runs).
## 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
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge
Refs #20484, #20385.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
DottaandClaude authored and GitHub committed
1 parent
b5f62b3384
commit
7368a0b15c
8 files changed
+514
-10
No files matched your search
@@ -0,0 +1,167 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { agents, authUsers, companyMemberships, createDb, heartbeatRuns, issueComments, issueRecoveryActions, issues } from "@paperclipai/db";
|
||||
import type { NativeRunIdentity, NativeSession, NativeSessionBackend, PrpEvent } from "@paperclipai/paperclip-runner";
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { heartbeatService, type HeartbeatServiceOptions } from "../services/heartbeat.js";
|
||||
import { instanceSettingsService } from "../services/instance-settings.js";
|
||||
import { closeIdleWarmNativeSessionsForRestart } from "../services/native-runtime/native-session-executor.js";
|
||||
import { nativeRestartFixture, waitForCondition } from "./helpers/native-restart-fixture.js";
|
||||
import { getRecoveryPostgresSupport } from "./helpers/recovery-postgres-fixture.js";
|
||||
|
||||
// Budget admission is the last database write before the dispatch gate enters
|
||||
// the native executor. Exhausted serialization retries there fail the run with
|
||||
// no provider work started.
|
||||
const admissionFaults = vi.hoisted(() => ({ remaining: 0 }));
|
||||
vi.mock("../services/budget-reservations.js", async importOriginal => {
|
||||
const actual = await importOriginal<typeof import("../services/budget-reservations.js")>();
|
||||
const { AccountingContentionError } = await import("../services/accounting-transaction.js");
|
||||
return {
|
||||
...actual,
|
||||
reserveRunBudget: async (...args: Parameters<typeof actual.reserveRunBudget>) => {
|
||||
if (admissionFaults.remaining > 0) {
|
||||
admissionFaults.remaining -= 1;
|
||||
throw new AccountingContentionError(Object.assign(new Error("could not serialize access due to concurrent update"),
|
||||
{ name: "PostgresError", code: "40001" }), 4);
|
||||
}
|
||||
return actual.reserveRunBudget(...args);
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
const support = await getRecoveryPostgresSupport({});
|
||||
|
||||
function nativeBackend(db: ReturnType<typeof createDb>, options: { failDescriptorOnce?: { failed: boolean } } = {}):
|
||||
NonNullable<HeartbeatServiceOptions["nativeSessionBackendFactory"]> {
|
||||
return execution => {
|
||||
let identity: NativeRunIdentity;
|
||||
let sourceInstanceId = "";
|
||||
let turn = Promise.withResolvers<string>();
|
||||
const capabilities = { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true };
|
||||
const session: NativeSession = {
|
||||
identity: () => identity,
|
||||
capabilities: async () => capabilities,
|
||||
attachRun: async input => { identity = input.identity; },
|
||||
startTurn: async () => {
|
||||
const turnId = randomUUID();
|
||||
turn.resolve(turnId);
|
||||
return { turnId };
|
||||
},
|
||||
async *events() {
|
||||
const turnId = await turn.promise;
|
||||
const event = (sourceSeq: number, eventType: PrpEvent["eventType"], payload: Record<string, unknown>): PrpEvent => ({
|
||||
schema: "paperclip.prp.event.v1", sourceKind: "runner", sourceInstanceId,
|
||||
sourceEventId: `${sourceInstanceId}:${identity.runId}:${sourceSeq}`, sourceSeq,
|
||||
runId: identity.runId, normalizedSessionId: identity.sessionId, turnId,
|
||||
schemaVersion: 1, priority: 0, emittedAt: new Date().toISOString(), eventType, payload,
|
||||
});
|
||||
yield event(1, "turn.started", {});
|
||||
yield event(2, "item.completed", { kind: "agentMessage", channel: "final", text: "Done." });
|
||||
turn = Promise.withResolvers<string>();
|
||||
yield event(3, "turn.completed", { status: "completed" });
|
||||
},
|
||||
result: async () => null,
|
||||
snapshot: async () => ({ backendKind: "mock", sessionId: "predispatch-session", identity,
|
||||
providerSessionId: "predispatch-provider-session", cursor: null, activeTurnId: null, pendingRuntimeRequests: [], lineage: [] }),
|
||||
close: async () => {},
|
||||
};
|
||||
const backend: NativeSessionBackend = {
|
||||
descriptor: async () => {
|
||||
if (options.failDescriptorOnce && !options.failDescriptorOnce.failed) {
|
||||
options.failDescriptorOnce.failed = true;
|
||||
throw Object.assign(new Error("could not serialize access due to concurrent update"), { name: "PostgresError", code: "40001" });
|
||||
}
|
||||
const [run] = await db.select({ runnerInstanceId: heartbeatRuns.runnerInstanceId }).from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, execution.binding.runId));
|
||||
if (!run?.runnerInstanceId) throw new Error("Normal native admission did not persist a runner identity");
|
||||
sourceInstanceId = run.runnerInstanceId;
|
||||
return { kind: "mock", name: "native-predispatch-retry", version: "1", capabilities,
|
||||
runtimeContextCapabilities: { instructions: "native", skills: "native", mcp: "native" } };
|
||||
},
|
||||
openSession: async input => { identity = input.identity; return session; },
|
||||
recoverSession: async snapshot => { identity = snapshot.identity; return { recovered: true, session }; },
|
||||
};
|
||||
return backend;
|
||||
};
|
||||
}
|
||||
|
||||
describe.skipIf(!support.supported)("heartbeat native pre-dispatch failure retry", () => {
|
||||
const fixture = nativeRestartFixture({});
|
||||
|
||||
async function startTask(db: ReturnType<typeof createDb>, heartbeat: ReturnType<typeof heartbeatService>) {
|
||||
const { companyId, agentId } = fixture.current;
|
||||
const userId = randomUUID();
|
||||
await db.insert(authUsers).values({ id: userId, name: "Task owner", email: `${userId}@example.invalid`,
|
||||
createdAt: new Date(), updatedAt: new Date() });
|
||||
await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, membershipRole: "owner" });
|
||||
await db.update(agents).set({ status: "idle", adapterConfig: { provider: "codex" } }).where(eq(agents.id, agentId));
|
||||
await instanceSettingsService(db).updateExperimental({ enableAgentChat: true, enableNativeRunner: true });
|
||||
const [issue] = await db.insert(issues).values({ companyId, title: "Survive a database fault", status: "in_progress",
|
||||
assigneeAgentId: agentId, responsibleUserId: userId, conversationAgentId: agentId, conversationUserId: userId,
|
||||
conversationState: "active" }).returning();
|
||||
const [message] = await db.insert(issueComments).values({ companyId, issueId: issue!.id,
|
||||
authorType: "user", authorUserId: userId, body: "Start." }).returning();
|
||||
const run = await heartbeat.wakeup(agentId, { source: "on_demand", triggerDetail: "manual", reason: "issue_commented",
|
||||
requestedByActorType: "user", requestedByActorId: userId, manualUserWake: true,
|
||||
payload: { issueId: issue!.id, commentId: message!.id },
|
||||
contextSnapshot: { issueId: issue!.id, taskKey: issue!.id, wakeCommentId: message!.id } });
|
||||
expect(run).not.toBeNull();
|
||||
await waitForCondition("the first run to settle", () => heartbeat.getTaskDrainStatus().quiescent);
|
||||
return { issueId: issue!.id, runId: run!.id };
|
||||
}
|
||||
|
||||
it("re-queues a native run that failed at budget admission and completes it without a Retry", async () => {
|
||||
const { companyId } = fixture.current;
|
||||
const db = createDb(fixture.current.temporary.connectionString, { maxConnections: 4 });
|
||||
const heartbeat = heartbeatService(db, { nativeSessionBackendFactory: nativeBackend(db) });
|
||||
try {
|
||||
admissionFaults.remaining = 1;
|
||||
const { issueId, runId } = await startTask(db, heartbeat);
|
||||
const [failed] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
|
||||
expect(failed).toMatchObject({ status: "failed", runtimeMode: "native", errorCode: "adapter_failed",
|
||||
resultJson: expect.objectContaining({ failureCause: "40001",
|
||||
nativePreDispatchFailure: { version: 1, providerWorkStarted: false } }) });
|
||||
const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId),
|
||||
eq(heartbeatRuns.retryOfRunId, runId)));
|
||||
expect(successors).toEqual([expect.objectContaining({ scheduledRetryReason: "interrupted_preparation",
|
||||
scheduledRetryAttempt: 1 })]);
|
||||
expect(successors[0]!.contextSnapshot?.executionRetryAccounting).toMatchObject({ failureRetries: 0 });
|
||||
expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]?.status).not.toBe("blocked");
|
||||
expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId))).toHaveLength(0);
|
||||
|
||||
// The first attempt is due at once; the scheduler promotes it and the
|
||||
// queue starts it, exactly once.
|
||||
expect(await heartbeat.promoteDueScheduledRetries()).toMatchObject({ promoted: 1 });
|
||||
await heartbeat.resumeQueuedRuns();
|
||||
await waitForCondition("the retry to succeed", async () =>
|
||||
(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, successors[0]!.id)))[0]?.status === "succeeded");
|
||||
await waitForCondition("the retry to settle", () => heartbeat.getTaskDrainStatus().quiescent);
|
||||
expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId),
|
||||
eq(heartbeatRuns.retryOfRunId, runId)))).toHaveLength(1);
|
||||
expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId))).toHaveLength(0);
|
||||
} finally {
|
||||
admissionFaults.remaining = 0;
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
await closeIdleWarmNativeSessionsForRestart();
|
||||
await db.$client.end();
|
||||
}
|
||||
}, 90_000);
|
||||
|
||||
it("keeps the hold when the failure follows entry into the native executor", async () => {
|
||||
const { companyId } = fixture.current;
|
||||
const db = createDb(fixture.current.temporary.connectionString, { maxConnections: 4 });
|
||||
const heartbeat = heartbeatService(db, { nativeSessionBackendFactory: nativeBackend(db, { failDescriptorOnce: { failed: false } }) });
|
||||
try {
|
||||
const { runId } = await startTask(db, heartbeat);
|
||||
const [failed] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
|
||||
expect(failed?.status).toBe("failed");
|
||||
expect(failed?.resultJson ?? {}).not.toHaveProperty("nativePreDispatchFailure");
|
||||
expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId),
|
||||
eq(heartbeatRuns.retryOfRunId, runId), eq(heartbeatRuns.scheduledRetryReason, "interrupted_preparation")))).toHaveLength(0);
|
||||
} finally {
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
await closeIdleWarmNativeSessionsForRestart();
|
||||
await db.$client.end();
|
||||
}
|
||||
}, 90_000);
|
||||
});
|
||||
@@ -30,6 +30,7 @@ import {
|
||||
import { drainHeartbeatRunsToQuiescence } from "./helpers/drain-heartbeat-runs.js";
|
||||
import { registerServerAdapter, unregisterServerAdapter } from "../adapters/index.ts";
|
||||
import { createPostgresRunDispatchAdapter } from "../modules/run-dispatch/adapters/postgres.js";
|
||||
import { INTERRUPTED_PREPARATION_RETRY_MAX_ATTEMPTS } from "../services/execution-recovery-attempt.js";
|
||||
|
||||
const mockTelemetryClient = vi.hoisted(() => ({ track: vi.fn() }));
|
||||
const mockTrackAgentTaskRun = vi.hoisted(() => vi.fn());
|
||||
@@ -536,6 +537,40 @@ describeEmbeddedPostgres("heartbeat bounded retry scheduling", () => {
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId))).toHaveLength(3);
|
||||
});
|
||||
|
||||
it("re-queues a native pre-dispatch database failure without charging an attempt, until its lane is spent", async () => {
|
||||
const { agentId, issueId, runId, now } = await seedMaxTurnFixture();
|
||||
await db.update(agents).set({ adapterType: "paperclip_runner", adapterConfig: { provider: "codex", model: "gpt-6.1-sol" } })
|
||||
.where(eq(agents.id, agentId));
|
||||
const resultJson = { failureCause: "40001", nativePreDispatchFailure: { version: 1, providerWorkStarted: false } };
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "native", errorCode: "adapter_failed", resultJson })
|
||||
.where(eq(heartbeatRuns.id, runId));
|
||||
let predecessor = runId;
|
||||
for (const attempt of [1, 2]) {
|
||||
const outcome = await heartbeatService(db).scheduleBoundedRetry(predecessor,
|
||||
{ now, random: () => 0, retryReason: "interrupted_preparation" });
|
||||
expect(outcome.outcome).toBe("scheduled");
|
||||
const [child] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, predecessor));
|
||||
expect(child).toMatchObject({ status: "scheduled_retry", scheduledRetryReason: "interrupted_preparation",
|
||||
scheduledRetryAttempt: attempt });
|
||||
// Uncharged: the failure budget stays untouched; the first attempt is due at once.
|
||||
expect(child!.contextSnapshot?.executionRetryAccounting).toMatchObject({ failureRetries: 0, interruptedPreparations: attempt });
|
||||
if (attempt === 1) expect(child!.scheduledRetryAt?.getTime()).toBe(now.getTime());
|
||||
expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]?.executionRunId).toBe(child!.id);
|
||||
predecessor = child!.id;
|
||||
await db.update(heartbeatRuns).set({ status: "failed", runtimeMode: "native", errorCode: "adapter_failed", resultJson, finishedAt: now })
|
||||
.where(eq(heartbeatRuns.id, predecessor));
|
||||
}
|
||||
// A failure that keeps recurring ends visibly failed once the lane is spent.
|
||||
const [last] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, predecessor));
|
||||
await db.update(heartbeatRuns).set({ contextSnapshot: { ...last!.contextSnapshot, executionRetryAccounting: {
|
||||
version: 1, failureRetries: 0, maxTurnContinuations: 0, interruptedPreparations: INTERRUPTED_PREPARATION_RETRY_MAX_ATTEMPTS } } })
|
||||
.where(eq(heartbeatRuns.id, predecessor));
|
||||
expect(await heartbeatService(db).scheduleBoundedRetry(predecessor, { now, retryReason: "interrupted_preparation" }))
|
||||
.toMatchObject({ outcome: "retry_exhausted", maxAttempts: INTERRUPTED_PREPARATION_RETRY_MAX_ATTEMPTS });
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, predecessor))).toHaveLength(0);
|
||||
expect((await heartbeat.getRun(predecessor))?.status).toBe("failed");
|
||||
});
|
||||
|
||||
it("retains the reviewed chat admission on a durable retry without approving the task", async () => {
|
||||
const { companyId, agentId, issueId, runId, now } = await seedMaxTurnFixture({ issueStatus: "in_review" });
|
||||
const wakeCommentIds = [randomUUID(), randomUUID()];
|
||||
|
||||
@@ -868,6 +868,35 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => {
|
||||
expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0].statusVersion).toBe(blockedIssue.statusVersion);
|
||||
});
|
||||
|
||||
it.each([true, false])("holds a native pre-dispatch failure only when no automatic retry owns it (retry scheduled: %s)", async (retryScheduled) => {
|
||||
const companyId = await seedCompany();
|
||||
const agentId = await seedAgent({ companyId });
|
||||
const issueId = await seedIssue({ companyId, assigneeAgentId: agentId, status: "in_progress" });
|
||||
const runId = await seedRun({ companyId, agentId, status: "failed", contextSnapshot: { issueId } });
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "native", errorCode: "adapter_failed",
|
||||
resultJson: { failureCause: "40001", nativePreDispatchFailure: { version: 1, providerWorkStarted: false } } })
|
||||
.where(eq(heartbeatRuns.id, runId));
|
||||
await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId));
|
||||
if (retryScheduled) {
|
||||
await db.insert(heartbeatRuns).values({ companyId, agentId, invocationSource: "automation", status: "scheduled_retry",
|
||||
retryOfRunId: runId, scheduledRetryReason: "interrupted_preparation", scheduledRetryAttempt: 1,
|
||||
contextSnapshot: { issueId, retryReason: "interrupted_preparation" } });
|
||||
}
|
||||
const adapter = createPostgresWakeQueueAdapter(db, stubDeps);
|
||||
await adapter.withIssueExecutionLock({ companyId, runId, now: new Date() }, async () => undefined as never);
|
||||
const actions = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId));
|
||||
const issue = (await db.select().from(issues).where(eq(issues.id, issueId)))[0];
|
||||
if (retryScheduled) {
|
||||
// The uncharged retry owns the next attempt: no incident, no blocked task, no Retry click.
|
||||
expect(issue.status).toBe("in_progress");
|
||||
expect(actions).toHaveLength(0);
|
||||
} else {
|
||||
// An exhausted lane keeps today's hold, whose task offers a manual Retry.
|
||||
expect(issue.status).toBe("blocked");
|
||||
expect(actions[0]).toMatchObject({ cause: "native_continuation_requires_reconciliation" });
|
||||
}
|
||||
});
|
||||
|
||||
it("does not open a reconciliation incident for a native run whose model the provider rejected", async () => {
|
||||
const companyId = await seedCompany();
|
||||
const agentId = await seedAgent({ companyId });
|
||||
|
||||
@@ -2044,6 +2044,11 @@ export function heartbeatService(
|
||||
};
|
||||
let instructionWorkingCopy: ReturnType<typeof createHeartbeatInstructionWorkingCopy> | null = null;
|
||||
let nativeDispatchStarted = false;
|
||||
// Set once the dispatch gate enters the adapter or native executor. Unlike
|
||||
// nativeDispatchStarted it also covers a gate that fails after entering.
|
||||
let dispatchGateEntered = false;
|
||||
const nativeDispatchMayHaveStarted = () => nativeDispatchStarted || dispatchGateEntered || nativeOwnershipHeld
|
||||
|| Boolean(runOptions.nativeRestartRecovery || runOptions.nativeLeaseOwner);
|
||||
const leaveToNextController = async (error: unknown) => {
|
||||
if (!shutdownHandedOff() || legacyAdapterEntered || nativeDispatchStarted) return false;
|
||||
nativeSessionResumeScheduled = true;
|
||||
@@ -3004,6 +3009,7 @@ export function heartbeatService(
|
||||
try {
|
||||
return await dispatchResolvedInteractionContinuationAfterAdmission((markDispatchStarted) => {
|
||||
entered = true;
|
||||
dispatchGateEntered = true;
|
||||
return dispatch(markDispatchStarted);
|
||||
}, reservation);
|
||||
} catch (error) {
|
||||
@@ -3837,6 +3843,7 @@ export function heartbeatService(
|
||||
session: { runtimeForAdapter, previousSessionParams, taskKey, configuredModel, sessionConfigMetadata },
|
||||
readFailureReportSecrets,
|
||||
runId, err, identityRedactor, requiredWorkspaceRestoreEvidence, legacyAdapterEntered,
|
||||
nativeDispatchMayHaveStarted: nativeDispatchMayHaveStarted(),
|
||||
goalCheckpointSession, previousSessionDisplayId, taskSession,
|
||||
});
|
||||
}
|
||||
@@ -3893,6 +3900,7 @@ export function heartbeatService(
|
||||
await runCompletion.failRunSetup({
|
||||
run, runId, outerErr, signal: executionControl.controller.signal,
|
||||
providerDispatchStarted: legacyAdapterEntered || nativeDispatchStarted,
|
||||
nativeDispatchMayHaveStarted: nativeDispatchMayHaveStarted(),
|
||||
identityRedactor, readFailureReportSecrets,
|
||||
});
|
||||
legacyPreparationFailedBeforeDispatch = !legacyAdapterEntered && !nativeDispatchStarted;
|
||||
|
||||
@@ -1466,6 +1466,110 @@ describe.skipIf(!support.supported && !process.env.PAPERCLIP_TEST_DATABASE_URL)(
|
||||
expect(failureLog?.[0]).toMatchObject({ runId: f.run.id, errorCode: "adapter_failed", failureCause: "40001" });
|
||||
});
|
||||
|
||||
describe("a native run that fails before provider work", () => {
|
||||
const contention = () => new AccountingContentionError(Object.assign(new Error("deadlock detected"),
|
||||
{ name: "PostgresError", code: "40P01" }), 4);
|
||||
const nativeFixture = async () => {
|
||||
const f = await fixture({ runtimeMode: "native" }, `NPD${randomUUID().slice(0, 6)}`);
|
||||
f.deps.scheduleBoundedRetryForRun.mockResolvedValue({ outcome: "scheduled", run: f.run } as never);
|
||||
return { ...f, preDispatch: (err: unknown, nativeDispatchMayHaveStarted?: boolean) =>
|
||||
({ ...f.failure(err), legacyAdapterEntered: false, nativeDispatchMayHaveStarted }) };
|
||||
};
|
||||
|
||||
it("records that no provider work started and re-queues itself in the uncharged preparation lane", async () => {
|
||||
const f = await nativeFixture();
|
||||
await f.completion.failRun(f.preDispatch(contention(), false));
|
||||
const failed = await read(f.run.id);
|
||||
expect(failed).toMatchObject({ status: "failed", errorCode: "adapter_failed",
|
||||
resultJson: { failureCause: "40P01", nativePreDispatchFailure: { version: 1, providerWorkStarted: false } } });
|
||||
expect(f.deps.scheduleBoundedRetryForRun).toHaveBeenCalledWith(expect.objectContaining({ id: f.run.id }), f.agent,
|
||||
{ retryReason: "interrupted_preparation" });
|
||||
expect(f.deps.releaseIssueExecutionAndPromote).toHaveBeenCalledWith(expect.objectContaining({ id: f.run.id }),
|
||||
expect.objectContaining({ suppressImmediateRecovery: true }));
|
||||
});
|
||||
|
||||
it("re-queues a setup failure on a transient database fault", async () => {
|
||||
const f = await nativeFixture();
|
||||
await f.completion.failRunSetup({ ...f.setup(Object.assign(new Error("could not serialize access"),
|
||||
{ name: "PostgresError", code: "40001" })), nativeDispatchMayHaveStarted: false });
|
||||
expect(await read(f.run.id)).toMatchObject({ status: "failed", errorCode: "setup_failed",
|
||||
resultJson: { failureCause: "40001", nativePreDispatchFailure: { version: 1, providerWorkStarted: false } } });
|
||||
expect(f.deps.scheduleBoundedRetryForRun).toHaveBeenCalledWith(expect.objectContaining({ id: f.run.id }), f.agent,
|
||||
{ retryReason: "interrupted_preparation" });
|
||||
});
|
||||
|
||||
it("keeps the hold once the run's history shows a native process start", async () => {
|
||||
const f = await nativeFixture();
|
||||
await appendHeartbeatRunEvent(db, { companyId: f.run.companyId, runId: f.run.id, agentId: f.run.agentId,
|
||||
eventType: "native.process_start_requested", stream: "system", level: "info", message: "Native execution requested",
|
||||
payload: { providerAttempt: 1 } });
|
||||
await f.completion.failRun(f.preDispatch(contention(), false));
|
||||
const failed = await read(f.run.id);
|
||||
expect(failed.resultJson).not.toHaveProperty("nativePreDispatchFailure");
|
||||
expect(f.deps.scheduleBoundedRetryForRun).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it.each([
|
||||
["a dispatch that may have started", true],
|
||||
["a caller that does not report dispatch", undefined],
|
||||
] as const)("keeps the hold for %s", async (_label, mayHaveStarted) => {
|
||||
const f = await nativeFixture();
|
||||
await f.completion.failRun(f.preDispatch(contention(), mayHaveStarted));
|
||||
expect((await read(f.run.id)).resultJson).not.toHaveProperty("nativePreDispatchFailure");
|
||||
expect(f.deps.scheduleBoundedRetryForRun).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("keeps the hold when a native coordinator was claimed", async () => {
|
||||
const f = await nativeFixture();
|
||||
const [issue] = await db.insert(issues).values({ companyId: f.run.companyId, title: "Claimed native run" }).returning();
|
||||
await db.update(heartbeatRuns).set({ nativeIssueId: issue.id }).where(eq(heartbeatRuns.id, f.run.id));
|
||||
await db.insert(nativeRunFinalizations).values({ runId: f.run.id, companyId: f.run.companyId, issueId: issue.id, phase: "observed", attempt: 1, leaseOwner: "executor-lease" });
|
||||
await f.completion.failRun(f.preDispatch(contention(), false));
|
||||
expect((await read(f.run.id)).resultJson).not.toHaveProperty("nativePreDispatchFailure");
|
||||
expect(f.deps.scheduleBoundedRetryForRun).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("re-queues when runtime selection created the coordinator but no executor claimed it", async () => {
|
||||
const f = await nativeFixture();
|
||||
const [issue] = await db.insert(issues).values({ companyId: f.run.companyId, title: "Unclaimed native run" }).returning();
|
||||
await db.update(heartbeatRuns).set({ nativeIssueId: issue.id }).where(eq(heartbeatRuns.id, f.run.id));
|
||||
await db.insert(nativeRunFinalizations).values({ runId: f.run.id, companyId: f.run.companyId, issueId: issue.id, phase: "observed" });
|
||||
await f.completion.failRun(f.preDispatch(contention(), false));
|
||||
expect((await read(f.run.id)).resultJson).toMatchObject({ nativePreDispatchFailure: { version: 1, providerWorkStarted: false } });
|
||||
expect(f.deps.scheduleBoundedRetryForRun).toHaveBeenCalledOnce();
|
||||
// The unclaimed coordinator is closed so it cannot hold the successor as pending cleanup.
|
||||
expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.run.id)))[0])
|
||||
.toMatchObject({ phase: "terminal_failure", failureCode: "native_pre_dispatch_failure" });
|
||||
});
|
||||
|
||||
it.each([
|
||||
["a recorded process", { processPid: 4242 }],
|
||||
["a provider session", { sessionIdAfter: "provider-session" }],
|
||||
] as const)("keeps the hold when the run carries %s", async (_label, patch) => {
|
||||
const f = await nativeFixture();
|
||||
await db.update(heartbeatRuns).set(patch).where(eq(heartbeatRuns.id, f.run.id));
|
||||
await f.completion.failRun(f.preDispatch(contention(), false));
|
||||
expect((await read(f.run.id)).resultJson).not.toHaveProperty("nativePreDispatchFailure");
|
||||
expect(f.deps.scheduleBoundedRetryForRun).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("re-queues a follow-up whose process identity was inherited from its warm predecessor", async () => {
|
||||
const f = await nativeFixture();
|
||||
await db.update(heartbeatRuns).set({ processPid: 4242, runnerProfileJson: { inheritedProcessIdentityRunId: randomUUID() } })
|
||||
.where(eq(heartbeatRuns.id, f.run.id));
|
||||
await f.completion.failRun(f.preDispatch(contention(), false));
|
||||
expect((await read(f.run.id)).resultJson).toMatchObject({ nativePreDispatchFailure: { version: 1, providerWorkStarted: false } });
|
||||
expect(f.deps.scheduleBoundedRetryForRun).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("records the proof but does not re-queue a failure that is not a transient database fault", async () => {
|
||||
const f = await nativeFixture();
|
||||
await f.completion.failRun(f.preDispatch(new Error("adapter crashed"), false));
|
||||
expect((await read(f.run.id)).resultJson).toMatchObject({ nativePreDispatchFailure: { version: 1, providerWorkStarted: false } });
|
||||
expect(f.deps.scheduleBoundedRetryForRun).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
it("records a workspace provider's reason on the run and in the setup failure log without retrying", async () => {
|
||||
const f = await fixture();
|
||||
const logged = vi.spyOn(logger, "error");
|
||||
|
||||
@@ -1,7 +1,10 @@
|
||||
import { preserveAcceptedRunUsageProjectionSql } from "../usage-receipts.js";
|
||||
import { isContextCompactionRun } from "../context-compaction.js";
|
||||
import { providerFailureFromError } from "../../vendor/paperclip-runner/index.js";
|
||||
import { GENERIC_RUN_FAILURE_CODES, failureCauseFromAdapterResult, failureCauseFromError } from "./run-failure-cause.js";
|
||||
import { GENERIC_RUN_FAILURE_CODES, NATIVE_PRE_DISPATCH_FAILURE_CODE, NATIVE_PRE_DISPATCH_FAILURE_KEY, NATIVE_PRE_DISPATCH_FAILURE_RECORD, failureCauseFromAdapterResult, failureCauseFromError,
|
||||
isNativePreDispatchTransientFailure } from "./run-failure-cause.js";
|
||||
import { INTERRUPTED_PREPARATION_RETRY_REASON } from "../execution-recovery-attempt.js";
|
||||
import { hasNativeExecutionStartEvidence } from "../native-runtime/native-restart-recovery.js";
|
||||
import { withoutSqlStatementText } from "@paperclipai/shared";
|
||||
import { nativeTaskSessionPublicationHandled } from "../native-runtime/native-task-session-publication.js";
|
||||
import { boundOriginalMessage, describeRefusedExactResume, interruptedExactResumeCause } from "../native-runtime/native-exact-resume-failure.js";
|
||||
@@ -35,8 +38,8 @@ import { accountRunCost } from "../run-cost-accounting.js";
|
||||
import { isConversation, settleConversationTurn } from "../agent-conversations.js";
|
||||
import { legacyExecutionNeedsReconciliationWithEvidence } from "../legacy-execution-recovery.js";
|
||||
import { isWorkspaceGitScanError } from "../workspace-git-operation-scheduler.js";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { heartbeatRuns, issues, nativeRunFinalizations } from "@paperclipai/db";
|
||||
import { and, eq, isNull, sql } from "drizzle-orm";
|
||||
import { heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults } from "@paperclipai/db";
|
||||
import { HttpError } from "../../errors.js";
|
||||
import { logger } from "../../middleware/logger.js";
|
||||
import { normalizeResponsibleUserDenialCode } from "../responsible-user-denial-run-outcomes.js";
|
||||
@@ -152,6 +155,12 @@ export interface FailHeartbeatRunInput extends RunCompletionInput {
|
||||
identityRedactor: Pick<ReturnType<typeof createAgentIdentityRedactor>, "redact">;
|
||||
requiredWorkspaceRestoreEvidence: Record<string, unknown> | null;
|
||||
legacyAdapterEntered: boolean;
|
||||
/**
|
||||
* False only when this controller never entered the native dispatch gate and
|
||||
* is not resuming a restart claim. Omitted means provider work may have
|
||||
* started, so the failure keeps the native hold.
|
||||
*/
|
||||
nativeDispatchMayHaveStarted?: boolean;
|
||||
goalCheckpointSession: { current: { params: Record<string, unknown>; displayId: string } | null };
|
||||
previousSessionDisplayId: string | null;
|
||||
taskSession: TaskSession;
|
||||
@@ -164,6 +173,8 @@ export interface FailHeartbeatRunSetupInput {
|
||||
signal: AbortSignal;
|
||||
/** True once a legacy adapter or native dispatch began provider work. */
|
||||
providerDispatchStarted: boolean;
|
||||
/** See FailHeartbeatRunInput.nativeDispatchMayHaveStarted. */
|
||||
nativeDispatchMayHaveStarted?: boolean;
|
||||
identityRedactor: Pick<ReturnType<typeof createAgentIdentityRedactor>, "redact">;
|
||||
readFailureReportSecrets: () => string[];
|
||||
}
|
||||
@@ -1098,9 +1109,89 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC
|
||||
await settleAgent();
|
||||
}
|
||||
|
||||
/**
|
||||
* The pre-dispatch record for a native run that failed before provider work
|
||||
* could start, else nothing. Only a failure path calls it, so the normal
|
||||
* path gains no I/O. Missing or unreadable evidence keeps the native hold.
|
||||
*/
|
||||
async function nativePreDispatchFailureResult(runId: string, dispatchMayHaveStarted: boolean | undefined) {
|
||||
if (dispatchMayHaveStarted !== false) return null;
|
||||
try {
|
||||
const [row] = await db.select({ companyId: heartbeatRuns.companyId, runtimeMode: heartbeatRuns.runtimeMode,
|
||||
processPid: heartbeatRuns.processPid, processGroupId: heartbeatRuns.processGroupId,
|
||||
processStartedAt: heartbeatRuns.processStartedAt, sessionIdAfter: heartbeatRuns.sessionIdAfter,
|
||||
inheritedProcessIdentityRunId: sql<unknown>`${heartbeatRuns.runnerProfileJson}->>'inheritedProcessIdentityRunId'`,
|
||||
coordinator: { runId: nativeRunFinalizations.runId, phase: nativeRunFinalizations.phase,
|
||||
attempt: nativeRunFinalizations.attempt, leaseOwner: nativeRunFinalizations.leaseOwner,
|
||||
controllerBootId: nativeRunFinalizations.controllerBootId, resultId: nativeRunFinalizations.resultId } })
|
||||
.from(heartbeatRuns)
|
||||
.leftJoin(nativeRunFinalizations, and(eq(nativeRunFinalizations.runId, heartbeatRuns.id),
|
||||
eq(nativeRunFinalizations.companyId, heartbeatRuns.companyId)))
|
||||
.where(eq(heartbeatRuns.id, runId)).limit(1);
|
||||
if (!row || row.runtimeMode !== "native") return null;
|
||||
// A recorded process or provider session also means work may have run.
|
||||
// Runtime selection copies a warm predecessor's process identity onto a
|
||||
// follow-up and names that run; such an inherited identity is not this
|
||||
// run's process. Its own process would still record its start events.
|
||||
const inheritedProcessIdentity = typeof row.inheritedProcessIdentityRunId === "string";
|
||||
if ((!inheritedProcessIdentity && (row.processPid !== null || row.processGroupId !== null
|
||||
|| row.processStartedAt !== null)) || row.sessionIdAfter !== null) return null;
|
||||
// Runtime selection creates the coordinator unclaimed ("observed",
|
||||
// attempt 0). Only the executor, after entering dispatch, claims it with
|
||||
// a lease owner and controller; any claim, in any generation, disqualifies.
|
||||
const coordinator = row.coordinator;
|
||||
if (coordinator?.runId && (coordinator.phase !== "observed" || coordinator.attempt !== 0
|
||||
|| coordinator.leaseOwner !== null || coordinator.controllerBootId !== null || coordinator.resultId !== null)) return null;
|
||||
// The executor records native.process_start_requested before any backend
|
||||
// may spawn a runner; runner events carry a source event ID. Any of them,
|
||||
// in this or an earlier controller generation, means work may have run.
|
||||
// Restart recovery uses the same canonical proof.
|
||||
if (await hasNativeExecutionStartEvidence(db, { companyId: row.companyId, id: runId })) return null;
|
||||
const [result] = await db.select({ id: nativeRunResults.id }).from(nativeRunResults)
|
||||
.where(and(eq(nativeRunResults.companyId, row.companyId), eq(nativeRunResults.runId, runId))).limit(1);
|
||||
if (result) return null;
|
||||
return { [NATIVE_PRE_DISPATCH_FAILURE_KEY]: NATIVE_PRE_DISPATCH_FAILURE_RECORD };
|
||||
} catch (error) {
|
||||
logger.warn({ err: error, runId }, "could not prove a native failure started no provider work; keeping the native hold");
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Runtime selection leaves an unclaimed coordinator ("observed", attempt 0).
|
||||
* It counts as pending cleanup and would hold the re-queued successor, so a
|
||||
* failure proven to start no provider work closes it as a terminal failure.
|
||||
* The guard matches only an unclaimed row; a claimed one keeps its owner.
|
||||
*/
|
||||
async function settleUnclaimedNativeCoordinator(run: Pick<Run, "id" | "companyId">) {
|
||||
await db.update(nativeRunFinalizations).set({ phase: "terminal_failure", failureCode: NATIVE_PRE_DISPATCH_FAILURE_CODE,
|
||||
updatedAt: new Date() }).where(and(eq(nativeRunFinalizations.companyId, run.companyId),
|
||||
eq(nativeRunFinalizations.runId, run.id), eq(nativeRunFinalizations.phase, "observed"), eq(nativeRunFinalizations.attempt, 0),
|
||||
isNull(nativeRunFinalizations.leaseOwner), isNull(nativeRunFinalizations.controllerBootId), isNull(nativeRunFinalizations.resultId)));
|
||||
}
|
||||
|
||||
/**
|
||||
* A native run that failed on a transient database fault before provider
|
||||
* work re-queues itself in the uncharged, bounded interrupted-preparation
|
||||
* lane. An answered interaction keeps its own continuation retry.
|
||||
*/
|
||||
async function scheduleNativePreDispatchRetry(run: Run, agent: Agent) {
|
||||
if (!isNativePreDispatchTransientFailure(run) || isResolvedInteractionContinuationWakeContext(run.contextSnapshot)) return null;
|
||||
try {
|
||||
await settleUnclaimedNativeCoordinator(run);
|
||||
} catch (error) {
|
||||
logger.warn({ err: error, runId: run.id }, "could not close the unclaimed native coordinator; keeping the native hold");
|
||||
return null;
|
||||
}
|
||||
return scheduleBoundedRetryForRun(run, agent, { retryReason: INTERRUPTED_PREPARATION_RETRY_REASON }).catch((retryError) => {
|
||||
logger.warn({ err: retryError, runId: run.id }, "failed to re-queue a native run that failed before provider work");
|
||||
return null;
|
||||
});
|
||||
}
|
||||
|
||||
async function failRun(input: FailHeartbeatRunInput) {
|
||||
const { run, agent, issueId, issueRef, signal, output, session, readFailureReportSecrets, runId, err, identityRedactor, requiredWorkspaceRestoreEvidence,
|
||||
legacyAdapterEntered, goalCheckpointSession, previousSessionDisplayId, taskSession } = input;
|
||||
legacyAdapterEntered, nativeDispatchMayHaveStarted, goalCheckpointSession, previousSessionDisplayId, taskSession } = input;
|
||||
const { runtimeForAdapter, previousSessionParams, taskKey, configuredModel, sessionConfigMetadata } = session;
|
||||
const { handle, outputProgressState, flushOutputProgress } = output;
|
||||
// A process adapter may throw while its owned Stop is joining the
|
||||
@@ -1208,6 +1299,8 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC
|
||||
const stoppedDuringFailure = signal.aborted;
|
||||
const stopSnapshot = stoppedDuringFailure ? await getRun(run.id) : null;
|
||||
const failureOutcome = stoppedDuringFailure ? "cancelled" : "failed";
|
||||
const nativePreDispatchFailure = !stoppedDuringFailure && nativeTerminalFailureCode === null
|
||||
? await nativePreDispatchFailureResult(run.id, nativeDispatchMayHaveStarted) : null;
|
||||
const failedRunWrite = await measureRunPhase("completion.terminal_write", () => setRunStatusIfRunning(run.id, failureOutcome, {
|
||||
error: message,
|
||||
errorCode: stopSnapshot?.errorCode ?? failureErrorCode,
|
||||
@@ -1231,6 +1324,7 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC
|
||||
configurationIncompleteFailure?.resultJson ??
|
||||
{}),
|
||||
...(stoppedDuringFailure ? controllerLeaseLossResultJson(signal) : {}),
|
||||
...nativePreDispatchFailure,
|
||||
...(!legacyAdapterEntered && run.runtimeMode !== "native"
|
||||
? {
|
||||
executionRecovery: {
|
||||
@@ -1318,7 +1412,8 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC
|
||||
})
|
||||
: null;
|
||||
if (storageWait?.outcome === "retry_exhausted") await markStorageWaitExhausted(livenessRun);
|
||||
if (storageWait?.outcome !== "scheduled" && storageWait?.outcome !== "retry_exhausted") {
|
||||
const nativePreDispatchRetry = storageWait === null ? await scheduleNativePreDispatchRetry(livenessRun, agent) : null;
|
||||
if (storageWait?.outcome !== "scheduled" && storageWait?.outcome !== "retry_exhausted" && nativePreDispatchRetry === null) {
|
||||
await scheduleInteractionContinuationInfrastructureRetryIfEligible(
|
||||
livenessRun,
|
||||
agent,
|
||||
@@ -1330,7 +1425,7 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC
|
||||
// terminal failure, generic issue recovery must not create a
|
||||
// replacement retryOfRunId chain for the same provider work.
|
||||
suppressImmediateRecovery: nativeTerminalFailureCode !== null || storageWait?.outcome === "scheduled"
|
||||
|| storageWait?.outcome === "retry_exhausted",
|
||||
|| storageWait?.outcome === "retry_exhausted" || nativePreDispatchRetry?.outcome === "scheduled",
|
||||
});
|
||||
await handleIssueReviewPathDisposition(livenessRun);
|
||||
|
||||
@@ -1392,7 +1487,7 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC
|
||||
}
|
||||
|
||||
async function failRunSetup(input: FailHeartbeatRunSetupInput) {
|
||||
const { run, runId, outerErr, signal, providerDispatchStarted, identityRedactor, readFailureReportSecrets } = input;
|
||||
const { run, runId, outerErr, signal, providerDispatchStarted, nativeDispatchMayHaveStarted, identityRedactor, readFailureReportSecrets } = input;
|
||||
// The lease watchdog stopped preparation because this host no longer owns
|
||||
// the run. That is host infrastructure, not a defect in the agent or setup.
|
||||
// A post-dispatch error can also land here (for example a failed usage
|
||||
@@ -1558,6 +1653,8 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC
|
||||
...(!providerDispatchStarted
|
||||
? { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }
|
||||
: {}),
|
||||
...(!providerDispatchStarted && !signal.aborted
|
||||
? await nativePreDispatchFailureResult(runId, nativeDispatchMayHaveStarted) : null),
|
||||
};
|
||||
const setupFailureWrite = await setRunStatusIfRunning(runId, "failed", {
|
||||
error: message,
|
||||
@@ -1612,6 +1709,7 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC
|
||||
failedRun,
|
||||
).catch(() => failedRun);
|
||||
let storageWaitScheduled = false;
|
||||
let nativePreDispatchRetryScheduled = false;
|
||||
const setupFailureIssueId = readNonEmptyString(
|
||||
parseObject(livenessRun.contextSnapshot).issueId,
|
||||
);
|
||||
@@ -1675,7 +1773,8 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC
|
||||
: livenessRun.errorCode === STORAGE_UNAVAILABLE_CODE
|
||||
? scheduleBoundedRetryForRun(livenessRun, failedAgent, STORAGE_UNAVAILABLE_RETRY)
|
||||
: scheduleBoundedRetryForRun(livenessRun, failedAgent)
|
||||
: scheduleInteractionContinuationInfrastructureRetryIfEligible(livenessRun, failedAgent)
|
||||
: scheduleNativePreDispatchRetry(livenessRun, failedAgent).then(nativeRetry => nativeRetry
|
||||
?? scheduleInteractionContinuationInfrastructureRetryIfEligible(livenessRun, failedAgent))
|
||||
).catch((retryError) => {
|
||||
logger.warn(
|
||||
{ err: retryError, runId: livenessRun.id },
|
||||
@@ -1687,10 +1786,11 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC
|
||||
if (livenessRun.errorCode === STORAGE_UNAVAILABLE_CODE && retry?.outcome === "retry_exhausted") await markStorageWaitExhausted(livenessRun);
|
||||
storageWaitScheduled = livenessRun.errorCode === STORAGE_UNAVAILABLE_CODE
|
||||
&& (retry?.outcome === "scheduled" || retry?.outcome === "retry_exhausted");
|
||||
nativePreDispatchRetryScheduled = retry?.outcome === "scheduled" && isNativePreDispatchTransientFailure(livenessRun);
|
||||
}
|
||||
}
|
||||
await releaseIssueExecutionAndPromote(livenessRun, {
|
||||
suppressImmediateRecovery: storageWaitScheduled ||
|
||||
suppressImmediateRecovery: storageWaitScheduled || nativePreDispatchRetryScheduled ||
|
||||
readNonEmptyString(
|
||||
parseObject(livenessRun.contextSnapshot).goalControlRequestId,
|
||||
) !== null ||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { normalizeRunFailureCause } from "@paperclipai/shared";
|
||||
import { failureCauseFromAdapterResult, failureCauseFromError } from "./run-failure-cause.js";
|
||||
import { failureCauseFromAdapterResult, failureCauseFromError, isNativePreDispatchTransientFailure } from "./run-failure-cause.js";
|
||||
|
||||
describe("run failure cause", () => {
|
||||
it("reads the first safe code from the error and at most three nested causes", () => {
|
||||
@@ -35,4 +35,20 @@ describe("run failure cause", () => {
|
||||
expect(failureCauseFromAdapterResult({ ...base, errorMessage: "select * from secrets where id = $1" })).toBe("adapter_reported_error");
|
||||
expect(failureCauseFromAdapterResult(base)).toBe("unknown_error");
|
||||
});
|
||||
|
||||
it("re-queues only a recorded native pre-dispatch failure with a transient database cause", () => {
|
||||
const record = { version: 1, providerWorkStarted: false };
|
||||
const run = { runtimeMode: "native", status: "failed", errorCode: "adapter_failed",
|
||||
resultJson: { failureCause: "40001", nativePreDispatchFailure: record } };
|
||||
expect(isNativePreDispatchTransientFailure(run)).toBe(true);
|
||||
expect(isNativePreDispatchTransientFailure({ ...run, errorCode: "setup_failed", resultJson: { failureCause: "57P01", nativePreDispatchFailure: record } })).toBe(true);
|
||||
// A row saved by an older server has no record and keeps the native hold.
|
||||
expect(isNativePreDispatchTransientFailure({ ...run, resultJson: { failureCause: "40001" } })).toBe(false);
|
||||
expect(isNativePreDispatchTransientFailure({ ...run, resultJson: { failureCause: "40001",
|
||||
nativePreDispatchFailure: { version: 1, providerWorkStarted: true } } })).toBe(false);
|
||||
expect(isNativePreDispatchTransientFailure({ ...run, resultJson: { failureCause: "Error", nativePreDispatchFailure: record } })).toBe(false);
|
||||
expect(isNativePreDispatchTransientFailure({ ...run, errorCode: "native_session_interrupted" })).toBe(false);
|
||||
expect(isNativePreDispatchTransientFailure({ ...run, runtimeMode: "legacy" })).toBe(false);
|
||||
expect(isNativePreDispatchTransientFailure({ ...run, status: "cancelled" })).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -46,3 +46,48 @@ export function failureCauseFromAdapterResult(
|
||||
if (result.errorMessage) return "adapter_reported_error";
|
||||
return UNKNOWN_RUN_FAILURE_CAUSE;
|
||||
}
|
||||
|
||||
/**
|
||||
* Durable record on a failed native run that no provider work could have
|
||||
* started: the controller never entered the dispatch gate, the run was not a
|
||||
* resumed restart claim, and the run's whole history has no native coordinator
|
||||
* and no execution-start evidence (`native.process_start_requested` among them).
|
||||
* Rows saved before this record existed have none and keep the native hold.
|
||||
*/
|
||||
export const NATIVE_PRE_DISPATCH_FAILURE_KEY = "nativePreDispatchFailure";
|
||||
/** Failure code on the unclaimed native coordinator that such a failure closes. */
|
||||
export const NATIVE_PRE_DISPATCH_FAILURE_CODE = "native_pre_dispatch_failure";
|
||||
export const NATIVE_PRE_DISPATCH_FAILURE_RECORD = Object.freeze({ version: 1, providerWorkStarted: false } as const);
|
||||
|
||||
export function hasNativePreDispatchFailureRecord(resultJson: unknown): boolean {
|
||||
if (typeof resultJson !== "object" || resultJson === null || Array.isArray(resultJson)) return false;
|
||||
const record = (resultJson as Record<string, unknown>)[NATIVE_PRE_DISPATCH_FAILURE_KEY];
|
||||
if (typeof record !== "object" || record === null) return false;
|
||||
const fields = record as Record<string, unknown>;
|
||||
return fields.version === 1 && fields.providerWorkStarted === false;
|
||||
}
|
||||
|
||||
/**
|
||||
* Database faults that clear on their own: serialization and deadlock aborts,
|
||||
* a lock timeout, an administrator or crash shutdown, too many connections and
|
||||
* a dropped connection (SQLSTATEs and postgres.js codes). Storage outages keep
|
||||
* their own wait lane (`storage_unavailable`).
|
||||
*/
|
||||
const TRANSIENT_PRE_DISPATCH_FAILURE_CAUSES: ReadonlySet<string> = new Set([
|
||||
"40001", "40P01", "55P03", "57P01", "57P02", "57P03", "53300", "08000", "08001", "08003", "08006",
|
||||
"CONNECTION_CLOSED", "CONNECTION_ENDED", "CONNECTION_DESTROYED", "CONNECT_TIMEOUT",
|
||||
]);
|
||||
|
||||
/**
|
||||
* A failed native run that may re-queue itself without a person: it carries the
|
||||
* pre-dispatch record, a generic failure code and a transient database cause.
|
||||
* Anything else keeps the native hold and its manual Retry.
|
||||
*/
|
||||
export function isNativePreDispatchTransientFailure(run: {
|
||||
runtimeMode: string | null; status: string; errorCode: string | null; resultJson: unknown;
|
||||
}): boolean {
|
||||
if (run.runtimeMode !== "native" || run.status !== "failed" || !run.errorCode
|
||||
|| !GENERIC_RUN_FAILURE_CODES.has(run.errorCode) || !hasNativePreDispatchFailureRecord(run.resultJson)) return false;
|
||||
const cause = normalizeRunFailureCause((run.resultJson as Record<string, unknown>).failureCause);
|
||||
return cause !== null && TRANSIENT_PRE_DISPATCH_FAILURE_CAUSES.has(cause);
|
||||
}
|
||||
Reference in new issue
Block a user