mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-27 19:35:13 +01:00
fix: preserve native startup during controller restart (#17511)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The native runtime keeps a durable runner alive across controller restarts. > - A restart can begin while a provider session is still opening. > - The old controller fenced read-only run admission before it could publish that prepared session. > - Admission failure then closed the runner that shutdown needed to detach. > - This pull request lets only the existing startup finish admission within the bounded shutdown wait. > - The controller detaches that exact session before it can submit a turn. New work and state changes stay fenced. ## Linked Issues or Issue Description **What happened?** A controller restart during native provider startup could close the prepared runner with `native control-plane run admission failed`. Recovery then found no running process and placed the task in a reconciliation hold. The same admission path follows both fresh startup and replacement after a failed provider resume. **Expected behavior** Shutdown must detach the existing prepared session so the next controller can recover it through the normal identity checks. Shutdown must not admit a new executor, submit a turn, or override an explicit Stop. **Steps to reproduce** 1. Delay a native backend while it opens its provider session. 2. Request controller restart detachment for that run. 3. Let backend startup finish before the 30-second deadline. 4. Observe that read-only admission fails and shutdown reports the run as inactive instead of detached. **Additional context** Related work: #17073 handles remote process-monitor classification; #17459 handles macOS process fingerprints; #17097 covers broader hot upgrade recovery. This change fixes the startup admission fence. It does not relax process identity or provider launch-profile checks. ## What Changed - Separate the read-only admission guard from the guard on control-plane state changes. - Let shutdown await only the startup already registered for the run. Close its admission window when the wait ends. - Reject a new execution after its run is fenced for shutdown. - Detach the prepared session and exit before runtime-question attachment or provider turn submission. - If Stop is pending before or after publication, retain its exact cancellation acknowledgment and cleanup owner across authority reads and runtime-question attachment. - Share one 30-second admission/Stop-settlement deadline across the restart sweep. Report an unsettled run as pending while healthy runs continue hot-restart adoption. - Cover the real runtime and admission path, deadline expiry, new-start rejection, persisted binding checks, and write fences. - Document the bounded startup handoff in the restart ownership contract. ## Verification - Red: the new actual-runtime/admission regression failed on the previous code. Shutdown reported the run as inactive and did not detach it. - Red: review regressions reproduced detach before a post-publication Stop acknowledgment and an actual hot-restart preparation error when startup Stop cleanup timed out. - Green on `e10d0cb238`: 18 startup/restart tests passed, including Stop during authority reads and runtime-question attachment, the real recovery preparation and shutdown coordinator with a stalled Stop plus healthy peer, and the shared sweep deadline. The earlier 36-case Stop/restart selection also passed before this follow-up. - Both complete changed test files previously passed on head `8e9c2610fe`: 857 tests. They used real PostgreSQL 18 with a fresh database per fixture. An earlier local harness attempt reused one database and hit duplicate fixture keys; that harness was corrected before this complete pass. - Real PostgreSQL: 7 admission and revocation tests passed. These cover wrong runner identity, persisted binding mismatch, checkpoint/result rejection, and in-flight write revocation. The local test harness used a fresh Docker PostgreSQL 18 instance because macOS embedded Postgres exhausted the host semaphore limit. Only database provisioning was replaced; assertions and production port code were unchanged. - Transport safety acceptance: the current source transport controlled a real Linux ARM runner and a deterministic Codex test provider in a disposable Debian container. After startup, detach and generic cleanup preserved the process. A new controller authenticated the same PID and provider session. The provider log had exactly one session start and no turn start. Duplicate launch was forbidden, the retained workspace was unchanged, and no stop or suspend command was issued. - The Linux acceptance check tests the shared native transport safety contract. It is not a live Claude response, a hosted Daytona run, or recovery of the original task. Those remain unverified. No original task, sandbox, credentials, or permissions were changed. - `pnpm -r typecheck` and `pnpm build` passed before the review follow-up. Server typecheck passed on current head `e10d0cb238`; current-head CI also passed the full typecheck and build gates. - The full local `pnpm test:run` command was attempted, then stopped after repeated embedded PostgreSQL bootstrap setup failures. That broad local run is incomplete. The current-head four-file Docker PostgreSQL run passed 879 tests, with one existing warm-copy `claim-uncertain` case timing out and recovery failing to import because the temporary provisioning mock omitted a support export. The harness is corrected and affected cases are being rechecked. The timeout is unexplained; it is not being labeled an infrastructure failure. No assertions are skipped by the database provisioning replacement. - All 55 clean-host CI checks passed on current head `e10d0cb238` (2 conditional skips), including all general test shards and native runner verification. Greptile reviewed that exact head at 5/5 with no findings. The previous Paperclip review findings are addressed in [the review response](https://github.com/paperclipai/paperclip/pull/17511#issuecomment-6391520242); its new advisory review remains queued. ## Risks - Admission must stay read-only. All writes retain the original synchronous controller fence, and both admission checks still validate the exact persisted binding. - Startup that exceeds the 30-second shutdown wait still fails closed. It cannot claim a successful detach or start a turn. - No schema migration or provider configuration change is included. ## Model Used OpenAI Codex, based on GPT-6, with reasoning, repository tools, code execution, and test execution. The exact runtime model ID and context window are not exposed in 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
1 parent
8a85d9d647
commit
2a9d527c68
5 files changed
+435
-18
No files matched your search
@@ -1683,6 +1683,18 @@ controller must recover it through the authenticated ownership checks. This
|
||||
preserves active work and queued messages without treating a server restart as
|
||||
user cancellation.
|
||||
|
||||
During controller shutdown, already registered provider startups share one
|
||||
30-second sweep deadline to finish read-only run admission and publish their exact
|
||||
sessions for detachment. Its company, run, session, runner and completion-contract binding
|
||||
must still match persisted authority. The draining controller cannot checkpoint,
|
||||
publish events, submit a turn or settle the run. A new startup after the shutdown
|
||||
fence, or admission after the window expires, remains rejected. A Stop pending
|
||||
before or after session publication keeps cancellation and cleanup with the
|
||||
current owner. Shutdown waits for that owner only until the same sweep deadline;
|
||||
it does not detach the session or submit a prompt while Stop is pending. An
|
||||
unsettled owner remains explicitly pending for reconciliation. Its timeout does
|
||||
not abort detachment of healthy sessions or trigger a global graceful drain.
|
||||
|
||||
Recovery restores a structured result proposal from the durable run events when
|
||||
its append survived but the active turn's session checkpoint did not. Advancing
|
||||
the replay cursor must preserve that proposal for the exact run, session, source,
|
||||
|
||||
@@ -33,6 +33,7 @@ import { dirname, join } from "node:path";
|
||||
import {
|
||||
heartbeatRuns,
|
||||
heartbeatRunEvents,
|
||||
issueThreadInteractions,
|
||||
issues,
|
||||
nativeRunFinalizations,
|
||||
nativeRunResults,
|
||||
@@ -49,6 +50,7 @@ import {
|
||||
validatePrpStructuredRunResult,
|
||||
validatePrpEvent,
|
||||
parseNativeExecutionInput,
|
||||
executeNativeSession,
|
||||
type NativeExecutionInputV1,
|
||||
type NativeExecutionInputV3,
|
||||
type NativeExecutionInput,
|
||||
@@ -343,6 +345,7 @@ import {
|
||||
executePaperclipNativeSession,
|
||||
detachNativeSessionsForRestart,
|
||||
NativeControllerDetachedForRestartError,
|
||||
NativeCancellationPendingRecoveryError,
|
||||
getNativeSessionSteeringState,
|
||||
NativeSessionSteeringError,
|
||||
assertRemoteRunnerBuildMetadata,
|
||||
@@ -5695,6 +5698,10 @@ function leaseDb(
|
||||
agentId: boundExecution.binding.agentId,
|
||||
companyId: boundExecution.binding.companyId,
|
||||
nativeIssueId: boundExecution.binding.issueId,
|
||||
nativeSessionId: boundExecution.session.normalizedSessionId ?? `session-${boundExecution.binding.runId}`,
|
||||
runnerInstanceId: "runner",
|
||||
completionContractId: boundExecution.completionContract.id,
|
||||
completionContractSha256: boundExecution.completionContract.sha256,
|
||||
resultJson: runResultJson,
|
||||
runnerProfileJson,
|
||||
runtimeMode: "native",
|
||||
@@ -5892,6 +5899,306 @@ describe("native startup cancellation fence", () => {
|
||||
});
|
||||
|
||||
describe("native startup restart detachment", () => {
|
||||
it("uses one admission deadline for the whole restart sweep", async () => {
|
||||
const inputs = ["first", "second"].map(label => {
|
||||
const input = structuredClone(execution);
|
||||
input.binding.runId = `restart-shared-deadline-${label}`;
|
||||
input.binding.agentId = `restart-shared-deadline-agent-${label}`;
|
||||
return input;
|
||||
});
|
||||
const releases: Array<() => void> = [];
|
||||
const ready: Array<Promise<void>> = [];
|
||||
state.execute.mockReset();
|
||||
for (const _input of inputs) {
|
||||
let started!: () => void;
|
||||
ready.push(new Promise<void>(resolve => { started = resolve; }));
|
||||
const opening = new Promise<void>(resolve => { releases.push(resolve); });
|
||||
state.execute.mockImplementationOnce(async () => {
|
||||
started(); await opening;
|
||||
throw new Error("startup fixture released");
|
||||
});
|
||||
}
|
||||
const outcomes = inputs.map(input => executePaperclipNativeSession({
|
||||
db: leaseDb(input), execution: input, runnerInstanceId: "runner",
|
||||
}).catch(error => error));
|
||||
await Promise.all(ready);
|
||||
vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] });
|
||||
const finished = vi.fn();
|
||||
const runIds = inputs.map(input => input.binding.runId);
|
||||
const shutdown = detachNativeSessionsForRestart(runIds).then(finished);
|
||||
try {
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
expect(finished).toHaveBeenCalledWith({ detachedRunIds: [], inactiveRunIds: [],
|
||||
unsupportedRunIds: [], pendingRunIds: runIds });
|
||||
} finally {
|
||||
vi.useRealTimers(); releases.forEach(release => release());
|
||||
await Promise.all(outcomes); await shutdown;
|
||||
}
|
||||
});
|
||||
|
||||
it.each(["authority", "questions"])("keeps a post-publication Stop owner through %s attachment and restart", async (seam) => {
|
||||
const restarting = structuredClone(execution);
|
||||
restarting.binding.runId = `restart-stop-after-publication-${seam}`;
|
||||
let published!: () => void, releaseRead!: () => void, acknowledgeStop!: () => void;
|
||||
let cleaning!: () => void, releaseCleanup!: () => void;
|
||||
const publication = new Promise<void>(resolve => { published = resolve; });
|
||||
const read = new Promise<void>(resolve => { releaseRead = resolve; });
|
||||
const receipt = new Promise<void>(resolve => { acknowledgeStop = resolve; });
|
||||
const cleanupStarted = new Promise<void>(resolve => { cleaning = resolve; });
|
||||
const cleanup = new Promise<void>(resolve => { releaseCleanup = resolve; });
|
||||
const detach = vi.fn(async () => undefined);
|
||||
const cancel = vi.fn(() => ({ cleanup: receipt }));
|
||||
const submit = vi.fn();
|
||||
const db = leaseDb(restarting);
|
||||
const select = db.select.bind(db);
|
||||
db.select = ((fields?: Record<string, unknown>) => {
|
||||
if (seam === "authority" && fields && Object.keys(fields).length === 2 && "status" in fields && "resultJson" in fields) {
|
||||
return { from: () => ({ where: () => ({ limit: () => {
|
||||
published();
|
||||
return read.then(() => [{ status: "running", resultJson: {} }]);
|
||||
} }) }) };
|
||||
}
|
||||
if (seam === "questions" && !fields) {
|
||||
const query = select();
|
||||
return { from: (table: unknown) => table === issueThreadInteractions
|
||||
? { where: () => ({ orderBy: () => { published(); return read.then(() => []); } }) }
|
||||
: query.from(table as never) };
|
||||
}
|
||||
return select(fields as never);
|
||||
}) as typeof db.select;
|
||||
state.execute.mockReset().mockImplementationOnce(async options => {
|
||||
try {
|
||||
await options.onSession({ detachControllerForRestart: detach, cancel, resolveRuntimeRequest: vi.fn() });
|
||||
submit();
|
||||
} finally {
|
||||
cleaning();
|
||||
await cleanup;
|
||||
await options.onSession(null);
|
||||
}
|
||||
});
|
||||
const outcome = executePaperclipNativeSession({ db, execution: restarting, runnerInstanceId: "runner" }).catch(error => error);
|
||||
await publication;
|
||||
const stopping = cancelNativeSession(restarting.binding.runId, "operator Stop after publication");
|
||||
const completed = vi.fn();
|
||||
const shutdown = detachNativeSessionsForRestart([restarting.binding.runId]).then(completed);
|
||||
try {
|
||||
releaseRead();
|
||||
await new Promise(resolve => setImmediate(resolve));
|
||||
expect(cancel).toHaveBeenCalledOnce();
|
||||
expect(detach).not.toHaveBeenCalled();
|
||||
expect(completed).not.toHaveBeenCalled();
|
||||
// Admission must not unwind into generic close while Stop awaits its ACK.
|
||||
const cleanupObserved = vi.fn();
|
||||
void cleanupStarted.then(cleanupObserved);
|
||||
await new Promise(resolve => setImmediate(resolve));
|
||||
expect(cleanupObserved).not.toHaveBeenCalled();
|
||||
acknowledgeStop();
|
||||
expect(await stopping).toBe(true);
|
||||
await cleanupStarted;
|
||||
expect(completed).not.toHaveBeenCalled();
|
||||
releaseCleanup();
|
||||
await outcome;
|
||||
await shutdown;
|
||||
expect(completed).toHaveBeenCalledWith(expect.objectContaining({
|
||||
detachedRunIds: [], inactiveRunIds: [restarting.binding.runId], pendingRunIds: [],
|
||||
}));
|
||||
expect(detach).not.toHaveBeenCalled();
|
||||
expect(submit).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
releaseRead(); acknowledgeStop(); releaseCleanup();
|
||||
await stopping; await outcome; await shutdown;
|
||||
}
|
||||
});
|
||||
|
||||
it.each(["startup", "active"])("keeps healthy hot-restart adoption when another %s Stop cleanup exceeds the sweep deadline", async (mode) => {
|
||||
const { coordinateHeartbeatSchedulerShutdown } = await import("../../shutdown.js");
|
||||
const { createHeartbeatRecovery } = await import("../heartbeat/recovery.js");
|
||||
const { writeHotRestartIntent, readHotRestartIntent } = await import("../hot-restart.js");
|
||||
const home = await mkdtemp(join(tmpdir(), "native-restart-stop-sweep-"));
|
||||
const previousHome = process.env.PAPERCLIP_HOME;
|
||||
process.env.PAPERCLIP_HOME = home;
|
||||
await writeHotRestartIntent({ previousServerPid: process.pid, previousServerIdentity: "test-controller" });
|
||||
const stoppingExecution = structuredClone(execution);
|
||||
stoppingExecution.binding.runId = `restart-${mode}-stop-timeout`;
|
||||
stoppingExecution.session.normalizedSessionId = "d371f9bb-6c46-44c6-9e3a-026fa9be0153";
|
||||
const healthyExecution = structuredClone(execution);
|
||||
healthyExecution.binding.runId = `restart-healthy-with-${mode}-stopping-peer`;
|
||||
healthyExecution.session.normalizedSessionId = "a8fa0cb8-9cf1-46b0-96f0-78518b15847e";
|
||||
healthyExecution.binding.agentId = "healthy-restart-agent";
|
||||
let stoppedReady!: () => void, healthyReady!: () => void;
|
||||
let releaseStopped!: () => void, releaseHealthy!: () => void, opening!: () => void, open!: () => void;
|
||||
const openingStarted = new Promise<void>(resolve => { opening = resolve; });
|
||||
const openingGate = new Promise<void>(resolve => { open = resolve; });
|
||||
const stopPublished = new Promise<void>(resolve => { stoppedReady = resolve; });
|
||||
const healthyPublished = new Promise<void>(resolve => { healthyReady = resolve; });
|
||||
const stoppedLifetime = new Promise<void>(resolve => { releaseStopped = resolve; });
|
||||
const healthyLifetime = new Promise<void>(resolve => { releaseHealthy = resolve; });
|
||||
const stoppedDetach = vi.fn(async () => undefined);
|
||||
const healthyDetach = vi.fn(async () => undefined);
|
||||
state.execute.mockReset()
|
||||
.mockImplementationOnce(async options => {
|
||||
opening(); await openingGate;
|
||||
try {
|
||||
await options.onSession({ detachControllerForRestart: stoppedDetach, cancel: () => ({ cleanup: Promise.resolve() }) });
|
||||
} finally {
|
||||
stoppedReady(); await stoppedLifetime;
|
||||
await options.onSession(null);
|
||||
}
|
||||
throw new Error("stopped owner settled");
|
||||
})
|
||||
.mockImplementationOnce(async options => {
|
||||
await options.onSession({ detachControllerForRestart: healthyDetach });
|
||||
healthyReady(); await healthyLifetime;
|
||||
await options.onSession(null);
|
||||
throw new Error("healthy controller detached");
|
||||
});
|
||||
const stoppedOutcome = executePaperclipNativeSession({ db: leaseDb(stoppingExecution), execution: stoppingExecution, runnerInstanceId: "runner" }).catch(error => error);
|
||||
await openingStarted;
|
||||
const stopping = mode === "startup" ? cancelNativeSession(stoppingExecution.binding.runId, "Stop while opening") : null;
|
||||
open();
|
||||
await stopPublished;
|
||||
await stopping;
|
||||
const healthyOutcome = executePaperclipNativeSession({ db: leaseDb(healthyExecution), execution: healthyExecution, runnerInstanceId: "runner" }).catch(error => error);
|
||||
await healthyPublished;
|
||||
if (mode === "active") await cancelNativeSession(stoppingExecution.binding.runId, "operator Stop after active admission");
|
||||
const snapshotRows = [stoppingExecution, healthyExecution].map(item => ({
|
||||
adapterType: "paperclip_runner", adapterConfig: {},
|
||||
run: { id: item.binding.runId, companyId: item.binding.companyId, agentId: item.binding.agentId,
|
||||
runtimeMode: "native", nativeSessionId: item.session.normalizedSessionId, status: "running", contextSnapshot: {} },
|
||||
}));
|
||||
const snapshotDb = { select: () => ({ from: () => ({ innerJoin: () => ({ where: async () => snapshotRows }) }) }) } as unknown as Db;
|
||||
const { logger } = await import("../../middleware/logger.js");
|
||||
const logged = vi.spyOn(logger, "info").mockImplementation(() => undefined);
|
||||
const recovery = createHeartbeatRecovery(snapshotDb, {
|
||||
enterShutdown: vi.fn(), appendRunEvent: vi.fn(async () => undefined),
|
||||
} as unknown as import("../heartbeat/recovery.js").HeartbeatRecoveryDependencies);
|
||||
vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] });
|
||||
const shutdown = coordinateHeartbeatSchedulerShutdown({ signal: "SIGTERM",
|
||||
waitForHeartbeatSchedulerIdle: async () => undefined,
|
||||
prepareHotRestartShutdown: recovery.prepareHotRestartShutdown,
|
||||
});
|
||||
try {
|
||||
// Wait for the real snapshot I/O before advancing its bounded sweep.
|
||||
await vi.waitFor(() => expect(vi.getTimerCount()).toBeGreaterThan(0));
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
expect(await shutdown).toMatchObject({ hotRestart: { mode: "hot_restart", skipDrain: true }, preparationError: null });
|
||||
expect(logged).toHaveBeenCalledWith(expect.objectContaining({ detachedNativeSessions: {
|
||||
detachedRunIds: [healthyExecution.binding.runId], inactiveRunIds: [], unsupportedRunIds: [],
|
||||
pendingRunIds: [stoppingExecution.binding.runId],
|
||||
} }), expect.any(String));
|
||||
expect((await readHotRestartIntent())?.shutdownSnapshot?.activeRuns.map(run => run.runId))
|
||||
.toEqual([stoppingExecution.binding.runId, healthyExecution.binding.runId]);
|
||||
expect(stoppedDetach).not.toHaveBeenCalled();
|
||||
expect(healthyDetach).toHaveBeenCalledOnce();
|
||||
} finally {
|
||||
vi.useRealTimers(); releaseStopped(); releaseHealthy();
|
||||
await stoppedOutcome; await healthyOutcome; await shutdown;
|
||||
logged.mockRestore();
|
||||
if (previousHome === undefined) delete process.env.PAPERCLIP_HOME;
|
||||
else process.env.PAPERCLIP_HOME = previousHome;
|
||||
await rm(home, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it.each(["detach", "expired", "stop"])("bounds read-only startup admission during drain: %s", async (mode) => {
|
||||
const expired = mode === "expired";
|
||||
const stopped = mode === "stop";
|
||||
const restarting = structuredClone(execution);
|
||||
restarting.binding.runId = `restart-real-admission-${mode}`;
|
||||
restarting.session.normalizedSessionId = "restart-real-admission-session";
|
||||
const identity = { ...restarting.binding, sessionId: restarting.session.normalizedSessionId };
|
||||
let release!: () => void, started!: () => void;
|
||||
const opening = new Promise<void>(resolve => { release = resolve; });
|
||||
const pending = new Promise<void>(resolve => { started = resolve; });
|
||||
const detach = vi.fn(async () => undefined);
|
||||
const close = vi.fn(async () => undefined);
|
||||
const startTurn = vi.fn();
|
||||
let acknowledgeStop!: () => void;
|
||||
const stopReceipt = new Promise<void>(resolve => { acknowledgeStop = resolve; });
|
||||
const cancel = vi.fn(() => ({ cleanup: stopReceipt }));
|
||||
const backend = {
|
||||
descriptor: async () => ({ kind: "mock" as const, name: "startup-drain", version: "1",
|
||||
capabilities: { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true },
|
||||
runtimeContextCapabilities: { instructions: "native", skills: "native", mcp: "native" } }),
|
||||
openSession: vi.fn(async () => {
|
||||
started(); await opening;
|
||||
return { identity: () => identity, detachControllerForRestart: detach, close, startTurn, cancel,
|
||||
snapshot: async () => ({ sessionId: identity.sessionId, identity, backendKind: "mock" as const }),
|
||||
};
|
||||
}),
|
||||
};
|
||||
let runtimeError: unknown;
|
||||
state.execute.mockReset().mockImplementationOnce(async options => {
|
||||
try { return await executeNativeSession({ ...options, onSessionAdmission: undefined }); } catch (error) { runtimeError = error; throw error; }
|
||||
});
|
||||
const outcome = executePaperclipNativeSession({ db: leaseDb(restarting), execution: restarting,
|
||||
runnerInstanceId: "runner", backend: backend as never }).catch(error => error);
|
||||
await Promise.race([pending, outcome.then(error => { throw runtimeError ?? error; })]);
|
||||
const stopping = stopped ? cancelNativeSession(restarting.binding.runId, "operator Stop during startup") : null;
|
||||
if (expired) vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
|
||||
const shutdown = detachNativeSessionsForRestart([restarting.binding.runId]).catch(error => error);
|
||||
try {
|
||||
if (expired) {
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
expect(await shutdown).toMatchObject({ pendingRunIds: [restarting.binding.runId], detachedRunIds: [] });
|
||||
vi.useRealTimers();
|
||||
} else {
|
||||
await new Promise(resolve => setImmediate(resolve));
|
||||
}
|
||||
release();
|
||||
if (stopped) {
|
||||
await vi.waitFor(() => expect(cancel).toHaveBeenCalledOnce());
|
||||
expect(detach).not.toHaveBeenCalled();
|
||||
acknowledgeStop();
|
||||
expect(await stopping).toBe(true);
|
||||
}
|
||||
expect(await outcome).toBeInstanceOf(NativeControllerDetachedForRestartError);
|
||||
if (stopped) {
|
||||
expect(await shutdown).toMatchObject({ detachedRunIds: [], inactiveRunIds: [restarting.binding.runId] });
|
||||
expect(detach).not.toHaveBeenCalled();
|
||||
} else if (expired) {
|
||||
expect(detach).not.toHaveBeenCalled();
|
||||
expect(close).toHaveBeenCalledWith({ reason: "native control-plane run admission failed" });
|
||||
} else {
|
||||
expect(await shutdown).toMatchObject({ detachedRunIds: [restarting.binding.runId], inactiveRunIds: [] });
|
||||
expect(detach).toHaveBeenCalledOnce();
|
||||
expect(close).not.toHaveBeenCalledWith({ reason: "native control-plane run admission failed" });
|
||||
}
|
||||
expect(backend.openSession).toHaveBeenCalledOnce();
|
||||
expect(startTurn).not.toHaveBeenCalled();
|
||||
} finally { vi.useRealTimers(); release(); acknowledgeStop(); await stopping; await outcome; await shutdown; }
|
||||
});
|
||||
it("bounds a pending startup Stop without detaching its cleanup owner", async () => {
|
||||
const restarting = structuredClone(execution);
|
||||
restarting.binding.runId = "restart-startup-stop-cleanup-timeout";
|
||||
let open!: () => void, started!: () => void, releaseCleanup!: () => void, cleaning!: () => void;
|
||||
const opening = new Promise<void>(resolve => { open = resolve; });
|
||||
const pending = new Promise<void>(resolve => { started = resolve; });
|
||||
const cleanup = new Promise<void>(resolve => { releaseCleanup = resolve; });
|
||||
const cleanupStarted = new Promise<void>(resolve => { cleaning = resolve; });
|
||||
const detach = vi.fn(async () => undefined);
|
||||
const cancel = vi.fn(() => ({ cleanup: Promise.resolve() }));
|
||||
state.execute.mockReset().mockImplementationOnce(async options => {
|
||||
started(); await opening;
|
||||
try { await options.onSession({ detachControllerForRestart: detach, cancel }); }
|
||||
finally { cleaning(); await cleanup; }
|
||||
});
|
||||
const outcome = executePaperclipNativeSession({ db: leaseDb(restarting), execution: restarting, runnerInstanceId: "runner" }).catch(error => error);
|
||||
await pending;
|
||||
const stopping = cancelNativeSession(restarting.binding.runId, "operator Stop before restart");
|
||||
vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
|
||||
const shutdown = detachNativeSessionsForRestart([restarting.binding.runId]).catch(error => error);
|
||||
try {
|
||||
open();
|
||||
await cleanupStarted;
|
||||
expect(await stopping).toBe(true);
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
expect(await shutdown).toMatchObject({ pendingRunIds: [restarting.binding.runId], detachedRunIds: [] });
|
||||
expect(detach).not.toHaveBeenCalled();
|
||||
expect(cancel).toHaveBeenCalledOnce();
|
||||
} finally { vi.useRealTimers(); open(); releaseCleanup(); await stopping; await outcome; await shutdown; }
|
||||
});
|
||||
|
||||
it("waits for a checkpointed governed settlement before relinquishing its controller", async () => {
|
||||
const restarting = structuredClone(execution);
|
||||
restarting.binding.runId = "restart-governed-settlement";
|
||||
@@ -6033,7 +6340,7 @@ describe("native startup restart detachment", () => {
|
||||
const detached = detachNativeSessionsForRestart([restarting.binding.runId]).catch(error => error);
|
||||
if (exceedDeadline) {
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
expect(await detached).toMatchObject({ message: "native_restart_startup_not_ready" });
|
||||
expect(await detached).toMatchObject({ pendingRunIds: [restarting.binding.runId], detachedRunIds: [] });
|
||||
vi.useRealTimers();
|
||||
} else {
|
||||
release();
|
||||
@@ -6049,18 +6356,13 @@ describe("native startup restart detachment", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("remembers shutdown while the session is still opening and detaches its late publication", async () => {
|
||||
it("rejects an entirely new startup after the run has been fenced for shutdown", async () => {
|
||||
const restarting = structuredClone(execution);
|
||||
restarting.binding.runId = "restart-during-session-open";
|
||||
const detach = vi.fn(async () => undefined);
|
||||
restarting.binding.runId = "restart-new-startup-after-drain";
|
||||
await expect(detachNativeSessionsForRestart([restarting.binding.runId])).resolves.toMatchObject({ inactiveRunIds: [restarting.binding.runId] });
|
||||
state.execute.mockReset().mockImplementationOnce(async (options) => {
|
||||
await options.onSession({ detachControllerForRestart: detach });
|
||||
expect(detach).toHaveBeenCalledOnce();
|
||||
await options.onSession(null);
|
||||
throw new Error("detachment closed the old event stream");
|
||||
});
|
||||
state.execute.mockReset();
|
||||
await expect(executePaperclipNativeSession({ db: leaseDb(restarting), execution: restarting, runnerInstanceId: "runner" })).rejects.toBeInstanceOf(NativeControllerDetachedForRestartError);
|
||||
expect(state.execute).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -227,6 +227,7 @@ import {
|
||||
type ActiveNativeSession = {
|
||||
session: NativeSession;
|
||||
cancelRequested: boolean;
|
||||
cancellationSettled?: Promise<void>;
|
||||
restartDetach?: Promise<void>;
|
||||
};
|
||||
|
||||
@@ -264,6 +265,7 @@ type NativeSessionStartup = {
|
||||
promise: Promise<ActiveNativeSession | null>;
|
||||
resolve: (session: ActiveNativeSession | null) => void;
|
||||
stopRequested?: boolean;
|
||||
restartAdmissionPending?: boolean;
|
||||
cancellationSettled?: Promise<void>;
|
||||
governedWait?: boolean;
|
||||
settled: Promise<void>;
|
||||
@@ -272,13 +274,14 @@ type NativeSessionStartup = {
|
||||
const nativeSessionStartups = new Map<string, NativeSessionStartup>();
|
||||
|
||||
function detachActiveNativeSessionForRestart(active: ActiveNativeSession) {
|
||||
active.restartDetach ??= Promise.resolve().then(() => active.session.detachControllerForRestart!());
|
||||
active.restartDetach ??= active.session.detachControllerForRestart!();
|
||||
return active.restartDetach;
|
||||
}
|
||||
|
||||
async function waitForNativeSessionStartup(
|
||||
runId: string,
|
||||
timeoutError: () => Error = () => new Error("native_restart_startup_not_ready"),
|
||||
timeoutMs = 30_000,
|
||||
) {
|
||||
const startup = nativeSessionStartups.get(runId);
|
||||
if (!startup) return null;
|
||||
@@ -287,7 +290,7 @@ async function waitForNativeSessionStartup(
|
||||
return await Promise.race([
|
||||
startup.promise,
|
||||
new Promise<never>((_resolve, reject) => {
|
||||
timer = setTimeout(() => reject(timeoutError()), 30_000);
|
||||
timer = setTimeout(() => reject(timeoutError()), timeoutMs);
|
||||
}),
|
||||
]);
|
||||
} finally {
|
||||
@@ -301,10 +304,13 @@ export async function detachNativeSessionsForRestart(
|
||||
detachedRunIds: string[];
|
||||
inactiveRunIds: string[];
|
||||
unsupportedRunIds: string[];
|
||||
pendingRunIds: string[];
|
||||
}> {
|
||||
const detachedRunIds: string[] = [];
|
||||
const inactiveRunIds: string[] = [];
|
||||
const unsupportedRunIds: string[] = [];
|
||||
const pendingRunIds: string[] = [];
|
||||
const restartDeadline = Date.now() + 30_000;
|
||||
const settlementDeadline = Date.now() + 20_000;
|
||||
for (const runId of new Set(runIds)) {
|
||||
// A governed stop already revoked new work and is collecting terminal
|
||||
@@ -320,11 +326,48 @@ export async function detachNativeSessionsForRestart(
|
||||
} finally { clearTimeout(timer); }
|
||||
}
|
||||
nativeRunsDetachingForRestart.add(runId);
|
||||
const active = activeNativeSessions.get(runId) ?? await waitForNativeSessionStartup(runId);
|
||||
let active = activeNativeSessions.get(runId);
|
||||
if (!active) {
|
||||
// Only a startup registered before this fence may complete read-only
|
||||
// admission. It must publish its exact session for detachment; writes and
|
||||
// new turns remain revoked, including after the bounded wait expires.
|
||||
const startup = nativeSessionStartups.get(runId);
|
||||
if (startup && Date.now() >= restartDeadline) {
|
||||
pendingRunIds.push(runId);
|
||||
continue;
|
||||
}
|
||||
if (startup) startup.restartAdmissionPending = true;
|
||||
try {
|
||||
active = await waitForNativeSessionStartup(runId, undefined, Math.max(0, restartDeadline - Date.now())) ?? undefined;
|
||||
} catch (error) {
|
||||
if (!(error instanceof Error) || error.message !== "native_restart_startup_not_ready") throw error;
|
||||
pendingRunIds.push(runId);
|
||||
continue;
|
||||
} finally {
|
||||
if (startup) startup.restartAdmissionPending = false;
|
||||
}
|
||||
}
|
||||
if (!active) {
|
||||
inactiveRunIds.push(runId);
|
||||
continue;
|
||||
}
|
||||
const stopping = nativeSessionStartups.get(runId);
|
||||
if (stopping?.stopRequested || active.cancelRequested) {
|
||||
// Stop owns cancellation and final cleanup, including after publication.
|
||||
// A deadline leaves this run pending; it must not abort healthy adoption
|
||||
// or claim that the transport/cleanup owner has settled.
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
try {
|
||||
const settled = stopping && await Promise.race([
|
||||
stopping.settled.then(() => true),
|
||||
new Promise<false>(resolve => {
|
||||
timer = setTimeout(() => resolve(false), Math.max(0, restartDeadline - Date.now()));
|
||||
}),
|
||||
]);
|
||||
(settled ? inactiveRunIds : pendingRunIds).push(runId);
|
||||
} finally { clearTimeout(timer); }
|
||||
continue;
|
||||
}
|
||||
if (active.session.detachControllerForRestart === undefined) {
|
||||
unsupportedRunIds.push(runId);
|
||||
continue;
|
||||
@@ -332,7 +375,7 @@ export async function detachNativeSessionsForRestart(
|
||||
await detachActiveNativeSessionForRestart(active);
|
||||
detachedRunIds.push(runId);
|
||||
}
|
||||
return { detachedRunIds, inactiveRunIds, unsupportedRunIds };
|
||||
return { detachedRunIds, inactiveRunIds, unsupportedRunIds, pendingRunIds };
|
||||
}
|
||||
const MAX_REMOTE_CHECKPOINT_ARCHIVE_BYTES = 64 * 1024 * 1024;
|
||||
const MAX_REMOTE_CHECKPOINT_EXPANDED_BYTES = 64 * 1024 * 1024;
|
||||
@@ -7230,6 +7273,7 @@ export async function cancelNativeSession(
|
||||
// Do not acknowledge a no-op Stop and let that owner submit a turn afterwards.
|
||||
const startup = runStop && !activeNativeSessions.has(runId) ? nativeSessionStartups.get(runId) : undefined;
|
||||
let settleStartupCancellation: (() => void) | undefined;
|
||||
let settleActiveCancellation: (() => void) | undefined;
|
||||
if (startup) {
|
||||
startup.stopRequested = true;
|
||||
startup.cancellationSettled = new Promise<void>(resolve => { settleStartupCancellation = resolve; });
|
||||
@@ -7242,6 +7286,7 @@ export async function cancelNativeSession(
|
||||
dispatched = true;
|
||||
if (!active.cancelRequested) {
|
||||
active.cancelRequested = true;
|
||||
active.cancellationSettled = new Promise<void>(resolve => { settleActiveCancellation = resolve; });
|
||||
try {
|
||||
if (active.session.cancel) {
|
||||
const cancellationAbort = new AbortController();
|
||||
@@ -7451,6 +7496,7 @@ export async function cancelNativeSession(
|
||||
return { dispatched, decision, decisionId, auditId };
|
||||
} finally {
|
||||
settleStartupCancellation?.();
|
||||
settleActiveCancellation?.();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -7674,6 +7720,9 @@ export async function executePaperclipNativeSession(input: {
|
||||
}
|
||||
}
|
||||
const runId = input.execution.binding.runId;
|
||||
if (nativeRunsDetachingForRestart.has(runId)) {
|
||||
throw new NativeControllerDetachedForRestartError();
|
||||
}
|
||||
if (nativeSessionStartups.has(runId)) {
|
||||
throw new Error("native_session_supervisor_busy");
|
||||
}
|
||||
@@ -8362,6 +8411,12 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
throw new NativeControllerDetachedForRestartError();
|
||||
}
|
||||
};
|
||||
const startupForAdmission = nativeSessionStartups.get(input.execution.binding.runId);
|
||||
const assertRunAdmissionActive = () => {
|
||||
if (startupForAdmission?.restartAdmissionPending
|
||||
&& nativeSessionStartups.get(input.execution.binding.runId) === startupForAdmission) return;
|
||||
assertControllerActive();
|
||||
};
|
||||
const controlPlane = new PaperclipControlPlanePort(
|
||||
input.db,
|
||||
{
|
||||
@@ -8378,6 +8433,7 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
{
|
||||
privateKeyPem: input.runnerEnvironment?.PAPERCLIP_AGENT_PRIVATE_KEY,
|
||||
assertControllerActive,
|
||||
assertRunAdmissionActive,
|
||||
onCommittedEvent: async (event) => {
|
||||
const textKind = !firstTokenRecorded ? observeDeliverableText(event) : null;
|
||||
if (textKind) {
|
||||
@@ -9220,6 +9276,13 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
eq(heartbeatRuns.agentId, input.execution.binding.agentId),
|
||||
)).limit(1);
|
||||
const cancellation = record(currentRun?.resultJson?.nativeCancellation);
|
||||
if (active.cancelRequested) {
|
||||
// A Stop can acquire the published handle while the authority
|
||||
// read above is in flight. Keep its ACK/cleanup owner before
|
||||
// unwinding admission; neither detach nor generic close wins.
|
||||
await active.cancellationSettled;
|
||||
throw new NativeCancellationPendingRecoveryError();
|
||||
}
|
||||
if (!currentRun || currentRun.status !== "running" ||
|
||||
currentRun.resultJson?.startupCancellation ||
|
||||
(cancellation.scope === "run" &&
|
||||
@@ -9229,11 +9292,15 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
// callback fails; no provider turn may follow publication.
|
||||
throw new NativeCancellationPendingRecoveryError();
|
||||
}
|
||||
if (session.resolveRuntimeRequest) await liveQuestions.attach();
|
||||
if (nativeRunsDetachingForRestart.has(input.execution.binding.runId)) {
|
||||
if (session.detachControllerForRestart) await detachActiveNativeSessionForRestart(active);
|
||||
throw new NativeControllerDetachedForRestartError();
|
||||
}
|
||||
if (session.resolveRuntimeRequest) await liveQuestions.attach();
|
||||
if (active.cancelRequested) {
|
||||
await active.cancellationSettled;
|
||||
throw new NativeCancellationPendingRecoveryError();
|
||||
}
|
||||
if (active.cancelRequested) throw new NativeCancellationPendingRecoveryError();
|
||||
} else {
|
||||
liveQuestions.close();
|
||||
activeNativeSessions.delete(input.execution.binding.runId);
|
||||
|
||||
@@ -300,6 +300,38 @@ describe("PaperclipControlPlanePort conformance", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("permits only read-only admission while an existing startup drains", async () => {
|
||||
const identity = { ...CONTROL_PLANE_CONFORMANCE_OPEN.identity, runId: randomUUID(), sessionId: randomUUID() };
|
||||
const runnerId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: identity.runId, companyId: identity.companyId, agentId: identity.agentId,
|
||||
status: "running", runtimeMode: "native", nativeIssueId: identity.issueId,
|
||||
nativeSessionId: identity.sessionId, runnerInstanceId: runnerId,
|
||||
completionContractId: contractId, completionContractSha256: contractSha,
|
||||
});
|
||||
let awaitingStartup = true;
|
||||
const port = new PaperclipControlPlanePort(db, {
|
||||
...identity, completionContractId: contractId, completionContractSha256: contractSha,
|
||||
sourceInstanceId: runnerId, controlPlaneSourceInstanceId: `draining-${identity.runId}`,
|
||||
}, {
|
||||
assertControllerActive: () => { throw new Error("controller_detached"); },
|
||||
assertRunAdmissionActive: () => { if (!awaitingStartup) throw new Error("controller_detached"); },
|
||||
});
|
||||
const open = { identity, backendKind: "mock" as const, sourceInstanceId: runnerId };
|
||||
await expect(port.openRun({ ...open, sourceInstanceId: randomUUID() })).rejects.toThrow("native_open_run_binding_mismatch");
|
||||
await db.update(heartbeatRuns).set({ runnerInstanceId: randomUUID() }).where(eq(heartbeatRuns.id, identity.runId));
|
||||
await expect(port.openRun(open)).rejects.toThrow("native_open_run_not_authorized");
|
||||
await db.update(heartbeatRuns).set({ runnerInstanceId: runnerId }).where(eq(heartbeatRuns.id, identity.runId));
|
||||
await expect(port.openRun(open)).resolves.toBeUndefined();
|
||||
await expect(port.checkpointSession({ backendKind: "mock", sessionId: identity.sessionId, identity })).rejects.toThrow("controller_detached");
|
||||
await expect(port.completeRun({ result: CONTROL_PLANE_CONFORMANCE_RESULT, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL })).rejects.toThrow("controller_detached");
|
||||
awaitingStartup = false;
|
||||
await expect(port.openRun(open)).rejects.toThrow("controller_detached");
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, identity.runId));
|
||||
expect(run.runnerProfileJson?.sessionCheckpoint).toBeUndefined();
|
||||
expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, identity.runId))).toEqual([]);
|
||||
});
|
||||
|
||||
it.each([
|
||||
["checkpoint", "detach"], ["result", "detach"], ["event", "detach"],
|
||||
["checkpoint", "abort"], ["result", "abort"], ["event", "abort"],
|
||||
|
||||
@@ -83,6 +83,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
|
||||
readonly #binding: PaperclipControlPlaneBinding;
|
||||
#sessionId: string | null = null;
|
||||
readonly #assertControllerActive?: () => void;
|
||||
readonly #assertRunAdmissionActive?: () => void;
|
||||
readonly #onCommittedEvent?: (event: PrpEvent) => Promise<void>;
|
||||
readonly #onDuplicateEvent?: (event: PrpEvent) => Promise<void>;
|
||||
|
||||
@@ -94,12 +95,15 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
|
||||
privateKeyPem?: string;
|
||||
/** Synchronous revocation fence for a controller handing off on restart. */
|
||||
assertControllerActive?: () => void;
|
||||
/** Read-only admission may finish an existing startup solely for restart detachment. */
|
||||
assertRunAdmissionActive?: () => void;
|
||||
onCommittedEvent?: (event: PrpEvent) => Promise<void>;
|
||||
onDuplicateEvent?: (event: PrpEvent) => Promise<void>;
|
||||
} = {},
|
||||
) {
|
||||
this.#db = db;
|
||||
this.#assertControllerActive = options.assertControllerActive;
|
||||
this.#assertRunAdmissionActive = options.assertRunAdmissionActive ?? options.assertControllerActive;
|
||||
this.#identityRedactor = createAgentIdentityRedactor(options.privateKeyPem);
|
||||
this.#binding = structuredClone(binding);
|
||||
this.#onCommittedEvent = options.onCommittedEvent;
|
||||
@@ -123,7 +127,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
|
||||
}
|
||||
|
||||
async openRun(input: OpenControlPlaneRunInput): Promise<void> {
|
||||
this.#assertControllerActive?.();
|
||||
this.#assertRunAdmissionActive?.();
|
||||
const identity = input.identity;
|
||||
if (
|
||||
identity.companyId !== this.#binding.companyId
|
||||
@@ -146,7 +150,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
|
||||
if (!run || !this.#matchesPersistedBinding(run)) {
|
||||
throw new Error("native_open_run_not_authorized");
|
||||
}
|
||||
this.#assertControllerActive?.();
|
||||
this.#assertRunAdmissionActive?.();
|
||||
this.#sessionId = identity.sessionId;
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user