mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-13 07:12:00 +01:00
fix(native): continue a governed-wait follow-up in a fresh session when its saved provider state is gone (#20809)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The restart chaos harness (`tests/restart-chaos/`) runs real agent tasks and breaks the control plane on purpose, so failures show up before staging users see them > - A census of 48 hours of staging failures found runs whose result save-back failed because the sandbox disk was full (class F). The harness never filled a sandbox disk, so it missed that class > - #20758 made a full-disk save end recovery at once with cause `sandbox_disk_full` and a Retry. Nothing in chaos exercised that path end to end > - A real full disk cannot be made on demand. The owner chose a test-only hook that makes the save report a full disk, not a small disk > - This pull request adds the hook at the save's sandbox boundary and a DF1 fault that arms it, clears it, and grades the result > - The benefit is that every wave can now check the disk-full rest, the Retry offer, and that Retry works after space is freed ## Linked Issues or Issue Description Refs #20758 (product behaviour under test), #20678 (finalization hands a full-disk copy-back to recovery) **What is the problem?** The chaos harness had no fault for a full sandbox disk during result save-back. Staging saw this failure (census class F in `doc/plans/2026-11-11-staging-failures-vs-chaos-coverage.md`), and chaos could not reproduce it. **What is the expected behaviour?** A wave can schedule DF1. While DF1 is armed, the task's final workspace saves fail as a full disk. When DF1 is cleared, saves succeed again. The task must rest with `sandbox_disk_full` and a Retry, and that Retry must finish the task. **Why does it matter?** Without a fault, a regression in the disk-full path (no Retry, a wrong cause, or a Retry that does not work) reaches staging users before chaos sees it. ## What Changed - `server/src/services/sandbox-disk-full-fault.ts` (new, test-only): `sandboxDiskFullFaultTarget` wraps the target of a final save only when the harness armed that save. The capture commands then exit 1 with the exact message the in-sandbox script writes on ENOSPC, and the provider download fails with `No space left on device`. The gate follows `@paperclipai/shared/crash-points`: `PAPERCLIP_DISK_FULL_FAULT_CONTROL` is read once at import and ignored under `NODE_ENV=production`. Disarmed, the hook is one comparison and returns the save's own target. - `server/src/services/remote-workspaces.ts`: the capture call in `saveRemoteWorkspace` passes its target through the hook. No other product change. `workspace-recovery.ts` is not touched. - `tests/restart-chaos/disk-full-faults.ts` (new): the DF1 kind, the control-file seam (atomic writes), the rest evidence (`activeRecoveryAction.evidence.systemRecovery.failure.cause.label`), the `disk_full_fault` T check, and the DF1 R3 rule (one expected Retry that must finish the task). - Harness wiring: `--faults DF1` (refused without `--env daytona`, with a storm, or with PV faults), no scheduled restart, arm at launch, clear at the first rest just before the Retry click, receipts in `<wave>/disk-full-faults.jsonl`, and ledger fields `diskFull: { faultedSaves, restedDiskFull, retryOffered, retryWorked }`. - Docs: a "Sandbox disk full (DF1)" section in `tests/restart-chaos/README.md`, and the census plan marks DF1 as built. ## Verification - `cd server && pnpm exec vitest run src/services/sandbox-disk-full-fault.test.ts`: the gate is off by default and under `NODE_ENV=production`, the save's own target comes back when disarmed, only a listed final save is faulted, the product capture turns it into the named disk-full failure, and clearing the file disarms the next save. - `pnpm exec vitest run --config tests/restart-chaos/vitest.config.ts tests/restart-chaos/disk-full-faults.test.ts tests/restart-chaos/grader.test.ts tests/restart-chaos/scheduler.test.ts tests/restart-chaos/provider-faults.test.ts`: planning, the control file, and grading. - `node scripts/typecheck-test-projects.mjs tests/restart-chaos` and the server `tsc --noEmit`: clean. - Live wave: a Daytona DF1 wave is running. Its result is added here before merge. ## Risks - Low risk for production. The hook reads its variable once at startup and is ignored under `NODE_ENV=production`. Without the variable it does one comparison and returns the same target object. - The hook fails every sandbox command of an armed capture, including the read-only fallback's inventory. A real full disk can let that inventory succeed and fail later at the download. The outcome is the same (`sandbox_disk_full`), but the fallback's inventory step is not exercised under DF1. - DF1 runs only with `--env daytona`. A local run has no sandbox workspace save, so the harness refuses DF1 there. ## Model Used - Claude Opus 5.5 (Anthropic, model ID `claude-opus-5-5`), in Claude Code with tool use (shell, file edits, GitHub CLI). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
DottaandClaude authored and GitHub committed
1 parent
5f38226290
commit
fd1460bb24
11 files changed
+477
-6
No files matched your search
@@ -3029,6 +3029,24 @@ ends with `native_provider_state_incompatible`, and Retry or the next message
|
||||
continues in a fresh provider session in the same workspace. A session with
|
||||
work in flight keeps the unclassified behavior.
|
||||
|
||||
A follow-up after a governed wait (`allow_replacement_after_governed_wait`)
|
||||
normally requires the same provider session. For this one refusal it continues
|
||||
on its own in a fresh provider session in the same run, without a Retry, when
|
||||
all of these hold: the driver reports the Runner's typed missing-state refusal
|
||||
for a session with nothing in flight; the reconciled checkpoint is the untouched
|
||||
rebind of the previous run (no cursor progress, turn, runtime request, proposed
|
||||
result, terminal, or governed wait of the follow-up run, and the at-most-once
|
||||
recovery marker present); and the control plane proves from durable rows,
|
||||
across every runner instance, that the follow-up run has no provider event,
|
||||
source-sequenced runner event, tool invocation, or accepted result. The follow-up
|
||||
has not started a turn, so a fresh session cannot repeat provider work or a tool
|
||||
side effect; only the earlier conversation context is lost. The fresh session
|
||||
gets the full task prompt, the interaction responses, the stored fresh-session
|
||||
handoff, and queued input, as any replacement does, and the run log records the
|
||||
existing provider session continuity break with the reason. Any other
|
||||
incompatible state, a control plane without that proof, or a checkpoint saved
|
||||
without those fields keeps `native_provider_state_incompatible` and its Retry.
|
||||
|
||||
Bootstrap retries, exact-checkpoint resumes, and fresh replacement sessions share three total provider attempts, including the original attempt. Linked run IDs, controller restarts, and duplicate wakes do not reset this budget. A resume after a graceful shutdown drain suspended the run for reattachment does not consume an attempt, because the run did not cause that restart. A crash restart (no drain record) consumes an attempt only when the attempt it interrupted spent provider work without progress. This is one rule over recorded outcomes, not over intents. The attempt counts as started only when the provider accepted a turn: a `turn.accepted` event or the provider's own `turn.started`. A startup still marked `beforeSpawn`, a runner that spawned but was never admitted, and a `turn.start` the provider did not accept are not charged. Progress means that an accepted turn then completed work: a tool execution or a completed item other than a usage report. An accepted turn with only usage, a failure, or no events is charged, so a resume that fails each time still uses up the budget. The first restart judges the original execution. Each later restart claim records the run's event-log position, and the next claim judges only the attempt recorded after it; a normal retry in between starts a new attempt (at its `native.process_start_requested`), so earlier work does not count for it. A reattachment adopts a provider that already started, so for it only progress counts. A resume that launches a new runner and that the controller died before starting used no provider work, so the next claim replaces it and the attempt ends at the attempt that resume was to run as. Uncharged restarts stop after 12 controller generations, so a crash loop that never starts the provider, or that makes progress and then crashes the controller each time, is still bounded. Legacy runs follow the same rule: a preparation interrupted before provider work started (`providerWorkStarted: false`) is retried in its own `interrupted_preparation` lane, which never spends the failure retry budget and is bounded at 12 retries. This holds for a hard kill as well as a graceful drain: once the killed controller's recorded process is proven gone, restart recovery fails the run as `process_lost` with the bootstrap certificate and schedules it in that lane. A run saved by an older release without a controller process stamp cannot be proven and stays in the charged lane. Exact-checkpoint resumes retain the 30-second delay. Replacement scheduling and predecessor lineage commit together, with one successor per predecessor and admission through the normal task locks, authorization, pause, approval, and budget gates.
|
||||
|
||||
A due failed remote retry can retain its provider PID after heartbeat completion
|
||||
|
||||
@@ -167,6 +167,7 @@ export class HarnessDriverBackend implements NativeSessionBackend {
|
||||
recovered: false,
|
||||
reason: recovered.reason,
|
||||
...(recovered.failureClass ? { failureClass: recovered.failureClass } : {}),
|
||||
...(recovered.resumeStateUnavailable ? { resumeStateUnavailable: true as const } : {}),
|
||||
};
|
||||
}
|
||||
try {
|
||||
|
||||
@@ -488,6 +488,7 @@ describe("native backend factory", () => {
|
||||
recovered: false,
|
||||
reason: expect.stringContaining("PRP provider resume state is unavailable"),
|
||||
failureClass: PROVIDER_STATE_INCOMPATIBLE,
|
||||
resumeStateUnavailable: true,
|
||||
});
|
||||
});
|
||||
|
||||
@@ -499,6 +500,7 @@ describe("native backend factory", () => {
|
||||
const recovered = await backend.recoverSession!({ ...snapshot, ...inFlight } as never, { signal: new AbortController().signal });
|
||||
expect(recovered).toMatchObject({ recovered: false, reason: expect.stringContaining("PRP provider resume state is unavailable") });
|
||||
expect(recovered).not.toHaveProperty("failureClass");
|
||||
expect(recovered).not.toHaveProperty("resumeStateUnavailable");
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -99,4 +99,11 @@ export interface ControlPlanePort {
|
||||
result: NativeRunResult | CompleteControlPlaneRunInput,
|
||||
options?: FinalizeControlPlaneOperationOptions,
|
||||
): Promise<void>;
|
||||
/**
|
||||
* Durable proof, across every runner instance, that this run has recorded no
|
||||
* provider event, tool invocation, or accepted result. True only when the
|
||||
* control plane can prove it; a port without this proof keeps every
|
||||
* recovery refusal unchanged.
|
||||
*/
|
||||
provesNoRunProviderWork?(options: FinalizeControlPlaneOperationOptions): Promise<boolean>;
|
||||
}
|
||||
@@ -557,6 +557,10 @@ export interface HarnessSessionRecoveryResult {
|
||||
reason?: string;
|
||||
/** Closed runner class of a deterministic recovery failure. */
|
||||
failureClass?: RunnerRecoveryFailureClass;
|
||||
/** Set only with PROVIDER_STATE_INCOMPATIBLE: the Runner proved it holds no
|
||||
* saved state for this session (NativeRunnerResumeStateUnavailableError) and
|
||||
* the checkpoint had no turn or runtime request in flight. */
|
||||
resumeStateUnavailable?: true;
|
||||
}
|
||||
|
||||
/** Reads the closed class of an authenticated runner command failure, or of a
|
||||
|
||||
@@ -86,6 +86,7 @@ export interface NativeSessionRecoveryResult {
|
||||
session?: NativeSession;
|
||||
reason?: string;
|
||||
failureClass?: HarnessSessionRecoveryResult["failureClass"];
|
||||
resumeStateUnavailable?: HarnessSessionRecoveryResult["resumeStateUnavailable"];
|
||||
}
|
||||
|
||||
/** Runner-owned cause: the process was restored, but its active turn was lost. */
|
||||
|
||||
@@ -773,13 +773,15 @@ export class CodexAppServerDriver implements HarnessDriver {
|
||||
// resume it here. With nothing in flight, the runtime's recovery policy
|
||||
// decides, as for a moved workspace root. An in-flight session keeps the
|
||||
// unclassified failure.
|
||||
const failureClass = runnerRecoveryFailureClass(error)
|
||||
?? (isRunnerResumeStateUnavailable(error) && !persistedSessionInFlight(snapshot)
|
||||
? PROVIDER_STATE_INCOMPATIBLE : undefined);
|
||||
const classified = runnerRecoveryFailureClass(error);
|
||||
const resumeStateUnavailable = !classified && isRunnerResumeStateUnavailable(error)
|
||||
&& !persistedSessionInFlight(snapshot);
|
||||
const failureClass = classified ?? (resumeStateUnavailable ? PROVIDER_STATE_INCOMPATIBLE : undefined);
|
||||
return {
|
||||
recovered: false,
|
||||
reason: redactCodexDiagnostic(String(error)),
|
||||
...(failureClass ? { failureClass } : {}),
|
||||
...(resumeStateUnavailable ? { resumeStateUnavailable: true as const } : {}),
|
||||
};
|
||||
} finally {
|
||||
cancellation.detach();
|
||||
|
||||
@@ -17,6 +17,7 @@ import type { NativeRunIdentity } from "./contracts/types.js";
|
||||
import type {
|
||||
NativeSession,
|
||||
NativeSessionBackend,
|
||||
NativeSessionRecoveryResult,
|
||||
PersistedNativeSession,
|
||||
} from "./contracts/native-session-backend.js";
|
||||
import {
|
||||
@@ -6664,6 +6665,171 @@ describe("executeNativeSession recovery", () => {
|
||||
}
|
||||
});
|
||||
|
||||
// Staging class P after #20757: a follow-up after a governed (approval) wait
|
||||
// rebinds the previous run's session, but the new sandbox holds no saved
|
||||
// Runner state for it. The follow-up has not started a turn, so a fresh
|
||||
// provider session cannot repeat provider work or a tool side effect; only
|
||||
// the earlier conversation context is lost. The run continues in a fresh
|
||||
// session with the full task, the answer and the stored handoff.
|
||||
describe("a governed-wait follow-up whose saved provider state is gone", () => {
|
||||
const approval = { interactionId: "interaction-approval", kind: "request_confirmation" as const,
|
||||
response: { outcome: "accepted", note: "APPROVED_BY_BOARD" } };
|
||||
const followUpInput = {
|
||||
...input,
|
||||
task: { ...input.task, prompt: "FULL_TASK_CONTEXT\nAPPROVAL_FOLLOW_UP" },
|
||||
interactionResponses: [approval],
|
||||
};
|
||||
const reboundCheckpoint = (): PersistedNativeSession => ({
|
||||
backendKind: "mock", driverKind: "codex_app_server", sessionId: "driver-lost", identity,
|
||||
providerSessionId: "provider-lost", providerRecoveryPolicy: "allow_replacement_after_governed_wait",
|
||||
cursor: null, semanticResult: null, terminal: null, activeTurnId: null, terminalTurns: [],
|
||||
pendingRuntimeRequests: [], dispositionOnlyRecoveryConsumed: false, dispositionOnlyRecoveryTurnId: null,
|
||||
lineage: [],
|
||||
});
|
||||
const stateUnavailable: NativeSessionRecoveryResult = {
|
||||
recovered: false, failureClass: PROVIDER_STATE_INCOMPATIBLE, resumeStateUnavailable: true,
|
||||
reason: "Error: NativeRunnerResumeStateUnavailableError: PRP provider resume state is unavailable",
|
||||
};
|
||||
function harness(options: {
|
||||
checkpoint?: PersistedNativeSession;
|
||||
recovery?: Awaited<ReturnType<NonNullable<NativeSessionBackend["recoverSession"]>>>;
|
||||
proof?: boolean | null | "rejects" | "hangs";
|
||||
replay?: PrpEvent[];
|
||||
} = {}) {
|
||||
const startTurn = vi.fn<NativeSession["startTurn"]>(async () => ({ turnId: "turn-fresh" }));
|
||||
const replacementSnapshot: PersistedNativeSession = {
|
||||
...reboundCheckpoint(), sessionId: "driver-fresh", providerSessionId: "provider-fresh",
|
||||
};
|
||||
const replacementSession: NativeSession = {
|
||||
identity: () => identity,
|
||||
capabilities: async () => ({ resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }),
|
||||
async *events() { yield runnerEvent(1, "turn.completed"); },
|
||||
startTurn,
|
||||
result: async () => ({ result, terminal, turnId: "turn-fresh" }),
|
||||
snapshot: async () => structuredClone(replacementSnapshot),
|
||||
close: async () => {},
|
||||
};
|
||||
const openReplacementSession = vi.fn(async () => replacementSession);
|
||||
const openSession = vi.fn(async () => { throw new Error("replacement seam must be used"); });
|
||||
const recoverSession = vi.fn(async () => options.recovery ?? stateUnavailable);
|
||||
const backend: NativeSessionBackend = {
|
||||
descriptor: async () => ({ kind: "mock", name: "governed-follow-up", version: "1",
|
||||
capabilities: { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true } }),
|
||||
openSession, openReplacementSession, recoverSession,
|
||||
};
|
||||
const events: PrpEvent[] = [];
|
||||
const provesNoRunProviderWork = vi.fn(async ({ signal }: { signal: AbortSignal }) => {
|
||||
if (options.proof === "rejects") throw new Error("database unavailable");
|
||||
if (options.proof === "hangs") return await new Promise<boolean>((_resolve, reject) =>
|
||||
signal.addEventListener("abort", () => reject(signal.reason), { once: true }));
|
||||
return options.proof === true;
|
||||
});
|
||||
const openRun = vi.fn(async () => undefined);
|
||||
const replay = options.replay ?? [];
|
||||
const controlPlane: ControlPlanePort = {
|
||||
openRun,
|
||||
loadSessionCheckpoint: async () => structuredClone(options.checkpoint ?? reboundCheckpoint()),
|
||||
checkpointSession: async () => {},
|
||||
async appendEvent(event) {
|
||||
events.push(structuredClone(event as PrpEvent));
|
||||
return { cursor: events.length, highestContiguousSourceSeq: highestContiguous(events), disposition: "committed" };
|
||||
},
|
||||
replayEvents: async ({ afterSourceSeq }) => {
|
||||
const page = replay.filter(event => event.sourceSeq > afterSourceSeq);
|
||||
return { events: page, highestContiguousSourceSeq: highestContiguous(page) };
|
||||
},
|
||||
completeRun: async () => {},
|
||||
...(options.proof === null ? {} : { provesNoRunProviderWork }),
|
||||
};
|
||||
const onContinuityBreak = vi.fn(async () => undefined);
|
||||
const getFreshSessionHandoff = vi.fn(async () => "FRESH_HANDOFF: prior conversation summary");
|
||||
const run = () => executeNativeSession({
|
||||
input: followUpInput, backend, controlPlane, getFreshSessionHandoff, onContinuityBreak,
|
||||
runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery",
|
||||
});
|
||||
return { run, startTurn, openReplacementSession, openSession, recoverSession, provesNoRunProviderWork,
|
||||
openRun, onContinuityBreak, getFreshSessionHandoff };
|
||||
}
|
||||
|
||||
it("continues in a fresh provider session with the task, the answer and the stored handoff", async () => {
|
||||
const h = harness({ proof: true });
|
||||
await expect(h.run()).resolves.toMatchObject({ providerSessionId: "provider-fresh" });
|
||||
expect(h.recoverSession).toHaveBeenCalledOnce();
|
||||
expect(h.provesNoRunProviderWork).toHaveBeenCalledOnce();
|
||||
expect(h.openReplacementSession).toHaveBeenCalledOnce();
|
||||
expect(h.openSession).not.toHaveBeenCalled();
|
||||
expect(h.startTurn).toHaveBeenCalledOnce();
|
||||
const envelope = JSON.parse(h.startTurn.mock.calls[0]![0].message.text) as {
|
||||
task: { prompt: string }; interactionResponses: unknown[];
|
||||
};
|
||||
expect(envelope.task.prompt).toBe("FRESH_HANDOFF: prior conversation summary\n\nFULL_TASK_CONTEXT\nAPPROVAL_FOLLOW_UP");
|
||||
expect(envelope.interactionResponses).toEqual([approval]);
|
||||
expect(h.startTurn.mock.calls[0]![0]).not.toHaveProperty("continuation");
|
||||
// The user sees that the earlier conversation context was reset.
|
||||
expect(h.onContinuityBreak).toHaveBeenCalledWith({
|
||||
reason: expect.stringContaining("saved provider state is unavailable, so the follow-up continues in a fresh provider session without the earlier conversation context"),
|
||||
previousDriverSessionId: "driver-lost", previousProviderSessionId: "provider-lost",
|
||||
replacementDriverSessionId: "driver-fresh", replacementProviderSessionId: "provider-fresh",
|
||||
});
|
||||
});
|
||||
|
||||
it.each([
|
||||
["the control plane cannot prove the run started no provider work", { proof: false }],
|
||||
["the control plane predates the provider-work proof", { proof: null }],
|
||||
["the provider-work proof fails", { proof: "rejects" }],
|
||||
["the saved state is incompatible for another reason", { proof: true, recovery: {
|
||||
recovered: false as const, failureClass: PROVIDER_STATE_INCOMPATIBLE, reason: "saved launch profile changed" } }],
|
||||
["the checkpoint requires the same session", { proof: true,
|
||||
checkpoint: { ...reboundCheckpoint(), providerRecoveryPolicy: "same_session_only" as const } }],
|
||||
["this run already completed a turn", { proof: true,
|
||||
checkpoint: { ...reboundCheckpoint(), cursor: "2", terminalTurns: [{ turnId: "turn-own", fingerprint: "completed" }] } }],
|
||||
["this run already advanced the durable cursor", { proof: true, checkpoint: { ...reboundCheckpoint(), cursor: "3" } }],
|
||||
["this run already recorded a provider event", { proof: true, replay: [runnerEvent(1, "turn.started")] }],
|
||||
["this run already proposed a result", { proof: true, checkpoint: { ...reboundCheckpoint(), semanticResult: result } }],
|
||||
["the checkpoint was saved without the at-most-once recovery marker", { proof: true,
|
||||
checkpoint: (() => { const { dispositionOnlyRecoveryConsumed: _omitted, ...older } = reboundCheckpoint(); return older; })() }],
|
||||
] as const)("keeps the named refusal and a working Retry when %s", async (_label, options) => {
|
||||
const h = harness(options as Parameters<typeof harness>[0]);
|
||||
const failure = await h.run().catch((error: unknown) => error);
|
||||
expect(failure).toBeInstanceOf(NativeProviderStateIncompatibleError);
|
||||
expect((failure as NativeProviderStateIncompatibleError).code).toBe("native_provider_state_incompatible");
|
||||
expect(h.openReplacementSession).not.toHaveBeenCalled();
|
||||
expect(h.openSession).not.toHaveBeenCalled();
|
||||
expect(h.openRun).not.toHaveBeenCalled();
|
||||
expect(h.startTurn).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("never replaces a session whose checkpoint records this run's own governed wait", async () => {
|
||||
const h = harness({ proof: true, checkpoint: { ...reboundCheckpoint(),
|
||||
governedWait: { sourceEvent: runnerEvent(1, "turn.started"), result: yieldedResult } } });
|
||||
await expect(h.run()).rejects.toThrow();
|
||||
expect(h.openReplacementSession).not.toHaveBeenCalled();
|
||||
expect(h.startTurn).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("keeps the named refusal when the provider-work proof does not answer in time", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const h = harness({ proof: "hangs" });
|
||||
const failure = h.run().catch((error: unknown) => error);
|
||||
await vi.advanceTimersByTimeAsync(30_001);
|
||||
expect(await failure).toBeInstanceOf(NativeProviderStateIncompatibleError);
|
||||
expect(h.provesNoRunProviderWork).toHaveBeenCalledOnce();
|
||||
expect(h.openReplacementSession).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("stops by name when the follow-up's saved session still owns a runtime request", async () => {
|
||||
const h = harness({ proof: true, recovery: { recovered: false as const, failureClass: PROVIDER_SIDE_EFFECT_PENDING,
|
||||
reason: "native_runner_source_runtime_reconciliation_required" } });
|
||||
await expect(h.run()).rejects.toBeInstanceOf(NativeProviderSideEffectPendingError);
|
||||
expect(h.provesNoRunProviderWork).not.toHaveBeenCalled();
|
||||
expect(h.openReplacementSession).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
// A remote workspace epoch was abandoned, so the run was assigned a new
|
||||
// sibling root while its checkpoint still names the old one. A checkpoint
|
||||
// that requires the same session (an unfinished goal) ends with the
|
||||
|
||||
@@ -18,7 +18,7 @@ import type {
|
||||
NativeSession,
|
||||
NativeSessionBackend,
|
||||
} from "./contracts/native-session-backend.js";
|
||||
import type { PersistedNativeSession } from "./contracts/native-session-backend.js";
|
||||
import type { NativeSessionRecoveryResult, PersistedNativeSession } from "./contracts/native-session-backend.js";
|
||||
import {
|
||||
PROVIDER_ATTACH_REFUSED,
|
||||
PROVIDER_SIDE_EFFECT_PENDING,
|
||||
@@ -1917,6 +1917,61 @@ function checkpointCursor(cursor: string | null | undefined): number {
|
||||
return Number.isSafeInteger(parsed) && parsed >= 0 ? parsed : 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* A follow-up after a governed wait normally requires the same provider
|
||||
* session. When the Runner proved the saved session state is gone, and nothing
|
||||
* proves this run started provider work of its own, a fresh session cannot
|
||||
* repeat a turn or a tool side effect: only the earlier conversation context is
|
||||
* lost. Every other case keeps the named refusal. The reconciled checkpoint must
|
||||
* be the untouched rebind of the previous run (no cursor progress, turn,
|
||||
* runtime request, result, terminal or governed wait of this run), and the
|
||||
* control plane must prove durably, across runner instances, that the run has
|
||||
* no provider event, tool invocation or accepted result. A control plane
|
||||
* without that proof, or a checkpoint missing those fields, keeps the refusal.
|
||||
*/
|
||||
const GOVERNED_WAIT_PROVIDER_WORK_PROOF_TIMEOUT_MS = 30_000;
|
||||
|
||||
async function governedWaitFollowUpMayStartFresh(input: {
|
||||
input: NativeExecutionInput;
|
||||
checkpoint: PersistedNativeSession;
|
||||
recovery: NativeSessionRecoveryResult;
|
||||
restartInterruption: boolean;
|
||||
controlPlane: ControlPlanePort;
|
||||
timeoutMs: number;
|
||||
}): Promise<boolean> {
|
||||
const { checkpoint, recovery } = input;
|
||||
if (
|
||||
checkpoint.providerRecoveryPolicy !== "allow_replacement_after_governed_wait" ||
|
||||
("homeMode" in input.input && input.input.homeMode === "preconfigured") ||
|
||||
recovery.failureClass !== PROVIDER_STATE_INCOMPATIBLE ||
|
||||
recovery.resumeStateUnavailable !== true ||
|
||||
input.restartInterruption ||
|
||||
checkpoint.identity.runId !== input.input.binding.runId ||
|
||||
!(checkpoint.cursor === null || checkpoint.cursor === "0") ||
|
||||
checkpoint.activeTurnId !== null ||
|
||||
!Array.isArray(checkpoint.terminalTurns) || checkpoint.terminalTurns.length > 0 ||
|
||||
!Array.isArray(checkpoint.pendingRuntimeRequests) || checkpoint.pendingRuntimeRequests.length > 0 ||
|
||||
checkpoint.semanticResult !== null ||
|
||||
checkpoint.terminal !== null ||
|
||||
checkpoint.governedWait !== undefined ||
|
||||
checkpoint.dispositionOnlyRecoveryConsumed !== false ||
|
||||
!input.controlPlane.provesNoRunProviderWork
|
||||
) return false;
|
||||
const controlPlane = input.controlPlane;
|
||||
// A proof that fails or expires is no proof: keep the named refusal and its
|
||||
// Retry rather than surfacing a generic interruption.
|
||||
const timeoutMs = Math.min(input.timeoutMs, GOVERNED_WAIT_PROVIDER_WORK_PROOF_TIMEOUT_MS);
|
||||
try {
|
||||
return await runAbortableOperationWithin({
|
||||
timeoutMs,
|
||||
timeoutMessage: `native session provider-work proof timed out after ${timeoutMs}ms`,
|
||||
operation: (signal) => controlPlane.provesNoRunProviderWork!({ signal }),
|
||||
}) === true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
async function reconcileRecoveryCheckpoint(input: {
|
||||
controlPlane: ControlPlanePort;
|
||||
checkpoint: PersistedNativeSession;
|
||||
@@ -2520,7 +2575,15 @@ export async function executeNativeSession(
|
||||
if (recovery.failureClass === PROVIDER_ATTACH_REFUSED) {
|
||||
throw new NativeProviderStateIncompatibleError(recovery.reason ?? "unknown");
|
||||
}
|
||||
if (!replacementAllowed) {
|
||||
const governedWaitFollowUpLostState = !replacementAllowed && await governedWaitFollowUpMayStartFresh({
|
||||
input,
|
||||
checkpoint: providerRecoveryCheckpoint,
|
||||
recovery,
|
||||
restartInterruption: Boolean(checkpointedInterruption),
|
||||
controlPlane: options.controlPlane,
|
||||
timeoutMs: recoveryTimeoutMs,
|
||||
});
|
||||
if (!replacementAllowed && !governedWaitFollowUpLostState) {
|
||||
if (recovery.failureClass === PROVIDER_STATE_INCOMPATIBLE) {
|
||||
throw new NativeProviderStateIncompatibleError(recovery.reason ?? "unknown");
|
||||
}
|
||||
@@ -2529,7 +2592,9 @@ export async function executeNativeSession(
|
||||
);
|
||||
}
|
||||
continuityBreak = {
|
||||
reason: recovery.reason ?? "provider session is no longer recoverable",
|
||||
reason: governedWaitFollowUpLostState
|
||||
? `saved provider state is unavailable, so the follow-up continues in a fresh provider session without the earlier conversation context (${recovery.reason ?? "unknown"})`
|
||||
: recovery.reason ?? "provider session is no longer recoverable",
|
||||
previousDriverSessionId: providerRecoveryCheckpoint.sessionId,
|
||||
previousProviderSessionId:
|
||||
providerRecoveryCheckpoint.providerSessionId ?? null,
|
||||
|
||||
+175
@@ -0,0 +1,175 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { mkdtemp, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { afterEach, expect, it, vi } from "vitest";
|
||||
import { completionContracts, heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, toolInvocations } from "@paperclipai/db";
|
||||
import { describeEmbeddedPostgres, nativeRestartFixture } from "../../__tests__/helpers/native-restart.js";
|
||||
import { DurablePrpControlPlane, parseNativeExecutionInput, type NativeSessionBackend, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js";
|
||||
import { createNativeControllerRecovery } from "./native-controller-recovery.js";
|
||||
import { nativeRunnerCheckpointWriter } from "./native-runner-checkpoint.js";
|
||||
import { executePaperclipNativeSession, nativeControllerMigrationDescriptor } from "./native-session-executor.js";
|
||||
import { PaperclipControlPlanePort } from "./paperclip-control-plane-port.js";
|
||||
import { nativeToolContractFingerprintForTarget, rebindNativeSessionCheckpoint } from "./native-session-resume.js";
|
||||
import { prepareNativeHeartbeatRun } from "./prepare-native-run.js";
|
||||
import { nativeRuntimeContextFixture } from "./runtime-context.test-fixture.js";
|
||||
|
||||
vi.mock("./paperclip-runner-tool-authority.js", () => ({ PaperclipRunnerToolAuthority: class {
|
||||
readonly binding = {};
|
||||
async definitions() { return []; }
|
||||
close() {}
|
||||
} }));
|
||||
|
||||
// Staging class P after #20757: run A yielded for an approval, its harness
|
||||
// backup was never saved, and its sandbox was destroyed. The follow-up run B
|
||||
// rebinds A's session in a new sandbox whose Runner has no saved state for it.
|
||||
// B has not started a turn, so a fresh provider session cannot repeat provider
|
||||
// work or a tool side effect. B continues in that fresh session on its own,
|
||||
// with no Retry. Any durable sign that B already did provider work keeps the
|
||||
// permanent named refusal.
|
||||
describeEmbeddedPostgres("governed-wait follow-up whose saved provider state is gone", () => {
|
||||
const environment = nativeRestartFixture();
|
||||
const homes: string[] = [];
|
||||
afterEach(async () => { vi.unstubAllEnvs(); for (const home of homes.splice(0)) await rm(home, { recursive: true, force: true }); });
|
||||
|
||||
async function followUpRun() {
|
||||
vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "61".repeat(32));
|
||||
const f = await environment.seedRun(randomUUID());
|
||||
const { companyId, agentId, runtimeRoot } = environment.current;
|
||||
const home = await mkdtemp(join(tmpdir(), "native-governed-lost-state-")); homes.push(home);
|
||||
vi.stubEnv("PAPERCLIP_HOME", home);
|
||||
const [sourceRun] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
const [contract] = await f.db.select().from(completionContracts).where(eq(completionContracts.id, sourceRun!.completionContractId!));
|
||||
const execution = parseNativeExecutionInput({
|
||||
schema: "paperclip.native-execution-input.v5", executionMode: "default", planningContext: null,
|
||||
provider: { kind: "codex", model: null, approvalPolicy: "never" },
|
||||
binding: { companyId, agentId, issueId: f.issueId, runId: f.runId, executionWorkspaceId: randomUUID() },
|
||||
task: { identifier: "GWL-1", title: "Approved work", description: null, prompt: "Continue after the approval.", workMode: "standard" },
|
||||
workspace: { cwd: runtimeRoot, repoUrl: null, repoRef: null, branchName: null },
|
||||
session: { normalizedSessionId: f.native.normalizedSessionId, driverKind: "codex_app_server", protocolVersion: 1, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } },
|
||||
completionContract: { id: contract!.id, sha256: contract!.canonicalSha256, schemaVersion: contract!.schemaVersion, contract: contract!.contractJson },
|
||||
runtimeContext: nativeRuntimeContextFixture(), credentialBindings: [], interactionResponses: [],
|
||||
});
|
||||
const yielded = { schema: "paperclip.run_result.v1", reportedWorkDisposition: "yielded", summary: "Waiting for approval",
|
||||
completionClaim: { contractRevision: "1", objectiveSatisfied: false, criteria: [{ criterionId: "objective", status: "satisfied", evidenceRefs: [] }], remainingWork: [] },
|
||||
evidence: [], verification: [], attentionRequests: [], artifacts: [],
|
||||
continuation: { kind: "response_wake", summary: "Continue after the answer", idempotencyKey: "fixture-approval" } };
|
||||
const snapshot: PersistedNativeSession = { backendKind: "runner", driverKind: "codex_app_server", sessionId: "lost-driver-session",
|
||||
providerSessionId: "lost-provider-session", identity: { companyId, agentId, issueId: f.issueId, runId: f.runId, sessionId: f.native.normalizedSessionId },
|
||||
cursor: "0", activeTurnId: null, pendingRuntimeRequests: [], terminalTurns: [], lineage: [], semanticResult: yielded as never, goal: null };
|
||||
const bootId = randomUUID(), leaseOwner = randomUUID();
|
||||
await f.db.update(nativeRunFinalizations).set({ controllerBootId: bootId, controllerGeneration: 1, controllerPid: 2_147_483_000,
|
||||
controllerProcessStartedAt: new Date(Date.now() - 60_000), leaseOwner, leaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
const core = new DurablePrpControlPlane({ stateDirectory: join(runtimeRoot, randomUUID()), identity: { runId: f.runId, normalizedSessionId: f.native.normalizedSessionId,
|
||||
runnerInstanceId: f.native.runnerInstanceId, environmentLeaseId: f.native.environmentLeaseId, turnId: f.native.turnId, itemId: f.native.itemId },
|
||||
expectedRunnerVersion: "governed-lost-state-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` });
|
||||
for (const type of ["run.prepare", "session.open", "turn.start", "runner.suspend"]) Object.assign(core.queueCommand(type), { status: "completed", result: { status: "completed" } });
|
||||
core.store.state.committedEvents.push({ sourceSeq: 1, sourceEventId: "fixture-terminal", eventType: "run.terminal", priority: 0, envelope: { ...core.store.state.identity }, deliveryCount: 1, logicalEffectCount: 1 });
|
||||
core.store.state.ackedSourceSeq = 1;
|
||||
core.setCheckpointPersistence(nativeRunnerCheckpointWriter(f.db, { companyId, issueId: f.issueId, runId: f.runId, authorityRunId: f.runId, bootId, generation: 1, leaseOwner }));
|
||||
await core.flushCheckpoint(); await core.stop();
|
||||
const [result] = await f.db.insert(nativeRunResults).values({ companyId, issueId: f.issueId, runId: f.runId, completionContractId: sourceRun!.completionContractId!,
|
||||
serverFingerprint: randomUUID(), schemaStatus: "accepted", resultJson: yielded, canonicalSha256: "a".repeat(64) }).returning();
|
||||
const fingerprint = nativeToolContractFingerprintForTarget("remote", "codex");
|
||||
const descriptor = nativeControllerMigrationDescriptor(execution, fingerprint);
|
||||
const sourceProfile = { ...f.preparedProfile, nativeExecutionInput: execution, sessionCheckpoint: snapshot, nativeToolContractFingerprint: fingerprint,
|
||||
nativeControllerRecovery: createNativeControllerRecovery(snapshot, { configDigest: descriptor.configDigest, processConfigDigest: null },
|
||||
{ runId: f.runId, runnerInstanceId: f.native.runnerInstanceId, normalizedSessionId: f.native.normalizedSessionId }) };
|
||||
await f.db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date(), runnerProfileJson: sourceProfile }).where(eq(heartbeatRuns.id, f.runId));
|
||||
await f.db.update(nativeRunFinalizations).set({ phase: "committed", resultId: result!.id, leaseOwner: null, leaseExpiresAt: null }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
|
||||
const runId = randomUUID();
|
||||
const [run] = await f.db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: f.issueId, issueId: f.issueId, status: "running",
|
||||
runtimeMode: "native", nativeSessionId: f.native.normalizedSessionId, runnerInstanceId: f.native.runnerInstanceId,
|
||||
completionContractId: sourceRun!.completionContractId, completionContractSha256: sourceRun!.completionContractSha256 }).returning();
|
||||
const native = await prepareNativeHeartbeatRun({ db: f.db, run: run!, issue: { id: f.issueId, title: "Approved work", description: null }, environmentLeaseId: randomUUID() });
|
||||
const current = { ...execution, binding: { ...execution.binding, runId } };
|
||||
const rebound = rebindNativeSessionCheckpoint({ previousRun: { id: f.runId, companyId, agentId, nativeSessionId: f.native.normalizedSessionId, runnerProfileJson: sourceProfile },
|
||||
currentExecution: current, executionTargetKind: "remote" });
|
||||
expect(rebound?.providerRecoveryPolicy).toBe("allow_replacement_after_governed_wait");
|
||||
const [prepared] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
|
||||
await f.db.update(heartbeatRuns).set({ runnerProfileJson: { ...prepared!.runnerProfileJson, nativeExecutionInput: current, sessionCheckpoint: rebound } }).where(eq(heartbeatRuns.id, runId));
|
||||
await f.db.insert(nativeRunFinalizations).values({ runId, companyId, issueId: f.issueId, phase: "observed" });
|
||||
await f.db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, f.issueId));
|
||||
|
||||
const recover = vi.fn(async () => ({ recovered: false as const, failureClass: "provider_state_incompatible" as const, resumeStateUnavailable: true as const,
|
||||
reason: "Error: NativeRunnerResumeStateUnavailableError: PRP provider resume state is unavailable" }));
|
||||
const replacementObserved = new Error("fixture fresh provider session opened");
|
||||
const fresh = vi.fn(async () => { throw replacementObserved; });
|
||||
const backend: NativeSessionBackend = {
|
||||
descriptor: async () => ({ kind: "runner", name: "controlled provider boundary", version: "fixture",
|
||||
capabilities: { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true },
|
||||
runtimeContextCapabilities: { instructions: "native", skills: "native", mcp: "native" } }),
|
||||
recoverSession: recover, openSession: fresh, openReplacementSession: fresh,
|
||||
};
|
||||
const execute = () => executePaperclipNativeSession({ db: f.db, execution: current, runnerInstanceId: native.runnerInstanceId, useRunnerd: true, backend,
|
||||
runnerExecutionTarget: { kind: "remote", transport: "sandbox", remoteCwd: runtimeRoot, environmentId: randomUUID(), leaseId: randomUUID(),
|
||||
providerKey: "fixture", reusableLeaseConfigured: true, runner: { execute: async () => { throw new Error("unexpected remote process execution"); } },
|
||||
sandboxLeaseAcquisition: { outcome: "created", providerLeaseId: "new-fixture-sandbox" } } as never }).catch((error: unknown) => error);
|
||||
return { f, companyId, agentId, runId, recover, fresh, replacementObserved, execute };
|
||||
}
|
||||
|
||||
const proofFor = (h: Awaited<ReturnType<typeof followUpRun>>) => new PaperclipControlPlanePort(h.f.db, {
|
||||
companyId: h.companyId, issueId: h.f.issueId, runId: h.runId, agentId: h.agentId, sessionId: "proof-session",
|
||||
completionContractId: "proof-contract", completionContractSha256: "proof-contract", sourceInstanceId: "proof-runner",
|
||||
controlPlaneSourceInstanceId: "proof-control",
|
||||
}).provesNoRunProviderWork({ signal: new AbortController().signal });
|
||||
|
||||
it("opens a fresh provider session for the follow-up without a Retry", async () => {
|
||||
const h = await followUpRun();
|
||||
expect(await proofFor(h)).toBe(true);
|
||||
const failure = await h.execute();
|
||||
expect(h.recover, String(failure)).toHaveBeenCalledOnce();
|
||||
// The fixture stops at the fresh session's boundary; on master the run
|
||||
// ended with native_provider_state_incompatible before reaching it.
|
||||
expect(failure).toBe(h.replacementObserved);
|
||||
expect(h.fresh).toHaveBeenCalledOnce();
|
||||
}, 60_000);
|
||||
|
||||
it.each([
|
||||
["a provider event", "event"],
|
||||
["a tool invocation", "tool"],
|
||||
["an accepted result", "result"],
|
||||
// A runner event of a non-provider type still carries a source identity:
|
||||
// a runner wrote it for this run, so the run is not untouched.
|
||||
["a sourced runner event of a non-provider type", "sourced_event"],
|
||||
// The foreign key normally forbids this. A row restored without its result
|
||||
// must still deny the proof rather than look like an untouched run.
|
||||
["a finalized result whose result row is missing", "dangling_result"],
|
||||
] as const)("keeps the named refusal when the follow-up run already recorded %s", async (_label, evidence) => {
|
||||
const h = await followUpRun();
|
||||
if (evidence === "event") await h.f.db.insert(heartbeatRunEvents).values({ companyId: h.companyId, runId: h.runId, agentId: h.agentId,
|
||||
seq: 1_000, eventType: "turn.started", stream: "system", level: "info", message: "turn started", payload: {} });
|
||||
if (evidence === "tool") await h.f.db.insert(toolInvocations).values({ companyId: h.companyId, runId: h.runId, agentId: h.agentId,
|
||||
toolName: "paperclip_get_issue", riskLevel: "read", status: "succeeded" } as never);
|
||||
if (evidence === "result") {
|
||||
const [run] = await h.f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, h.runId));
|
||||
await h.f.db.insert(nativeRunResults).values({ companyId: h.companyId, issueId: h.f.issueId, runId: h.runId, completionContractId: run!.completionContractId!,
|
||||
serverFingerprint: randomUUID(), schemaStatus: "accepted", resultJson: {}, canonicalSha256: "b".repeat(64) });
|
||||
}
|
||||
if (evidence === "sourced_event") await h.f.db.insert(heartbeatRunEvents).values({ companyId: h.companyId, runId: h.runId, agentId: h.agentId,
|
||||
seq: 1_000, eventType: "native.diagnostic", stream: "system", level: "info", message: "runner diagnostic", payload: {},
|
||||
sourceInstanceId: "foreign-runner-instance", sourceEventId: "foreign-runner-instance:event:1", sourceSeq: 1 });
|
||||
if (evidence === "dangling_result") await h.f.db.transaction(async (tx) => {
|
||||
await tx.execute(sql`set local session_replication_role = 'replica'`);
|
||||
await tx.update(nativeRunFinalizations).set({ resultId: randomUUID() }).where(eq(nativeRunFinalizations.runId, h.runId));
|
||||
});
|
||||
if (evidence === "sourced_event" || evidence === "dangling_result") {
|
||||
// The fixture is the only evidence: no tool invocation or result row.
|
||||
expect(await h.f.db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, h.runId))).toEqual([]);
|
||||
expect(await h.f.db.select().from(toolInvocations).where(eq(toolInvocations.runId, h.runId))).toEqual([]);
|
||||
}
|
||||
// The durable proof itself refuses each kind of evidence.
|
||||
expect(await proofFor(h)).toBe(false);
|
||||
const failure = await h.execute();
|
||||
expect(h.fresh).not.toHaveBeenCalled();
|
||||
if (evidence === "dangling_result") {
|
||||
// Admission stops even earlier on a finalized result: still no fresh session.
|
||||
expect(String(failure)).toContain("native_result_pending_finalization");
|
||||
return;
|
||||
}
|
||||
expect(h.recover, String(failure)).toHaveBeenCalledOnce();
|
||||
expect((failure as { code?: string }).code).toBe("native_provider_state_incompatible");
|
||||
}, 60_000);
|
||||
});
|
||||
@@ -10,6 +10,7 @@ import {
|
||||
nativeRunFinalizations,
|
||||
nativeRunResults,
|
||||
nativeRunnerCheckpoints,
|
||||
toolInvocations,
|
||||
} from "@paperclipai/db";
|
||||
import type {
|
||||
CompleteControlPlaneRunInput,
|
||||
@@ -36,6 +37,7 @@ import { eligibleNativeTaskSessionPublication, publishNativeTaskSessionInTransac
|
||||
type NativeTaskSessionPublicationContext, type NativeTaskSessionPublicationIssue } from "./native-task-session-publication.js";
|
||||
import { lockNativeRunnerCheckpointOwner, nativeRunnerCheckpointOwnerLock, validateNativeRunnerCheckpointOwner, inspectNativeRunnerUpgradeHeader, type NativeRunnerCheckpointOwner } from "./native-runner-checkpoint.js";
|
||||
import { crashPoint } from "@paperclipai/shared/crash-points";
|
||||
import { NATIVE_PROVIDER_DURABLE_EVENT_TYPES } from "./native-retired-backup-source.js";
|
||||
|
||||
export interface PaperclipControlPlaneBinding {
|
||||
companyId: string;
|
||||
@@ -512,6 +514,34 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
|
||||
return { events, highestContiguousSourceSeq: cursor };
|
||||
}
|
||||
|
||||
/**
|
||||
* Proves from durable rows, across every runner instance, that this run has
|
||||
* no provider event, source-sequenced runner event, tool invocation, or
|
||||
* accepted result. Any such row, or a read failure, denies the proof.
|
||||
*/
|
||||
async provesNoRunProviderWork(options: { signal: AbortSignal }): Promise<boolean> {
|
||||
options.signal.throwIfAborted();
|
||||
const { companyId, runId } = this.#binding;
|
||||
const providerEvents = sql.join([...NATIVE_PROVIDER_DURABLE_EVENT_TYPES].map(value => sql`${value}`), sql`, `);
|
||||
const result = await this.#db.execute(sql`select
|
||||
exists (select 1 from ${heartbeatRunEvents} event where event.company_id = ${companyId} and event.run_id = ${runId}
|
||||
and (event.source_event_id is not null or event.source_instance_id is not null
|
||||
or event.event_type in (${providerEvents})
|
||||
or event.event_type ~ '^(turn|item|tool|semantic_tool|run[.]result)[.]')) as "providerEvent",
|
||||
exists (select 1 from ${toolInvocations} invocation where invocation.company_id = ${companyId}
|
||||
and invocation.run_id = ${runId}) as "toolInvocation",
|
||||
exists (select 1 from ${nativeRunResults} result where result.company_id = ${companyId}
|
||||
and result.run_id = ${runId}) as "acceptedResult",
|
||||
exists (select 1 from ${nativeRunFinalizations} coordinator where coordinator.run_id = ${runId}
|
||||
and coordinator.result_id is not null) as "finalizedResult"`);
|
||||
options.signal.throwIfAborted();
|
||||
const rows = (Array.isArray(result) ? result : (result as { rows?: unknown[] }).rows ?? []) as Array<Record<string, unknown>>;
|
||||
const row = rows[0];
|
||||
if (!row) return false;
|
||||
return row.providerEvent === false && row.toolInvocation === false
|
||||
&& row.acceptedResult === false && row.finalizedResult === false;
|
||||
}
|
||||
|
||||
async completeRun(value: NativeRunResult | CompleteControlPlaneRunInput, options?: { signal: AbortSignal }): Promise<void> {
|
||||
this.#assertActive(options);
|
||||
if (this.#deferredEventObservers.size > 0 || this.#deferredObserverFailure) await this.#settleDeferredObservers(options);
|
||||
|
||||
Reference in new issue
Block a user