mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-28 12:04:41 +01:00
## Thinking Path
> - Paperclip is the open source app people use to manage AI agents for
work
> - Each heartbeat run stores the process identity of its runner in
`heartbeat_runs.process_pid`, `process_group_id` and
`process_started_at`
> - Only cancellations (`release_run_resources`), restart recovery and
heartbeat recovery cleared these fields
> - Failed native runs take the `schedule_retry` path, so they kept a
PID after the run finished, for weeks
> - Liveness checks for old runs then depend on PID-reuse defenses, and
code that reads a set PID as "running" is misled
> - This pull request releases the process identity of settled native
runs when the process is confirmed gone, and backfills old sandbox rows
> - The benefit is that a finished run no longer looks like it owns a
live process, while live, warm and retrying processes keep their
identity
## Linked Issues or Issue Description
Refs #17358. Bug report described in-PR (bug report template).
**What happened?**
On one instance, 16 of 23 old native workspace sync folders belonged to
failed or cancelled runs that finished more than a week earlier and
still had `process_pid` set. The PID is written when the run's process
starts (for remote targets it is the sandbox PID). The status arbiter
clears it only through `release_run_resources`, which only cancellations
emit. Failed native runs take `schedule_retry`, and other terminal paths
(finalization reconciler, restart recovery terminal failures, watchdog
timeouts) do not clear it either.
**Expected behavior**
When a run is terminal and its process is confirmed gone (or no longer
owned), the process identity is cleared. A process that may still be
alive keeps its identity.
**Steps to reproduce**
1. Run a native task whose provider turn fails, so the arbiter decides
`run_failed_partial_evidence_preserved` with a `schedule_retry` effect.
2. Wait for the run to finish and the runner to exit.
3. Read the run row: `status = 'failed'` and `process_pid` /
`process_group_id` / `process_started_at` are still set.
**Paperclip version or commit:** `master` at 75fcb0b3fc.
**Deployment mode:** any.
**Agent adapter(s) involved:** native runtime (`paperclip_runner`);
local and sandbox targets.
## What Changed
- `releaseSettledNativeRunProcess` (in `native-local-process-stop.ts`)
clears `process_pid`, `process_group_id` and `process_started_at` of a
native run under a row lock, only when all of these are true:
- the status is `failed`, `cancelled`, `timed_out` or `interrupted`, and
`finished_at` is set;
- the finalization coordinator is absent, `committed` or
`terminal_failure` (a `retryable_failure` same-run retry keeps the
existing heartbeat recovery path; unfinished finalization keeps its
identity);
- the run has a session checkpoint or provider evidence, so native
session resume never mistakes it for an unused bootstrap;
- local target: every lease is released without failed cleanup, the
latest `native.process_identity_recorded` event names this host and the
same PID/group, and the PID and group probe absent. The release writes
the existing `native.local_process_stopped` receipt in the same
transaction, so `hasNativeLocalProcessStop`,
`readNativeLocalProcessStop`, explicit continuation and acknowledged
Stop keep their proof;
- sandbox target: every lease has a remote termination receipt (sandbox
PIDs are never probed on this host).
- Succeeded runs keep their identity: a warm runner can serve the next
turn, and chat retry matches warm predecessors by PID.
- The heartbeat executor calls the release after lease and scratch
cleanup at a terminal boundary (not when a same-run resume, workspace
finalization or held ownership is pending).
- Background work stays idle (see
`rules/sleep-aware-background-work.mdc`):
- Trigger: the executor settlement calls `settleNativeRunProcess`. If
the release is not ready yet (process still alive, local lease not
released, remote termination receipt missing), the run goes into an
in-memory pending set.
- Startup: the first `reapOrphanedRuns` pass after boot runs
`discoverSettledNativeRunProcesses` once, in pages of 50, over the
durable rows. This reconstructs work after a restart and covers terminal
transitions outside an executor (reconciler, restart recovery, watchdog,
timeouts).
- Retry: later passes call `retryPendingSettledNativeRunProcesses` only
for pending runs. It makes no database call when the set is empty. A run
leaves the set when it is released, becomes ineligible, or reaches 20
attempts.
- Runs that this process still executes are skipped.
- Stop receipts now also record `processStartedAt`.
`withReleasedNativeRunProcess` restores a released identity from the
latest stop receipt. Retained session cleanup
(`reconcileRetainedNativeSessionCleanup`) and warm transition restart
checks use it. Quiescent runnerd state recovery reads the stop receipt
when the prior run's identity was released. Without a receipt, all three
still treat the identity as unproven.
- Migration `0377_clear_settled_native_run_process_identity` backfills
sandbox runs (without bumping `updated_at`) with the same rules: native,
settled status, `finished_at` older than 1 day, settled coordinator
phase, checkpoint or provider evidence, and at least one lease where
every lease has a valid remote termination receipt. The snapshot is
carried forward as `0377_snapshot.json`.
- Not changed: legacy (non-native) runs. Case handoff release, scratch
cleanup and conversation continuation read the terminal row's local
PID/group as process-tree absence evidence, so clearing them needs
separate work. `release_run_resources` cancellation behavior and
`native_run_finalizations.controller_*` fields (they track the server
controller, not the run process) are unchanged.
### Migration margin
The 1-day margin is defense in depth. The real proof is the remote
termination receipt: the provider confirmed that the sandbox stopped or
was destroyed. One day is far longer than lease release, sandbox stop,
restart recovery takeover and the finalization reconciler sweep, so no
in-flight teardown or same-run recovery can still read these rows. Local
rows are not backfilled in SQL: only the host that recorded the process
can verify that it stopped and write the stop receipt. The runtime sweep
releases them on that host.
## Verification
- `pnpm exec vitest run
src/__tests__/native-settled-process-release.test.ts` (new, 20 tests):
release with a stop receipt and restore of the identity from it;
`cancelled`/`timed_out`/`interrupted`; live process kept, then released
after exit; kept for succeeded, `retryable_failure`,
`workspace_finalizing`, unreleased local lease, another host, no
evidence, unconfirmed sandbox, mixed leases; sandbox release with
receipts; startup discovery paging and skip; pending retry; retry cap;
no database access while nothing is pending; `reapOrphanedRuns`
discovery, then retry after the process exits; executor release for a
failed native run and no release for a succeeded run.
- `native-provider-failure.integration.test.ts`: a real `schedule_retry`
decision (`run_failed_partial_evidence_preserved`) leaves the failed run
with its PID after finalization, and the release then clears it and the
stop receipt reads back.
- `native-session-executor.test.ts`: quiescent recovery and retained
session cleanup accept a released identity with a stop receipt and
reject one without a receipt.
- `packages/db` `client.test.ts` migration rule: the backfill does not
set `updated_at`.
- `native-finalization-migration.test.ts`: migration 0377 clears only
old settled native sandbox rows with confirmed termination (15 fixtures,
including recent, succeeded, running, legacy, unsettled phases, no
evidence, unconfirmed or local leases, no lease).
- Related suites, all passing: `pnpm exec vitest run` on
heartbeat-process-recovery, native-workspace-sync-retention(-config),
native-finalization-{migration,settled-candidates,recovery},
chat-failed-run-retry-boundary, acknowledged-native-stop,
explicit-native-continuation, conversation-continuation,
native-session-resume, native-run-finalizer-telemetry,
native-session-executor, native-safe-replacement,
native-finalization-reconciler-timing, native-restart-recovery,
native-finalization-recovery-state, run-lifecycle,
agent-instruction/directory working copies, heartbeat-case-handoff (1920
tests across 23 files).
- `pnpm run typecheck` in `server/`: passes. `pnpm run check:migrations`
and the snapshot drift test in `packages/db`: pass.
- Mutation checks (each reverted alone; a test failed every time):
executor call, startup discovery, pending retry in `reapOrphanedRuns`,
retry cap, pending-set cleanup on retry, start time in the stop receipt,
retained cleanup receipt fallback, coordinator phase guard, succeeded
exclusion, liveness probe, host check, remote receipt check,
provider-evidence guard, local lease check, quiescent recovery receipt
fallback, migration margin, migration NULL-safe receipt check, migration
phase guard. The NULL-safe check came from a real bug the migration test
found. The warm transition restart call site has no direct test: that
path has no existing fixture. It uses the same tested helper.
- Not run: the full server suite and e2e tests.
## Risks
- Behavior change: more native runs now have a null `process_pid` after
they settle. Every native reader was checked: explicit continuation,
acknowledged Stop and agent stop use the stop receipt for null
identities; cancelled-startup detection also requires no process events,
so a released run is not mistaken for a cancelled startup; native
session resume selection is protected by the evidence guard; chat retry
reads its warm-predecessor PIDs from succeeded runs, which are not
released.
- A recycled local PID keeps a row unreleased (conservative), and
sandbox runs from providers without termination receipts keep their
identity.
- Background cost: one paged discovery at startup. After that, only
pending runs are checked, at most 20 times each. With nothing pending
there is no recurring query.
- A transition outside an executor that happens after startup, with no
executor settlement, is released at the next startup discovery.
- Migration: a data-only `UPDATE` on `heartbeat_runs` with a selective
`WHERE`; it passes the migration safety check. It is safe to re-run.
## Model Used
Claude Opus 5.5 (`claude-opus-5-5`), 1M context, through Claude Code
with tool use.
## 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
- [ ] I will address all Greptile and reviewer comments before
requesting merge
---------
Co-authored-by: Paperclip <noreply@paperclip.ing>
306 lines
29 KiB
TypeScript
306 lines
29 KiB
TypeScript
import { randomUUID } from "node:crypto";
|
|
import { eq } from "drizzle-orm";
|
|
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
|
import { agents, agentWakeupRequests, companies, completionContracts, createDb, heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, issueThreadInteractions, issues, nativeRunFinalizations, nativeRunResults, statusDecisions, workAssessments } from "@paperclipai/db";
|
|
import type { ControlPlanePort, NativeExecutionInputV1, PrpEvent, PrpTerminalState } from "@paperclipai/paperclip-runner";
|
|
import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
|
|
import { CONTROL_PLANE_CONFORMANCE_RESULT } from "../../vendor/paperclip-runner/testing.js";
|
|
|
|
const provider = vi.hoisted(() => ({ execute: vi.fn() }));
|
|
vi.mock("../../vendor/paperclip-runner/index.js", async (importOriginal) => ({
|
|
...(await importOriginal<typeof import("../../vendor/paperclip-runner/index.js")>()),
|
|
executeNativeSession: provider.execute,
|
|
}));
|
|
|
|
import { executePaperclipNativeSession } from "./native-session-executor.js";
|
|
import { buildNativeCompletionContract } from "./completion-contracts.js";
|
|
import { PaperclipControlPlanePort } from "./paperclip-control-plane-port.js";
|
|
import { prepareNativeHeartbeatRun } from "./prepare-native-run.js";
|
|
import { NATIVE_MODEL_REJECTION_DIAGNOSTIC, NATIVE_MODEL_REJECTION_MESSAGE } from "./native-provider-failure.js";
|
|
import { collectRunFailureDiagnostics } from "../run-failure-diagnostics.js";
|
|
import { finalizeNativeRun } from "./native-run-finalizer.js";
|
|
import { readPersistedNativeFailureEvidence, readPersistedNativeModelRejection } from "./native-provider-failure-evidence.js";
|
|
import { heartbeatService } from "../heartbeat.js";
|
|
import { getNativeReviewAssignment, readNativeReviewAssignmentContext } from "./native-review-participant.js";
|
|
import { claimQueuedNativeReviewRun } from "./native-review-dispatch.js";
|
|
import { commitNativeStatusDecision, NativeStatusRaceError } from "./status-decision-committer.js";
|
|
import { NATIVE_STATUS_ARBITER_POLICY_VERSION } from "./status-arbiter.js";
|
|
import { issueThreadInteractionService } from "../issue-thread-interactions.js";
|
|
import { createReleaseIssueExecution } from "../../modules/wake-queue/application/use-cases.js";
|
|
import { createPostgresWakeQueueAdapter } from "../../modules/wake-queue/adapters/postgres.js";
|
|
import { persistHeartbeatRunProcessMetadata } from "../heartbeat/run-lifecycle.js";
|
|
import { readNativeLocalProcessStop, releaseSettledNativeRunProcess } from "../native-local-process-stop.js";
|
|
|
|
const EXITED_PID = 4_194_304;
|
|
|
|
describe("native completed failure diagnostics", () => {
|
|
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
|
let db: ReturnType<typeof createDb>;
|
|
const companyId = randomUUID();
|
|
const agentId = randomUUID();
|
|
const workerId = randomUUID();
|
|
const model = "test-model";
|
|
|
|
beforeAll(async () => {
|
|
temporary = await startEmbeddedPostgresTestDatabase("native-model-failure-");
|
|
db = createDb(temporary.connectionString);
|
|
await db.insert(companies).values({ id: companyId, name: "Native failure regression", issuePrefix: "NMF" });
|
|
await db.insert(agents).values({ id: agentId, companyId, name: "Native failure agent", status: "active", adapterType: "paperclip_runner", adapterConfig: { provider: "codex", model } });
|
|
await db.insert(agents).values({ id: workerId, companyId, name: "Review source worker", status: "active", adapterType: "paperclip_runner" });
|
|
});
|
|
afterAll(async () => { await temporary?.cleanup(); });
|
|
|
|
it.each(["committed", "duplicate", "pending-review", "superseded-review", "resolved-review", "reassigned", "successor-race", "different-turn", "unknown-error", "opencode-lost-server", "untrusted-source", "untrusted-run", "untrusted-session", "succeeded", "cancelled"] as const)(
|
|
"preserves accepted results and only labels the matching provider rejection (%s)", async (scenario) => {
|
|
const issueId = randomUUID();
|
|
const runId = randomUUID();
|
|
const opencode = scenario === "opencode-lost-server";
|
|
const executionProvider = opencode ? { kind: "opencode" as const, model: `openrouter/${model}` } : { kind: "codex" as const, model };
|
|
const [issue] = await db.insert(issues).values({ id: issueId, companyId, title: "Explain a provider rejection", status: "in_progress", workMode: "standard", assigneeAgentId: agentId }).returning();
|
|
const [run] = await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", runtimeMode: "native", nativeIssueId: issueId, invocationSource: "assignment", triggerDetail: "system", contextSnapshot: { issueId } }).returning();
|
|
await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId));
|
|
const native = await prepareNativeHeartbeatRun({ db, run, issue, environmentLeaseId: randomUUID() });
|
|
const [contract] = await db.select().from(completionContracts).where(eq(completionContracts.issueId, issueId));
|
|
const completionInput = buildNativeCompletionContract(issue, { revision: contract.revision });
|
|
await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "observed" });
|
|
const execution: NativeExecutionInputV1 = {
|
|
schema: "paperclip.native-execution-input.v1",
|
|
binding: { companyId, runId, issueId, agentId, executionWorkspaceId: runId },
|
|
provider: executionProvider,
|
|
task: { identifier: issueId, title: issue.title, description: null, prompt: "Reply to the current message.", workMode: "standard" },
|
|
workspace: { cwd: "/tmp", repoUrl: null, repoRef: null, branchName: null },
|
|
session: { normalizedSessionId: native.normalizedSessionId, driverKind: opencode ? "opencode_server" : "codex_app_server", protocolVersion: 1, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } },
|
|
completionContract: { id: contract.id, sha256: contract.canonicalSha256, schemaVersion: "paperclip.completion-contract.v1", contract: completionInput },
|
|
interactionResponses: [], credentialBindings: [],
|
|
};
|
|
await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeExecutionInput: execution } }).where(eq(heartbeatRuns.id, runId));
|
|
if (scenario.endsWith("-review")) {
|
|
const sourceRunId = randomUUID(), sourceResultId = randomUUID(), assessmentId = randomUUID();
|
|
const decisionId = randomUUID(), interactionId = randomUUID();
|
|
await db.insert(heartbeatRuns).values({ id: sourceRunId, companyId, agentId: workerId, nativeIssueId: issueId, runtimeMode: "native", status: "succeeded", completionContractId: contract.id, completionContractSha256: contract.canonicalSha256 });
|
|
await db.insert(nativeRunResults).values({ id: sourceResultId, companyId, issueId, runId: sourceRunId, completionContractId: contract.id, serverFingerprint: sourceResultId, schemaStatus: "accepted", resultJson: {}, canonicalSha256: sourceResultId });
|
|
await db.insert(workAssessments).values({ id: assessmentId, companyId, issueId, runId: sourceRunId, contractId: contract.id, resultId: sourceResultId, triggerKind: "native_result", triggerActorCompanyId: companyId, priorIssueStatus: "in_progress", priorStatusVersion: 0, policyVersion: "fixture", assessmentJson: {}, inputDigest: assessmentId });
|
|
const [reviewIssue] = await db.update(issues).set({ status: "in_review", assigneeAgentId: workerId }).where(eq(issues.id, issueId)).returning();
|
|
await db.insert(statusDecisions).values({ id: decisionId, companyId, issueId, runId: sourceRunId, assessmentId, decisionVersion: 1, policyVersion: "fixture", fromStatus: "in_progress", toStatus: "in_review", reasonCode: "explicit_review", decisionJson: { projectedStatusVersion: reviewIssue.statusVersion }, decisionDigest: decisionId, applicationState: "applied" });
|
|
await db.update(issues).set({ lastStatusDecisionId: decisionId }).where(eq(issues.id, issueId));
|
|
await db.insert(issueThreadInteractions).values({ id: interactionId, companyId, issueId, kind: "request_confirmation", sourceRunId, addresseeAgentId: agentId, createdByAgentId: workerId,
|
|
continuationPolicy: "wake_assignee",
|
|
status: scenario === "resolved-review" ? "accepted" : "pending",
|
|
...(scenario === "resolved-review" ? { resolvedByRunId: runId, resolvedByAgentId: agentId } : {}),
|
|
payload: { version: 1, prompt: "Review the completed work.", target: { type: "custom", key: "native_completion_review", revisionId: decisionId } },
|
|
});
|
|
const contextSnapshot = { issueId, wakeReason: "native_completion_review", nativeReviewInteractionId: interactionId, nativeReviewDecisionId: decisionId };
|
|
await db.update(heartbeatRuns).set({ contextSnapshot }).where(eq(heartbeatRuns.id, runId));
|
|
if (scenario === "superseded-review") await db.update(issues).set({ lastStatusDecisionId: null }).where(eq(issues.id, issueId));
|
|
const review = await getNativeReviewAssignment(db, { companyId, issueId, agentId, contextSnapshot, allowResolvedByRunId: runId });
|
|
if (scenario === "superseded-review") expect(review).toBeNull();
|
|
else expect(review?.interaction.status).toBe(scenario === "resolved-review" ? "accepted" : "pending");
|
|
}
|
|
const event: PrpEvent = {
|
|
schema: "paperclip.prp.event.v1", runId, normalizedSessionId: native.normalizedSessionId,
|
|
turnId: native.turnId, sourceInstanceId: native.runnerInstanceId, sourceEventId: "model-terminal", sourceSeq: 1,
|
|
sourceKind: "runner", eventType: "turn.failed", schemaVersion: 1, priority: 0, emittedAt: new Date().toISOString(),
|
|
payload: opencode ? { status: "failed", providerTerminalObserved: false, error: {
|
|
code: "opencode_server_exited", message: "Private stream failure",
|
|
opencodeFailure: { kind: "server_exited", signal: "SIGKILL", connectionError: "ECONNREFUSED" },
|
|
} } : { status: "failed", error: { codexErrorInfo: "other", message: JSON.stringify({
|
|
type: "error", status: 400, error: { type: "invalid_request_error", message: scenario === "unknown-error" ? "Private unknown failure" : `The '${model}' model is not supported when using Codex with a ChatGPT account.` },
|
|
}) } },
|
|
};
|
|
const openInput = { identity: { companyId, issueId, runId, agentId, sessionId: native.normalizedSessionId }, backendKind: "mock" as const, sourceInstanceId: native.runnerInstanceId };
|
|
if (scenario === "duplicate") {
|
|
const prior = new PaperclipControlPlanePort(db, { companyId, issueId, runId, agentId, sessionId: native.normalizedSessionId, sourceInstanceId: native.runnerInstanceId, controlPlaneSourceInstanceId: "prior-controller", completionContractId: contract.id, completionContractSha256: contract.canonicalSha256 });
|
|
await prior.openRun(openInput);
|
|
expect((await prior.appendEvent(event)).disposition).toBe("committed");
|
|
}
|
|
const turnId = scenario === "different-turn" ? "later-turn" : native.turnId;
|
|
const terminal: PrpTerminalState = {
|
|
schema: "paperclip.prp.terminal.v1", turnTerminalState: scenario === "succeeded" ? "completed" : scenario === "cancelled" ? "cancelled" : "failed",
|
|
runTerminalState: scenario === "succeeded" ? "succeeded" : scenario === "cancelled" ? "cancelled" : "failed", reportedWorkDisposition: "blocked",
|
|
};
|
|
const result = { ...CONTROL_PLANE_CONFORMANCE_RESULT, reportedWorkDisposition: "blocked" as const,
|
|
blocker: { reasonCode: "provider_rejected", owner: { kind: "user" as const, name: "Task owner" }, unblockAction: "Select a compatible connection.", scope: "task_wide" as const },
|
|
completionClaim: { ...CONTROL_PLANE_CONFORMANCE_RESULT.completionClaim, contractRevision: completionInput.revision } };
|
|
provider.execute.mockReset().mockImplementation(async (options: { controlPlane: ControlPlanePort }) => {
|
|
await options.controlPlane.openRun(openInput);
|
|
if (scenario === "untrusted-source") {
|
|
await expect(options.controlPlane.appendEvent({ ...event, sourceInstanceId: "wrong-runner" })).rejects.toThrow("native_event_source_binding_mismatch");
|
|
} else if (scenario === "untrusted-run" || scenario === "untrusted-session") {
|
|
await expect(options.controlPlane.appendEvent({ ...event,
|
|
...(scenario === "untrusted-run" ? { runId: randomUUID() } : { normalizedSessionId: "wrong-session" }),
|
|
})).rejects.toThrow("native_event_binding_mismatch");
|
|
} else {
|
|
expect((await options.controlPlane.appendEvent(event)).disposition).toBe(scenario === "duplicate" ? "duplicate" : "committed");
|
|
}
|
|
await options.controlPlane.completeRun({ result, terminal, turnId });
|
|
return { result, terminal, turnId, normalizedSessionId: native.normalizedSessionId, providerSessionId: "fixture-provider-session", driverKind: execution.session.driverKind, driverVersion: "fixture", nativeEventCount: 1, highestContiguousSourceSeq: 1, usage: null };
|
|
});
|
|
const adapter = await executePaperclipNativeSession({ db, execution, runnerInstanceId: native.runnerInstanceId, backend: {
|
|
descriptor: async () => { throw new Error("Unexpected provider call"); },
|
|
openSession: async () => { throw new Error("Unexpected provider call"); },
|
|
} });
|
|
const classified = scenario === "committed" || scenario === "duplicate" || scenario.endsWith("-review") || scenario === "reassigned" || scenario === "successor-race";
|
|
const reviewAuthorityEnded = scenario === "superseded-review" || scenario === "resolved-review";
|
|
const ownerChanged = scenario === "reassigned" || scenario === "successor-race";
|
|
expect(adapter.errorCode).toBe(classified ? "native_provider_model_rejected" : undefined);
|
|
expect(adapter.errorMessage).toBe(classified ? NATIVE_MODEL_REJECTION_MESSAGE : scenario === "succeeded" ? null : `Native session ${terminal.runTerminalState}`);
|
|
expect(adapter.resultJson?.nativeProviderFailure).toEqual(classified ? NATIVE_MODEL_REJECTION_DIAGNOSTIC : undefined);
|
|
// Only a committed, unclassified failure of the terminal turn explains the generic message.
|
|
const turnFailure = scenario === "unknown-error" ? { provider: "codex", category: "turn_failed", codexErrorInfo: "other" }
|
|
: opencode ? { provider: "opencode", category: "turn_failed", runnerFailureCode: "opencode_server_exited", streamFailure: "server_exited", signal: "SIGKILL", connectionError: "ECONNREFUSED" }
|
|
: undefined;
|
|
expect(adapter.resultJson?.nativeTurnFailure).toEqual(turnFailure);
|
|
expect(adapter.nativeFinalization).toMatchObject({ result, terminal, turnId, workspaceFinalizeStatus: "pending" });
|
|
expect(provider.execute).toHaveBeenCalledTimes(1);
|
|
expect(execution.provider).toEqual(executionProvider);
|
|
const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId));
|
|
expect(coordinator).toMatchObject({ phase: "workspace_finalizing", leaseOwner: null, attempt: 1, nextAttemptAt: null });
|
|
expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runId))).toHaveLength(1);
|
|
expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId))).toHaveLength(0);
|
|
expect(await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, runId))).toHaveLength(scenario.startsWith("untrusted-") ? 0 : 1);
|
|
const diagnostics = collectRunFailureDiagnostics({ ...run, resultJson: adapter.resultJson ?? null }, {});
|
|
expect(diagnostics.provider).toEqual(classified ? NATIVE_MODEL_REJECTION_DIAGNOSTIC : turnFailure ?? {});
|
|
expect(JSON.stringify(diagnostics)).not.toContain(model);
|
|
expect(JSON.stringify(diagnostics)).not.toContain("Private");
|
|
const [persistedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
|
|
expect(persistedRun.errorCode).toBeNull(); // No adapter metadata has been saved yet.
|
|
expect(await readPersistedNativeModelRejection(db, persistedRun, turnId, terminal)).toEqual(classified ? {
|
|
errorCode: "native_provider_model_rejected", errorMessage: NATIVE_MODEL_REJECTION_MESSAGE, diagnostic: NATIVE_MODEL_REJECTION_DIAGNOSTIC,
|
|
} : null);
|
|
expect((await readPersistedNativeFailureEvidence(db, persistedRun, turnId, terminal)).turnFailure).toEqual(turnFailure ?? null);
|
|
if (classified) {
|
|
if (scenario === "committed") {
|
|
// Neither saved display diagnostics nor a different execution binding
|
|
// can confer the right to change the finalizer's retry decision.
|
|
for (const changed of [
|
|
{ ...persistedRun, runnerProfileJson: null },
|
|
{ ...persistedRun, nativeSessionId: "foreign-session" },
|
|
{ ...persistedRun, runnerInstanceId: "foreign-source" },
|
|
{ ...persistedRun, companyId: randomUUID() },
|
|
{ ...persistedRun, completionContractSha256: "foreign-contract" },
|
|
{ ...persistedRun, runnerProfileJson: { nativeExecutionInput: { ...execution, provider: { kind: "codex", model: "different-model" } } } },
|
|
{ ...persistedRun, runnerProfileJson: { nativeExecutionInput: { ...execution, binding: { ...execution.binding, runId: randomUUID() } } } },
|
|
]) expect(await readPersistedNativeModelRejection(db, changed, turnId, terminal)).toBeNull();
|
|
const [stored] = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, runId));
|
|
await db.update(heartbeatRunEvents).set({ sourcePayloadSha256: "invalid-hash" }).where(eq(heartbeatRunEvents.id, stored.id));
|
|
expect(await readPersistedNativeModelRejection(db, persistedRun, turnId, terminal)).toBeNull();
|
|
await db.update(heartbeatRunEvents).set({ sourcePayloadSha256: stored.sourcePayloadSha256 }).where(eq(heartbeatRunEvents.id, stored.id));
|
|
}
|
|
// Simulate a restart after native result acceptance, before heartbeat's
|
|
// adapter update. Finalization must use durable provider evidence.
|
|
const acceptedBefore = await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runId));
|
|
if (scenario === "reassigned") {
|
|
await db.update(issues).set({ assigneeAgentId: workerId, executionRunId: null }).where(eq(issues.id, issueId));
|
|
} else if (scenario === "successor-race") {
|
|
const [snapshot] = await db.select().from(issues).where(eq(issues.id, issueId));
|
|
const newerRunId = randomUUID(), raceAssessmentId = randomUUID();
|
|
await db.insert(heartbeatRuns).values({ id: newerRunId, companyId, agentId, status: "running", nativeIssueId: issueId });
|
|
await db.update(issues).set({ executionRunId: newerRunId }).where(eq(issues.id, issueId));
|
|
const [changed] = await db.select().from(issues).where(eq(issues.id, issueId));
|
|
expect(changed.statusVersion).toBe(snapshot.statusVersion);
|
|
await db.insert(workAssessments).values({ id: raceAssessmentId, companyId, issueId, runId, contractId: contract.id, resultId: acceptedBefore[0].id, triggerKind: "native_result", triggerActorCompanyId: companyId, priorIssueStatus: snapshot.status, priorStatusVersion: snapshot.statusVersion, policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, assessmentJson: {}, inputDigest: raceAssessmentId });
|
|
await expect(commitNativeStatusDecision({ db, companyId, issueId, runId, assessmentId: raceAssessmentId,
|
|
priorStatus: snapshot.status, priorStatusVersion: snapshot.statusVersion, priorDecisionId: snapshot.lastStatusDecisionId,
|
|
requireModelRejectionOwner: { agentId, reviewContext: null },
|
|
decision: { policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, statusAction: "blocked", toStatus: "blocked", reasonCode: "native_provider_model_rejected", unblockDescriptor: { owner: "board", action: NATIVE_MODEL_REJECTION_MESSAGE }, effects: [{ kind: "bind_blocker", owner: "board", action: NATIVE_MODEL_REJECTION_MESSAGE }] },
|
|
})).rejects.toBeInstanceOf(NativeStatusRaceError);
|
|
expect(await db.select().from(statusDecisions).where(eq(statusDecisions.runId, runId))).toHaveLength(0);
|
|
}
|
|
for (let retry = 0; retry < 2; retry++) {
|
|
await finalizeNativeRun({ db, runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true });
|
|
}
|
|
expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runId))).toEqual(acceptedBefore);
|
|
const [finalRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
|
|
expect(finalRun).toMatchObject({ status: "failed", errorCode: "native_provider_model_rejected", error: NATIVE_MODEL_REJECTION_MESSAGE });
|
|
expect(finalRun.resultJson?.nativeTurnFailure).toBeUndefined();
|
|
const [finalIssue] = await db.select().from(issues).where(eq(issues.id, issueId));
|
|
expect(finalIssue.status).toBe(scenario.endsWith("-review") ? "in_review" : ownerChanged ? "in_progress" : "blocked");
|
|
const decisions = await db.select().from(statusDecisions).where(eq(statusDecisions.runId, runId));
|
|
expect(decisions).toHaveLength(1);
|
|
expect(decisions[0].reasonCode).toBe(reviewAuthorityEnded ? "native_review_action_finished" : ownerChanged ? "run_failed_partial_evidence_preserved" : "native_provider_model_rejected");
|
|
const wakes = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, companyId));
|
|
expect(wakes.filter(wake => wake.payload?.issueId === issueId)).toHaveLength(ownerChanged ? 1 : 0);
|
|
if (!ownerChanged) expect(await heartbeatService(db).dispatchPendingNativeStatusWakeups({ companyId })).toEqual({ scanned: 0, dispatched: 0, recovered: 0, deferred: 0 });
|
|
expect(provider.execute).toHaveBeenCalledTimes(1);
|
|
if (scenario === "pending-review") {
|
|
const reviewContext = readNativeReviewAssignmentContext(finalRun.contextSnapshot)!;
|
|
expect(finalIssue).toMatchObject({ lastStatusDecisionId: reviewContext.nativeReviewDecisionId, executionRunId: null, checkoutRunId: null });
|
|
const originalReview = await getNativeReviewAssignment(db, { companyId, issueId, agentId, contextSnapshot: reviewContext });
|
|
expect(originalReview?.interaction.status).toBe("pending");
|
|
expect(finalIssue.statusVersion).toBe(originalReview!.sourceDecision.decisionJson.projectedStatusVersion);
|
|
expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId))).toHaveLength(0);
|
|
|
|
// Exercise the real release and periodic recovery lanes, rather than
|
|
// assuming that absence of a finalizer wake prevents later retries.
|
|
const unexpected = vi.fn(async () => { throw new Error("Unexpected automatic recovery"); });
|
|
const release = createReleaseIssueExecution({
|
|
issueLock: createPostgresWakeQueueAdapter(db, { resolveResponsibleUserId: unexpected, getRoutineEnv: unexpected, resolveSessionBeforeForWakeup: unexpected }),
|
|
recovery: { escalateStrandedAssignedIssue: unexpected, escalateStrandedRecoveryIssueInPlace: unexpected },
|
|
});
|
|
expect((await release({ companyId, runId, now: new Date() })).outcome.kind).toBe("released");
|
|
await heartbeatService(db).reconcileStrandedAssignedIssues();
|
|
expect(unexpected).not.toHaveBeenCalled();
|
|
expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId))).toHaveLength(0);
|
|
expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, companyId))).filter(wake => wake.payload?.issueId === issueId)).toHaveLength(0);
|
|
const [afterRecovery] = await db.select().from(issues).where(eq(issues.id, issueId));
|
|
expect(afterRecovery).toMatchObject({ status: "in_review", statusVersion: finalIssue.statusVersion, lastStatusDecisionId: finalIssue.lastStatusDecisionId });
|
|
|
|
// The operator repairs configuration and explicitly retries the
|
|
// assigned reviewer. No review rebinding or status reset is needed.
|
|
await db.update(agents).set({ adapterConfig: { provider: "codex", model: "repaired-model" }, runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } } }).where(eq(agents.id, agentId));
|
|
// Occupy the execution slot to exercise real wake admission without
|
|
// starting a provider. The normal atomic review claim follows below.
|
|
const occupiedRunId = randomUUID();
|
|
await db.insert(heartbeatRuns).values({ id: occupiedRunId, companyId, agentId, status: "running", startedAt: new Date() });
|
|
const retryRun = await heartbeatService(db, { runtimeEnv: {} }).wakeup(agentId, {
|
|
failedRunId: runId, source: "on_demand", triggerDetail: "manual", reason: "retry_failed_run",
|
|
requestedByActorType: "user", requestedByActorId: "review-repair-operator",
|
|
payload: { issueId }, contextSnapshot: { ...reviewContext },
|
|
});
|
|
expect(retryRun).toMatchObject({ status: "queued", retryOfRunId: runId, contextSnapshot: { issueId, ...reviewContext } });
|
|
const retryRunId = retryRun!.id;
|
|
await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date() }).where(eq(heartbeatRuns.id, occupiedRunId));
|
|
expect(await claimQueuedNativeReviewRun(db, { run: retryRun!, agentNameKey: "reviewer", claimedAt: new Date(), claimValues: { nativeIssueId: issueId, runtimeMode: "native" } }))
|
|
.toMatchObject({ id: retryRunId, status: "running" });
|
|
expect(await getNativeReviewAssignment(db, { companyId, issueId, agentId, contextSnapshot: reviewContext, actingRunId: retryRunId, issueExecutionRunId: retryRunId })).not.toBeNull();
|
|
await issueThreadInteractionService(db).acceptInteraction(
|
|
{ id: issueId, companyId, projectId: null, goalId: null, status: "in_review" },
|
|
reviewContext.nativeReviewInteractionId, {}, { agentId, runId: retryRunId },
|
|
);
|
|
const [resolved] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, reviewContext.nativeReviewInteractionId));
|
|
expect(resolved).toMatchObject({ status: "accepted", resolvedByAgentId: agentId, resolvedByRunId: retryRunId });
|
|
expect(await db.select().from(issues).where(eq(issues.id, issueId))).toMatchObject([{ status: "done", assigneeAgentId: workerId }]);
|
|
expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runId))).toEqual(acceptedBefore);
|
|
expect(provider.execute).toHaveBeenCalledTimes(1);
|
|
await db.update(agents).set({ adapterConfig: { provider: "codex", model } }).where(eq(agents.id, agentId));
|
|
}
|
|
} else if (scenario === "unknown-error" || opencode) {
|
|
// A lookalike diagnostic flag does not alter unknown failed-result policy.
|
|
await db.update(heartbeatRuns).set({ resultJson: { nativeProviderFailure: NATIVE_MODEL_REJECTION_DIAGNOSTIC } }).where(eq(heartbeatRuns.id, runId));
|
|
await persistHeartbeatRunProcessMetadata(db, runId, { pid: EXITED_PID, processGroupId: EXITED_PID, startedAt: new Date().toISOString(), targetKind: "local" });
|
|
await finalizeNativeRun({ db, runId, workspaceFinalizeStatus: "succeeded" });
|
|
const [decision] = await db.select().from(statusDecisions).where(eq(statusDecisions.runId, runId));
|
|
expect(decision.reasonCode).toBe("run_failed_partial_evidence_preserved");
|
|
// Recovery finalized before heartbeat saved the adapter result, so the
|
|
// turn failure must come from the persisted runner event.
|
|
const [finalRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
|
|
expect(finalRun.resultJson?.nativeTurnFailure).toEqual(turnFailure);
|
|
const wakes = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, companyId));
|
|
expect(wakes.filter(wake => wake.payload?.issueId === issueId)).toHaveLength(1);
|
|
if (scenario === "unknown-error") {
|
|
await finalizeNativeRun({ db, runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true });
|
|
const [settled] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
|
|
const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId));
|
|
expect(coordinator.phase).toBe("committed");
|
|
expect(settled).toMatchObject({ status: "failed", processPid: EXITED_PID, processGroupId: EXITED_PID });
|
|
await db.insert(heartbeatRunEvents).values({ companyId, runId, agentId, seq: 1_000, eventType: "harness.ready" });
|
|
expect(await releaseSettledNativeRunProcess(db, runId)).toBe("released");
|
|
const [released] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
|
|
expect(released).toMatchObject({ status: "failed", processPid: null, processGroupId: null, processStartedAt: null });
|
|
expect(await readNativeLocalProcessStop(db, companyId, runId)).toMatchObject({ processPid: EXITED_PID, processGroupId: EXITED_PID });
|
|
}
|
|
}
|
|
},
|
|
);
|
|
});
|