fix(recovery): preserve shutdown export ownership through finalizer drain (#16843)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Legacy adapters must save sandbox files before Paperclip releases
their source.
> - Graceful shutdown gave each adapter a shared 30-second settlement
wait.
> - When that wait ended, cleanup could stop the sandbox while the
adapter still exported files.
> - That race created a restore failure and blocked the next turn.
> - This pull request keeps export ownership through the finalizer drain
and renews its cleanup fence.
> - A successful late export can release its source and schedule one
bounded continuation.

## Linked Issues or Issue Description

Refs #15857. This corrects the timeout path introduced by the shutdown
settlement fence.
Refs #16478. That repair clears verified historical holds; this change
prevents an active export from losing its source.
Refs #16835. That merged fix handles remote native runner ownership.
This change handles legacy shutdown settlement.

## What Changed

- Keep a late adapter settlement in the shared execution finalizer set.
Wait for it before lease release, recovery classification, and retry
scheduling.
- Renew the exact owned settlement fence every 10 seconds. Use a
30-second lease, one renewal in flight, a one-second database lock
timeout, and a five-second statement timeout. Report bounded failure
categories without raw database errors or credentials.
- Refuse ordinary cleanup while the local adapter still owns settlement.
Preserve the explicit owner stop-and-retain path.
- Give the finalizer drain a further shared 30-second wait after the
initial shutdown drain.
- Document the ownership rule, the acceptance recipe, and the
controller-loss limit.

## Verification

- Red before the repair: the database-backed expired-wait regression
expected cleanup denial but received permission to stop the still-owned
source.
- Green on `615a4382c5`: all 526 tests in `recovery.test.ts`,
`legacy-workspace-restore-recovery.test.ts`, `shutdown.test.ts`, and
`heartbeat-process-recovery.test.ts` passed.
- The assembled heartbeat regression passed. It verifies no early
release or retry, joining the late owner, successful cleanup, one
durable scheduled continuation, and no duplicate after service reload.
- Safety coverage includes other-controller fencing, controller loss,
exact source retention, stop-only cancellation, completion winning the
status race, and real database lock contention during renewal.
- `pnpm -r typecheck`, `pnpm build`, and `node scripts/check-maps.mjs`
passed on the final source revision.
- A broad local `pnpm test:run` crossed the rebase and reported eight
companion supervision failures. It was stopped and its log retained. The
complete companion file then passed all 753 tests on current source.
This is not a passing full local suite; current-head CI covers the
complete repository checks.
- The 34 recovery tests include real lock-timeout classification and
credential-shaped error redaction.
- Current-head CI: 53 checks passed; two optional Storybook checks
skipped. Greptile is 5/5 on `615a4382c5`; no unresolved review threads.
- Live sandbox acceptance remains unverified. Three disposable
local-controller/Daytona attempts used legacy Codex ACP, `gpt-6-astra`,
API-key authentication, and the verified image
`ghcr.io/paperclipai/paperclip-daytona-runner@sha256:a2bd08cefb68d9217bdd75cf87a6b51f1705b18a4796eff3f97b2726a548b8fb`.
ACP initialization exited before model execution on macOS Node 26.4.0
and again on Node 24.21.0; one attempt exposed a local process-session
socket `ECONNRESET`. That does not establish the transport failure's
root cause. No attempt reached the delayed export or second turn, so
this PR remains draft pending faithful live acceptance. Probe setup
defects were retained separately from product failures, and all
test-owned sandboxes were cleaned up.

## Risks

- Shutdown can spend up to 30 more seconds joining finalizers. The
shipped Compose stop grace remains 90 seconds.
- Renewal adds at most one database operation per owned run every 10
seconds during shutdown. Lock and statement deadlines bound database
contention. The local adapter still protects its source if renewal
fails.
- Controller death or an export that outlives the whole process still
needs retained-source repair. This change does not reconstruct a legacy
export after process death or clear historical failed-restore holds.
- No schema or public API changes. Native restart ownership, approval
gates, and retry limits retain their existing behavior.
- This PR has not been deployed. It does not recover an existing
retained sandbox by itself.

## Model Used

OpenAI Codex, GPT-6 family, with reasoning, repository inspection, code
execution, and test tools. The exact backend model identifier and
context window were not exposed to this session.

## Checklist

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

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed 2026-10-19 09:19:09 -05:00
1 parent 0ed760ce9a
commit 016e40d532
8 files changed
+325 -98

No files matched your search

+19 -10
View File
@@ -1227,18 +1227,27 @@ Shutdown, process loss, and provider failure use the existing durable failure re
Graceful shutdown interrupts an owned legacy adapter through its cancellation
control, then waits within one shared 30-second window for provider stop and
adapter settlement across all affected runs. Before interruption becomes visible,
it commits a settlement fence on the exact sandbox lease. Cleanup on this server
or another server defers while that fence is pending. If the controller dies,
cleanup records exact stop-only retention and a repair hold after the deadline,
without claiming a successful restore. The shipped Compose services allow 90
seconds for settlement and subsequent shutdown cleanup.
adapter settlement across all affected runs. A run that exceeds that wait stays
in the server's finalizer drain, which has a further shared 30-second budget.
The wait ending does not prove the adapter or its file export stopped.
Before interruption becomes visible, shutdown commits a settlement fence on
the exact sandbox lease. The owning finalizer renews its 30-second deadline
every 10 seconds, with one renewal in flight per run. Renewal uses a one-second
database lock timeout and a five-second statement timeout. It cannot revive an
expired, settled, or foreign-owned fence. Cleanup on this server defers while
the adapter still owns settlement; another server reads the renewed fence.
If the controller dies, renewal stops. Cleanup records exact stop-only retention
and a repair hold after the deadline, without claiming a successful restore.
The shipped Compose services allow 90 seconds for settlement and subsequent
shutdown cleanup.
The sandbox lease remains available during workspace export and instruction
collection. Lease release and recovery classification follow settlement and use
the current execution evidence. If settlement exceeds the deadline with an
active sandbox source, shutdown records a restore failure and pins the exact
source for stop and retention before cleanup. A later run must pass the normal
workspace repair gate. Native runner ownership keeps its existing restart path.
the current execution evidence. A late successful export can finish cleanup
and queue one bounded continuation without a false workspace repair hold.
An actual restore failure, or loss of the controller before settlement, keeps
the exact source retained. A later run must pass the normal workspace repair
gate. This shutdown join does not reconstruct an interrupted legacy export
after process death. Native runner ownership keeps its existing restart path.
Real gates still apply: company and task ownership, active provider ownership, budget limits, agent availability, dependencies, pending approval/review paths, and explicit pause holds. Native runner reattachment and finalization retain their existing ownership protocol. Process, HTTP, and gateway adapters retain their recovery rules because invoking those adapters can itself repeat an external action rather than start a conversation turn.
+5
View File
@@ -24,6 +24,11 @@ Agent access: `available` — REST / native call_api; CLI issue diagnostics.
Policy: Task/run visibility, run ownership and source/action policy; workspace export
retry and some operator controls require board.
Graceful legacy shutdown retains the exact sandbox until its owned adapter
settles, including after the first shutdown wait. A renewed settlement fence
protects cleanup by other controllers. Loss of the owner retains the source
for repair after expiry. This automatic protection grants no agent operation,
execution permission, or bypass of workspace repair gates.
The server can recover a prepared remote bootstrap only with a durable
never-claimed executor receipt and no contradictory execution evidence. This
automatic classification grants no new agent action or permission; normal
+14
View File
@@ -211,6 +211,20 @@ repair, record the compared branches/revisions and preserve uncommitted work;
confirm the repair's actual result before resuming. Check the denied and stale
action paths. A changed card label is insufficient proof of workspace integrity.
For graceful legacy shutdown, interrupt a disposable sandbox run while file
export is in flight. Delay settlement past the first 30-second wait. Confirm
the lease stays available, no repair hold or successor appears before export,
and the finalizer drain waits for the same owner. After export, confirm saved
files, successful lease cleanup, and one bounded continuation. Reload before
checking the successor. Repeat with controller loss: the expired fence must
retain the exact source and require repair, never claim restored files or start
a second executor. The database-backed
[shutdown regression](../../server/src/__tests__/heartbeat-process-recovery.test.ts)
checks finalizer joining, release order, durable retry uniqueness, and reload.
The [settlement regressions](../../server/src/services/heartbeat/recovery.test.ts)
cover other-controller fencing and owner loss. Provider boundaries are controlled
in these tests; they do not qualify a live sandbox export.
For an abandoned native controller, follow the [stop-verification procedure](../../doc/native-workspace-finalization-recovery.md)
with a disposable deployment. A new platform-bound owner recovers automatically
only after the harness signs proof that the exact deployment fully stopped.
@@ -1,7 +1,7 @@
import { enableRunFailureReporting } from "./helpers/run-failure-reporting.js";
import { EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db";
import * as conversationContinuation from "../services/conversation-continuation.js";
import { environmentRuntimeService } from "../services/environment-runtime.js";
import { classifyEnvironmentCapabilities, environmentRuntimeService } from "../services/environment-runtime.js";
import * as workspaceRuntime from "../services/workspace-runtime.js";
import { nativeWorkspaceSyncInternals } from "../services/native-runtime/native-workspace-sync.js";
import { ensureNativeCompletionContract } from "../services/native-runtime/completion-contracts.js";
@@ -4081,6 +4081,63 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
});
});
it("drains late legacy settlement before releasing its source and scheduling one continuation", async () => {
const f = await seedRunFixture({ runtimeMode: "legacy", adapterType: "codex_local", agentStatus: "running" });
await db.update(heartbeatRuns).set({ scopeKind: "issue", issueId: f.issueId }).where(eq(heartbeatRuns.id, f.runId));
const { leaseId } = await seedEnvironmentLeaseFixture({ ...f, provider: "daytona", driver: "sandbox" });
await db.update(environmentLeases).set({ providerLeaseId: randomUUID(), metadata: { driver: "sandbox" } })
.where(eq(environmentLeases.id, leaseId));
const release = vi.fn(async () => {
const [saved] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
expect(saved.resultJson).toMatchObject({ workspaceRestored: true });
return (await db.update(environmentLeases).set({ status: "released", cleanupStatus: "success", releasedAt: new Date() })
.where(eq(environmentLeases.id, leaseId)).returning())[0];
});
const runtime = environmentRuntimeService(db, { drivers: [{ driver: "sandbox",
resolveCapabilities: async () => classifyEnvironmentCapabilities({}),
acquireRunLease: async () => { throw new Error("Unexpected acquire"); }, releaseRunLease: release }] });
const heartbeat = heartbeatService(db, { environmentRuntime: runtime });
const control = createAdapterExecutionControl();
adapterExecutionControls.set(f.runId, control);
vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
const shutdown = heartbeat.drainRunningRunsForShutdown("SIGTERM", new Date(), [f.runId]);
let drained = false;
let finalizers: Promise<void> | undefined;
try {
await vi.waitFor(() => expect(control.controller.signal.aborted).toBe(true));
await vi.advanceTimersByTimeAsync(30_000);
expect(await shutdown).toMatchObject({ interruptedRunIds: [f.runId], retryRunIds: [] });
finalizers = heartbeat.drainActiveRunExecutions().then(() => { drained = true; });
await Promise.resolve();
expect(drained).toBe(false);
expect(release).not.toHaveBeenCalled();
expect(await selectRecoveryRunsByAgent(f.agentId)).toHaveLength(1);
expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId))).toHaveLength(0);
await db.update(heartbeatRuns).set({ resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || '{"workspaceRestored":true}'::jsonb` })
.where(eq(heartbeatRuns.id, f.runId));
control.finish();
await finalizers;
expect(release).toHaveBeenCalledOnce();
const runs = await selectRecoveryRunsByAgent(f.agentId);
expect(runs).toHaveLength(2);
expect(runs.find(run => run.id !== f.runId)).toMatchObject({ status: "scheduled_retry", scheduledRetryAttempt: 1,
scopeKind: "issue", issueId: f.issueId });
const reloaded = heartbeatService(db, { environmentRuntime: runtime });
await reloaded.drainRunningRunsForShutdown("SIGTERM", new Date(), [f.runId]);
expect(await selectRecoveryRunsByAgent(f.agentId)).toHaveLength(2);
expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseId)))[0])
.toMatchObject({ status: "released", cleanupStatus: "success", metadata: { legacyShutdownWorkspaceSettlement: { state: "settled" } } });
expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId))).toHaveLength(0);
expect(mockAdapterExecute).not.toHaveBeenCalled();
} finally {
vi.useRealTimers();
control.finish();
adapterExecutionControls.delete(f.runId);
await shutdown;
await finalizers;
}
});
it("reports a selectively drained ACP run as lost when terminal persistence fails", async () => {
const child = spawnAliveProcess();
childProcesses.add(child);
+1
View File
@@ -2099,6 +2099,7 @@ async function startServerWithDatabaseTeardown(
await drainRunExecutionFinalizersForShutdown({
signal,
drain: drainHeartbeatExecutionFinalizers,
timeoutMs: 30_000,
log: logger,
});
}
+116 -8
View File
@@ -10,8 +10,9 @@ import { createHeartbeatRecovery, type HeartbeatRecoveryDependencies } from "./r
import type { HotRestartIntent } from "../hot-restart.js";
import type { NativeRestartRecoveryClaim, NativeRestartRecoveryDisposition } from "../native-runtime/index.js";
import { NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE } from "../native-runtime/native-runner-ownership.js";
import { allowLegacyShutdownWorkspaceCleanup, beginLegacyShutdownWorkspaceSettlement, finishLegacyShutdownWorkspaceSettlement, legacyShutdownWorkspaceResourceProtected, sweepLegacyShutdownWorkspaceSettlements } from "../legacy-shutdown-workspace-settlement.js";
import { allowLegacyShutdownWorkspaceCleanup, beginLegacyShutdownWorkspaceSettlement, finishLegacyShutdownWorkspaceSettlement, maintainLegacyShutdownWorkspaceSettlement, legacyShutdownWorkspaceResourceProtected, sweepLegacyShutdownWorkspaceSettlements } from "../legacy-shutdown-workspace-settlement.js";
import { adapterExecutionControls, createAdapterExecutionControl } from "../adapter-execution-control.js";
import { logger } from "../../middleware/logger.js";
type Run = typeof heartbeatRuns.$inferSelect;
@@ -290,6 +291,103 @@ describePostgres("heartbeat recovery module ownership", () => {
adapterExecutionControls.delete(f.run.id);
}
it("does not let an elapsed shutdown wait stop a sandbox whose exporter still owns settlement", async () => {
const f = await seedShutdownSource();
const control = createAdapterExecutionControl();
control.controller.abort();
adapterExecutionControls.set(f.run.id, control);
const stop = vi.fn(async () => ({ providerLeaseId: f.lease.providerLeaseId, state: "stopped" }));
const runtime = environmentRuntimeService(db, { drivers: [{ driver: "sandbox",
resolveCapabilities: async () => classifyEnvironmentCapabilities({}),
acquireRunLease: async () => { throw new Error("Unexpected acquire"); },
releaseRunLease: vi.fn(async () => null), retryPendingSandboxTeardown: stop }] });
try {
await db.update(heartbeatRuns).set({ status: "interrupted", errorCode: "server_shutdown_interrupted", finishedAt: new Date() })
.where(eq(heartbeatRuns.id, f.run.id));
const [source] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id));
expect(await allowLegacyShutdownWorkspaceCleanup(db, source, new Date(Date.now() + 60_000))).toBe(false);
await runtime.releaseRunLeases(f.run.id);
expect(stop).not.toHaveBeenCalled();
expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0].status).toBe("active");
expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.run.issueId!))).toHaveLength(0);
} finally { control.finish(); await deleteShutdownSource(f); }
});
it("renews the durable export fence for other controllers and expires it after owner loss", async () => {
const f = await seedShutdownSource();
const [initial] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id));
const originalDeadline = (initial.metadata!.legacyShutdownWorkspaceSettlement as { deadlineAt: string }).deadlineAt;
vi.useFakeTimers({ toFake: ["Date", "setInterval", "clearInterval"] });
const stopRenewing = maintainLegacyShutdownWorkspaceSettlement(db, f.run);
try {
await vi.advanceTimersByTimeAsync(10_000);
await vi.waitFor(async () => {
const [current] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id));
expect(Date.parse((current.metadata!.legacyShutdownWorkspaceSettlement as { deadlineAt: string }).deadlineAt))
.toBeGreaterThan(Date.parse(originalDeadline));
});
expect(await allowLegacyShutdownWorkspaceCleanup(db, initial, new Date(Date.parse(originalDeadline) + 1))).toBe(false);
await stopRenewing();
await vi.advanceTimersByTimeAsync(60_000);
await db.update(heartbeatRuns).set({ status: "interrupted", finishedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id));
await sweepLegacyShutdownWorkspaceSettlements(db);
expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0])
.toMatchObject({ status: "pending_cleanup", leasePolicy: "retain_on_failure",
metadata: { legacyShutdownWorkspaceSettlement: { state: "expired" }, sandboxStopAndRetain: { leaseId: f.lease.id } } });
} finally { await stopRenewing(); vi.useRealTimers(); await deleteShutdownSource(f); }
});
it("bounds a contended settlement renewal without blocking unrelated database work", async () => {
const f = await seedShutdownSource();
const warning = vi.spyOn(logger, "warn");
let unlock!: () => void;
let locked!: () => void;
const lockReady = new Promise<void>(resolve => { locked = resolve; });
const releaseLock = new Promise<void>(resolve => { unlock = resolve; });
const lock = db.transaction(async tx => {
await tx.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)).for("update");
locked();
await releaseLock;
});
await lockReady;
vi.useFakeTimers({ toFake: ["Date", "setInterval", "clearInterval"] });
const stopRenewing = maintainLegacyShutdownWorkspaceSettlement(db, f.run);
try {
await vi.advanceTimersByTimeAsync(10_000);
expect(await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)))
.toEqual([{ id: f.run.id }]);
await stopRenewing();
expect(warning).toHaveBeenCalledWith({ runId: f.run.id, errorKind: "lock_timeout" },
"Shutdown workspace settlement lease renewal failed; source remains protected locally");
} finally {
unlock();
await lock;
await stopRenewing();
vi.useRealTimers();
warning.mockRestore();
await deleteShutdownSource(f);
}
});
it.each([["57014", "statement_cancelled"], ["08006", "connection_failed"], ["private-credential", "unknown"]])(
"reports renewal failure %s without raw database diagnostics", async (code, errorKind) => {
const run = { id: randomUUID(), companyId: randomUUID() };
const privateValue = "postgres://private-credential@example.invalid/database";
const failure = new Error(privateValue, { cause: { code, query: privateValue, params: [privateValue] } });
const dbBoundary = { transaction: vi.fn().mockRejectedValue(failure) } as unknown as Db;
const warning = vi.spyOn(logger, "warn").mockImplementation(() => undefined);
vi.useFakeTimers({ toFake: ["setInterval", "clearInterval"] });
const stopRenewing = maintainLegacyShutdownWorkspaceSettlement(dbBoundary, run);
try {
await vi.advanceTimersByTimeAsync(10_000);
await stopRenewing();
expect(warning).toHaveBeenCalledExactlyOnceWith({ runId: run.id, errorKind },
"Shutdown workspace settlement lease renewal failed; source remains protected locally");
expect(JSON.stringify(warning.mock.calls)).not.toContain("private-credential");
} finally { await stopRenewing(); vi.useRealTimers(); warning.mockRestore(); }
},
);
it("does not turn a stale pending snapshot into a repair hold after settlement", async () => {
const f = await seedShutdownSource();
try {
@@ -425,12 +523,13 @@ describePostgres("heartbeat recovery module ownership", () => {
} finally { control.finish(); await deleteShutdownSource(f); }
});
it("pins an unjoined sandbox restore for repair before bounded shutdown cleanup", async () => {
it("joins a late successful export before cleanup and continuation after the shutdown wait", async () => {
const run = await seedRun();
await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: run.issueId } }).where(eq(heartbeatRuns.id, run.id));
const [lease] = await db.insert(environmentLeases).values({ companyId: run.companyId, heartbeatRunId: run.id,
issueId: run.issueId, status: "active", provider: "daytona", providerLeaseId: randomUUID(), leasePolicy: "ephemeral",
metadata: { driver: "sandbox", sandboxProviderPlugin: true, pluginId: "fixture-plugin", remoteCwd: "/home/daytona/workspace" } }).returning();
await db.update(agents).set({ adapterType: "codex_local" }).where(eq(agents.id, run.agentId));
const deps = callbacks(db);
deps.setRunStatusIfRunning.mockImplementation(async (id, status, patch) => {
expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, lease.id)))[0].metadata)
@@ -440,10 +539,10 @@ describePostgres("heartbeat recovery module ownership", () => {
});
deps.getRun.mockImplementation(async id => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0] ?? null);
deps.releaseEnvironmentLeasesForRun.mockImplementation(async () => {
expect((await deps.getRun(run.id))?.resultJson).toMatchObject({ workspaceRestoreFailure: "restore_failed",
workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [lease.id] } });
expect((await deps.getRun(run.id))?.resultJson).toEqual({ workspaceRestored: true, conversationContinuation: "continue_conversation_v1" });
expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, lease.id)))[0]).toMatchObject({
status: "pending_cleanup", leasePolicy: "retain_on_failure", metadata: { sandboxStopAndRetain: { leaseId: lease.id } } });
status: "active", leasePolicy: "ephemeral", metadata: { legacyShutdownWorkspaceSettlement: { state: "settled" } } });
await db.update(environmentLeases).set({ status: "released", cleanupStatus: "success", releasedAt: new Date() }).where(eq(environmentLeases.id, lease.id));
});
const control = createAdapterExecutionControl();
adapterExecutionControls.set(run.id, control);
@@ -452,15 +551,24 @@ describePostgres("heartbeat recovery module ownership", () => {
try {
await vi.waitFor(() => expect(control.controller.signal.aborted).toBe(true));
await vi.advanceTimersByTimeAsync(30_000);
await drain;
expect(await drain).toMatchObject({ interruptedRunIds: [run.id] });
expect(deps.activeRunExecutionPromises.size).toBe(1);
expect(deps.releaseEnvironmentLeasesForRun).not.toHaveBeenCalled();
expect(deps.scheduleBoundedRetryForRun).not.toHaveBeenCalled();
expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, run.issueId!))).toHaveLength(0);
await db.update(heartbeatRuns).set({ resultJson: { workspaceRestored: true, conversationContinuation: "continue_conversation_v1" } }).where(eq(heartbeatRuns.id, run.id));
control.finish();
await Promise.all([...deps.activeRunExecutionPromises]);
expect(deps.releaseEnvironmentLeasesForRun).toHaveBeenCalledOnce();
expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, run.issueId!)))
.toEqual([expect.objectContaining({ evidence: expect.objectContaining({ workspaceRestoreFailure: "restore_failed" }) })]);
expect(deps.scheduleBoundedRetryForRun).toHaveBeenCalledOnce();
expect(deps.scheduleBoundedRetryForRun).toHaveBeenCalledAfter(deps.releaseEnvironmentLeasesForRun);
expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, run.issueId!))).toHaveLength(0);
} finally {
vi.useRealTimers();
control.finish();
adapterExecutionControls.delete(run.id);
await drain;
await Promise.all([...deps.activeRunExecutionPromises]);
await db.delete(environmentLeases).where(eq(environmentLeases.id, lease.id));
await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, run.issueId!));
await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
+68 -79
View File
@@ -75,7 +75,7 @@ import {
} from "../hot-restart.js";
import { serverVersion } from "../../version.js";
import { adapterExecutionControls, waitForAdapterStop } from "../adapter-execution-control.js";
import { allowLegacyShutdownWorkspaceCleanup, beginLegacyShutdownWorkspaceSettlement, finishLegacyShutdownWorkspaceSettlement, sweepLegacyShutdownWorkspaceSettlements } from "../legacy-shutdown-workspace-settlement.js";
import { allowLegacyShutdownWorkspaceCleanup, beginLegacyShutdownWorkspaceSettlement, finishLegacyShutdownWorkspaceSettlement, maintainLegacyShutdownWorkspaceSettlement, sweepLegacyShutdownWorkspaceSettlements } from "../legacy-shutdown-workspace-settlement.js";
import type { environmentService } from "../environments.js";
import type { environmentRuntimeService, ProviderResourceDisposition } from "../environment-runtime.js";
@@ -1207,96 +1207,85 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
}),
},
);
if (!interruptedStatus.updated || !interruptedStatus.run) {
// A competing terminal write does not prove the owned adapter settled.
if (interruptedStatus.updated && interruptedStatus.run) interruptedRunIds.push(run.id);
const finishInterruptedRun = async () => {
if (control) {
control.controller.abort(new Error(message));
try {
await waitForAdapterStop(control.settled, Math.max(1, settlementDeadline.getTime() - Date.now()));
await finishLegacyShutdownWorkspaceSettlement(db, run);
await control.settled;
await finishLegacyShutdownWorkspaceSettlement(db, run);
}
if (!interruptedStatus.updated || !interruptedStatus.run) {
if (control) {
const current = await getRun(run.id, { includeExecutionEvidence: true });
if (current && HEARTBEAT_RUN_TERMINAL_STATUSES.some(status => status === current.status)) {
await releaseEnvironmentLeasesForRun({ runId: current.id, companyId: current.companyId,
agentId: current.agentId, status: current.status, failureReason: current.error ?? undefined });
}
} catch { /* The durable fence retains this source after controller loss. */ }
}
return;
}
let interrupted = interruptedStatus.run;
if (control) {
// Keep the exact lease open while its adapter stops the provider and
// copies files back. Provider release closes command admission and
// must not race the adapter's git export or instruction collection.
control.controller.abort(new Error(message));
try {
await waitForAdapterStop(control.settled, Math.max(1, settlementDeadline.getTime() - Date.now()), {
runId: run.id, adapterType: agent.adapterType, runtimeMode: run.runtimeMode,
abortRequested: true,
}, control);
await finishLegacyShutdownWorkspaceSettlement(db, run);
} catch {
logger.warn({ runId: run.id }, "Shutdown adapter settlement timed out; retaining the sandbox through normal cleanup");
const sources = await db.select().from(environmentLeases).where(and(
eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id),
sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'state' = 'pending'`,
));
// Stop-only settlement can already have released the source. Claim
// its current fence and pin recovery before classifying or retrying.
for (const source of sources) await allowLegacyShutdownWorkspaceCleanup(db, source,
new Date(Math.max(Date.now(), settlementDeadline.getTime())));
if (runningProcesses.get(run.id) === running && running) {
await terminateHeartbeatRunProcess({ pid: running.child.pid, processGroupId: running.processGroupId,
graceMs: Math.max(1, running.graceSec) * 1000 });
if (runningProcesses.get(run.id) === running) runningProcesses.delete(run.id);
}
return;
}
// Settlement can save restore-failure evidence after the shutdown CAS.
// Classification and retry must use the current row, not its snapshot.
interrupted = (await getRun(run.id, { includeExecutionEvidence: true })) ?? interrupted;
let interrupted = (control && await getRun(run.id, { includeExecutionEvidence: true })) ?? interruptedStatus.run;
await setWakeupStatus(run.wakeupRequestId, "cancelled", {
finishedAt: now,
error: null,
});
interrupted =
(await classifyAndPersistRunLiveness(
interrupted,
parseObject(interrupted.resultJson),
)) ?? interrupted;
await releaseEnvironmentLeasesForRun({
runId: interrupted.id,
companyId: interrupted.companyId,
agentId: interrupted.agentId,
status: interrupted.status,
failureReason: interrupted.error ?? undefined,
});
const retry = await enqueueProcessLossRetry(interrupted, agent, now);
if (!retry) {
await releaseIssueExecutionAndPromote(interrupted);
} else {
retryRunIds.push(retry.id);
}
await appendRunEvent(interrupted, {
eventType: "lifecycle",
stream: "system",
level: "warn",
message,
payload: {
signal,
...(run.processPid ? { processPid: run.processPid } : {}),
...(run.processGroupId ? { processGroupId: run.processGroupId } : {}),
...(retry ? { retryRunId: retry.id } : {}),
},
});
await finalizeAgentStatus(run.agentId, "interrupted", message, {
wasFirstHeartbeat: timerClaimWasFirstHeartbeat(run),
});
};
if (!control) {
await finishInterruptedRun();
return;
}
await setWakeupStatus(run.wakeupRequestId, "cancelled", {
finishedAt: now,
error: null,
const stopRenewing = maintainLegacyShutdownWorkspaceSettlement(db, run);
control.controller.abort(new Error(message));
const settlement = finishInterruptedRun().finally(stopRenewing);
const tracked = settlement.catch(() => {
logger.warn({ runId: run.id, errorKind: "shutdown_settlement_failed" }, "Shutdown workspace settlement failed; durable cleanup protection remains in place");
});
interrupted =
(await classifyAndPersistRunLiveness(
interrupted,
parseObject(interrupted.resultJson),
)) ?? interrupted;
await releaseEnvironmentLeasesForRun({
runId: interrupted.id,
companyId: interrupted.companyId,
agentId: interrupted.agentId,
status: interrupted.status,
failureReason: interrupted.error ?? undefined,
});
const retry = await enqueueProcessLossRetry(interrupted, agent, now);
if (!retry) {
await releaseIssueExecutionAndPromote(interrupted);
} else {
retryRunIds.push(retry.id);
activeRunExecutionPromises.add(tracked);
void tracked.finally(() => activeRunExecutionPromises.delete(tracked));
try {
await waitForAdapterStop(settlement, Math.max(1, settlementDeadline.getTime() - Date.now()), {
runId: run.id, adapterType: agent.adapterType, runtimeMode: run.runtimeMode, abortRequested: true,
}, control);
} catch {
logger.warn({ runId: run.id }, "Shutdown workspace settlement continues in the finalizer drain");
}
await appendRunEvent(interrupted, {
eventType: "lifecycle",
stream: "system",
level: "warn",
message,
payload: {
signal,
...(run.processPid ? { processPid: run.processPid } : {}),
...(run.processGroupId ? { processGroupId: run.processGroupId } : {}),
...(retry ? { retryRunId: retry.id } : {}),
},
});
await finalizeAgentStatus(run.agentId, "interrupted", message, {
wasFirstHeartbeat: timerClaimWasFirstHeartbeat(run),
});
interruptedRunIds.push(interrupted.id);
}));
for (const [index, settlement] of settlements.entries()) {
if (settlement.status === "rejected") logger.warn({ runId: activeRuns[index]!.run.id,
@@ -3,10 +3,53 @@ import { environmentLeases, heartbeatRuns, issues, type Db } from "@paperclipai/
import { adapterExecutionControls } from "./adapter-execution-control.js";
import { legacyControllerBootId } from "./legacy-controller-lease.js";
import { LEGACY_WORKSPACE_RECOVERY_SCHEMA } from "./workspace-restore-recovery-state.js";
import { logger } from "../middleware/logger.js";
const SCHEMA = "paperclip.legacy-shutdown-workspace-settlement.v1";
type Lease = Pick<typeof environmentLeases.$inferSelect, "id" | "companyId" | "heartbeatRunId" | "metadata">;
type Run = Pick<typeof heartbeatRuns.$inferSelect, "id" | "companyId">;
const SETTLEMENT_RENEW_MS = 10_000;
const SETTLEMENT_LEASE_MS = 30_000;
function settlementRenewalFailureKind(error: unknown) {
for (let depth = 0; depth < 4 && error && typeof error === "object"; depth++) {
const value = error as { code?: unknown; cause?: unknown };
if (value.code === "55P03") return "lock_timeout";
if (value.code === "57014") return "statement_cancelled";
if (typeof value.code === "string" && (/^08[0-9A-Z]{3}$/.test(value.code)
|| ["ECONNRESET", "ECONNREFUSED", "ETIMEDOUT", "CONNECT_TIMEOUT", "CONNECTION_CLOSED", "CONNECTION_ENDED", "CONNECTION_DESTROYED"].includes(value.code))) {
return "connection_failed";
}
error = value.cause;
}
return "unknown";
}
export function maintainLegacyShutdownWorkspaceSettlement(db: Db, run: Run) {
let stopped = false;
let pending: Promise<void> | null = null;
const renew = () => {
if (stopped || pending) return;
pending = db.transaction(async tx => {
await tx.execute(sql`SET LOCAL lock_timeout = '1s'`);
await tx.execute(sql`SET LOCAL statement_timeout = '5s'`);
await tx.update(environmentLeases).set({ metadata: sql`jsonb_set(${environmentLeases.metadata},
'{legacyShutdownWorkspaceSettlement,deadlineAt}', ${JSON.stringify(new Date(Date.now() + SETTLEMENT_LEASE_MS).toISOString())}::jsonb)` })
.where(and(eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id),
sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'schema' = ${SCHEMA}`,
sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'runId' = ${run.id}`,
sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'controllerBootId' = ${legacyControllerBootId}`,
sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'state' = 'pending'`,
sql`(${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'deadlineAt')::timestamptz > clock_timestamp()`,
));
}).catch(error => {
logger.warn({ runId: run.id, errorKind: settlementRenewalFailureKind(error) }, "Shutdown workspace settlement lease renewal failed; source remains protected locally");
}).finally(() => { pending = null; });
};
const timer = setInterval(renew, SETTLEMENT_RENEW_MS);
timer.unref?.();
return async () => { stopped = true; clearInterval(timer); await pending; };
}
export function pendingLegacyShutdownWorkspaceSettlement(lease: Lease) {
const value = lease.metadata?.legacyShutdownWorkspaceSettlement;
@@ -90,6 +133,7 @@ export async function allowLegacyShutdownWorkspaceCleanup(db: Db, lease: Lease,
// some adapters to settle. This exception never permits provider destruction.
if (options.ownerStopOnly && marker.controllerBootId === legacyControllerBootId
&& adapterExecutionControls.get(run.id)?.controller.signal.aborted) return true;
if (marker.controllerBootId === legacyControllerBootId && adapterExecutionControls.has(run.id)) return false;
const deadline = Date.parse(marker.deadlineAt as string);
if (!Number.isFinite(deadline) || deadline > now.getTime()) return false;
// A killed container cannot certify export. Preserve source and hold in