mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-09 07:10:34 +01:00
fix(runner): settle durable provider exits on open streams (#19657)
## Thinking Path > - Paperclip manages AI agents and their work. > - Runner keeps the native session and its event stream until a turn ends. > - A provider can exit with a terminal error while its event stream stays open. > - Waiting for that stream to close leaves the run and its resource owner active. > - The existing durable failure receipt already identifies the failed current turn, including an authoritative replay. > - This pull request uses that event to enter the existing failure and cleanup path. > - Recovery and later idle retirement can then proceed from a real terminal result. ## Linked Issues or Issue Description Refs #18988 for native resource retirement. This is a separate terminal-event defect found during failed-turn recovery qualification. A search of open and closed PRs found no duplicate `provider_exited` fix. The related merged #12596 changes Rust test polling; it does not change this TypeScript session owner. **What happened?** `executeNativeSession` records a nonrecoverable `provider_exited` event, but can continue to wait on an open event iterator. A proposed result does not authorize success, and stream EOF may never arrive. The run cannot reach its existing failure and resource cleanup path. **Expected behavior** Settle the committed or authoritative duplicate terminal failure for this exact run, session and provider. Close through the existing owner. Preserve recoverable goals, completed-turn and identity guards. A duplicate must not add another committed event count. **Steps to reproduce** 1. Supply a native session whose iterator emits a current, nonrecoverable, unexpected `provider_exited` failure. 2. Keep that iterator open until the session owner closes it. 3. Observe that the old implementation does not settle within the existing test bound. 4. Apply the guard below. The same execution fails with the typed terminal error and calls the existing close owner once. **Paperclip version or commit** The original regression was reproduced against master source `05a19768ae4d337a4ae88ecdbc259f251406487c`. Current candidate `f4bb8f1b68e2793062dd640ee56b87a9208b4fd9` normally merges fetched master `07a548073ab48f13cb93b2fdc31553a5da88a070`. All unrelated master paths are preserved exactly. The two-file PR diff contains only the guarded settlement and its tests. ## What Changed - Add a guard after the existing durable event receipt. It requires a committed or duplicate receipt, a nonrecoverable unexpected exit, no completed current or previous turn, and exact provider/run/session identity. - Throw the existing typed provider failure so the existing failure and close owners run. Add no scheduler, polling loop, provider parser or cleanup owner. Event counting remains committed-only. - Cover open streams with and without a result proposal, authoritative duplicate replay, all six safety predicates, and a recoverable goal that later completes. Preserve accepted-result and close-once assertions. ## Verification - The original open-stream regression fails before the production change. Review then found that the duplicate test released EOF before asserting settlement. The corrected duplicate control fails on the prior production code at its original timely-settlement assertion; it passes after accepting the authoritative duplicate. All 21 selected exit, safety, goal and compaction controls pass. - The normal durable transport validates authenticated current run/session/turn/item identity and canonical replay equality before delivery. The App port validates current authority and binding; its durable event transaction returns duplicate only for matching source identity and canonical payload. No absent or conflicting event authorizes settlement. The six negative consumer tests use a deliberately permissive mock port; they prove this consumer guard, not that production accepts foreign events. - The first remote Typecheck job found five errors in new mocks that accessed PRP fields without narrowing a native-or-PRP event. Both mocks now explicitly reject the wrong event shape. Complete original CI evidence is retained. No cast, suppression or narrowed test compiler program is used. - On current integrated candidate `f4bb8f1`, normal Runner emission and the full normal `tsconfig.test.json` source/test program pass. All 26 selected cases pass: the same 21 controls plus all five structured provider-failure/model-retry rows affected by the incoming master change. The other 197 cases were filtered, not counted as passing. - Before master integration, all 411 declaration paths and bytes were identical across the settlement change. Incoming master independently adds one OpenCode declaration and changes three OpenCode declarations; those deltas are recorded separately. This fix changes no runtime declaration. No Rust source or binary is changed by this PR. - Actual guarded workers recorded zero PG sockets, active listeners or blocked attempts. All known worker processes/groups closed. The working tree is clean after qualification. No install, Rust build, App run, provider connection or PostgreSQL suite was part of the final freshness loop. - Fresh exact-head remote CI and complete bot reviews remain required. A local whole-repository test/build, successful task retry or live-provider acceptance is not claimed. Run the selected controls from a normal dependency installation: ```sh pnpm --filter @paperclipai/paperclip-runner exec vitest run src/native-session-runtime.test.ts --testNamePattern='settles committed or duplicate provider exits while their stream stays open|retains EOF settlement when provider exit fails the|preserves a recoverable provider-exit goal until its later authoritative completion|preserves a durable session failure when its stream closes|preserves session-goal failure priority|context compaction|never continues task work when a restart loses a compaction turn|preserves structured provider failure and model retry classification' --maxWorkers=1 --no-file-parallelism ``` ## Risks - A false terminal classification could stop useful work. The guard requires the existing durable receipt and explicit current-turn failure fields; recoverable and completed-turn controls remain covered. An exact durable duplicate may settle the same failure without repeating accounting. - Close can still fail. This change reuses the existing close owner and does not manufacture resource retirement or clear retained failure debt. - This proves Runner failure settlement under a controlled session boundary. It does not establish successful task retry, whole-tenant quiet, live provider closure, or deployed suspension. ## Model Used OpenAI `gpt-6.1-sol` produced the delegated fix and qualification. OpenAI Codex, based on GPT-6, performed managing technical review and integration. Both used repository tools and code execution. The coordinator's exact backend snapshot and context-window size were not exposed. ## 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 linked existing issues or described the issue in-PR following the issue template - [x] I have not referenced internal or instance-local Paperclip issues or links - [x] My branch name describes the change and contains no internal Paperclip ticket id or instance-derived details - [x] I have run the selected tests locally and they pass - [x] I have added tests for the changed behavior - [x] I have reviewed documentation; no command or public contract changes require a documentation edit - [x] I have considered and documented the risks above - [ ] All Paperclip CI gates are green — fresh exact-head checks pending - [ ] Greptile is 5/5 with no open findings — fresh exact-head review pending - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
444b4de693
commit
19c2bebc9c
2 files changed
+146
No files matched your search
@@ -698,6 +698,146 @@ describe("executeNativeSession recovery", () => {
|
||||
},
|
||||
);
|
||||
|
||||
it.each([{ proposal: false, replayed: false }, { proposal: true, replayed: false }, { proposal: false, replayed: true }])("settles committed or duplicate provider exits while their stream stays open ($proposal, $replayed)", async ({ proposal, replayed }) => {
|
||||
const capabilities = { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true };
|
||||
let release!: () => void;
|
||||
const held = new Promise<void>(resolve => { release = resolve; });
|
||||
const close = vi.fn(async () => { release(); });
|
||||
const events: PrpEvent[] = [];
|
||||
const session: NativeSession = {
|
||||
identity: () => identity,
|
||||
async capabilities() { return capabilities; },
|
||||
async *events() {
|
||||
if (proposal) yield runnerEvent(1, "run.result.proposed", result);
|
||||
yield runnerEvent(proposal ? 2 : 1, "session.failed", { code: "provider_exited", provider: "codex",
|
||||
recoverable: false, expected: false, completedByExitedProcess: false,
|
||||
previousTurnCompleted: false, exitCode: 1, processGeneration: 1 });
|
||||
await held;
|
||||
},
|
||||
async startTurn() { return { turnId: "turn-recovery" }; },
|
||||
async result() { return null; },
|
||||
async snapshot() { return { backendKind: "mock", sessionId: "driver-recovery", identity,
|
||||
providerSessionId: "provider-recovery", cursor: null, activeTurnId: null, pendingRuntimeRequests: [], lineage: [] }; },
|
||||
close,
|
||||
};
|
||||
const completeRun = vi.fn();
|
||||
const execution = executeNativeSession({ input, backend: {
|
||||
async descriptor() { return { kind: "mock", name: "open-stream-exit", version: "1", capabilities }; },
|
||||
async openSession() { return session; },
|
||||
}, controlPlane: {
|
||||
async openRun() {}, async checkpointSession() {},
|
||||
async appendEvent(event) {
|
||||
if (!("sourceSeq" in event)) throw new Error("Expected provider PRP event");
|
||||
events.push(event); return { cursor: event.sourceSeq,
|
||||
highestContiguousSourceSeq: event.sourceSeq, disposition: replayed ? "duplicate" : "committed" }; },
|
||||
async replayEvents() { return { events: [], highestContiguousSourceSeq: 0 }; }, completeRun,
|
||||
}, runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery" });
|
||||
let settled = false, succeeded = false, failure: unknown;
|
||||
const observed = execution.then(() => { succeeded = true; settled = true; }, error => { failure = error; settled = true; });
|
||||
try {
|
||||
await vi.waitFor(() => expect(settled).toBe(true), { timeout: 100, interval: 5 });
|
||||
expect(succeeded).toBe(false);
|
||||
expect(failure).toMatchObject({ code: "native_provider_terminal_failed", providerCode: "provider_exited", recoverable: false });
|
||||
expect(completeRun).not.toHaveBeenCalled();
|
||||
expect(events.filter(event => event.eventType === "run.result.proposed").map(event => event.payload)).toEqual(proposal ? [result] : []);
|
||||
expect(events.some(event => event.eventType === "run.result.accepted")).toBe(false);
|
||||
expect(close).toHaveBeenCalledOnce();
|
||||
} finally { release(); await observed; }
|
||||
});
|
||||
|
||||
it.each(["expected", "completedByExitedProcess", "previousTurnCompleted", "provider", "run", "session"] as const)("retains EOF settlement when provider exit fails the %s safety predicate", async mismatch => {
|
||||
const capabilities = { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true };
|
||||
let release!: () => void;
|
||||
const held = new Promise<void>(resolve => { release = resolve; });
|
||||
let streamBlocked = false;
|
||||
const close = vi.fn(async () => { release(); });
|
||||
const event = { ...runnerEvent(1, "session.failed", { code: "provider_exited", provider: mismatch === "provider" ? "acpx" : "codex",
|
||||
recoverable: false, expected: mismatch === "expected", completedByExitedProcess: mismatch === "completedByExitedProcess",
|
||||
previousTurnCompleted: mismatch === "previousTurnCompleted", exitCode: 1, processGeneration: 1 }),
|
||||
...(mismatch === "run" ? { runId: "foreign-run" } : {}),
|
||||
...(mismatch === "session" ? { normalizedSessionId: "foreign-session" } : {}) };
|
||||
const session: NativeSession = {
|
||||
identity: () => identity, async capabilities() { return capabilities; },
|
||||
async *events() { yield event; streamBlocked = true; await held; },
|
||||
async startTurn() { return { turnId: "turn-recovery" }; },
|
||||
async result() { return null; },
|
||||
async snapshot() { return { backendKind: "mock", sessionId: "driver-recovery", identity,
|
||||
providerSessionId: "provider-recovery", cursor: null, activeTurnId: null, pendingRuntimeRequests: [], lineage: [] }; },
|
||||
close,
|
||||
};
|
||||
const completeRun = vi.fn();
|
||||
const execution = executeNativeSession({ input, backend: {
|
||||
async descriptor() { return { kind: "mock", name: "guarded-provider-exit", version: "1", capabilities }; },
|
||||
async openSession() { return session; },
|
||||
}, controlPlane: {
|
||||
async openRun() {}, async checkpointSession() {},
|
||||
async appendEvent(value) {
|
||||
if (!("sourceSeq" in value)) throw new Error("Expected provider PRP event");
|
||||
return { cursor: value.sourceSeq, highestContiguousSourceSeq: value.sourceSeq, disposition: "committed" }; },
|
||||
async replayEvents() { return { events: [], highestContiguousSourceSeq: 0 }; }, completeRun,
|
||||
}, runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery" });
|
||||
let settled = false, failure: unknown;
|
||||
const observed = execution.then(() => { settled = true; }, error => { failure = error; settled = true; });
|
||||
try {
|
||||
await vi.waitFor(() => expect(streamBlocked).toBe(true), { timeout: 100, interval: 5 });
|
||||
expect(settled).toBe(false); expect(close).not.toHaveBeenCalled(); expect(completeRun).not.toHaveBeenCalled();
|
||||
release();
|
||||
await observed;
|
||||
expect(failure).toMatchObject({ code: "native_provider_terminal_failed", providerCode: "provider_exited", recoverable: false });
|
||||
expect(close).toHaveBeenCalledOnce(); expect(completeRun).not.toHaveBeenCalled();
|
||||
} finally { release(); await observed; }
|
||||
});
|
||||
|
||||
it("preserves a recoverable provider-exit goal until its later authoritative completion", async () => {
|
||||
const capabilities = { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true };
|
||||
const activeGoal = { threadId: "provider-recovery", objective: "Finish work", status: "active" as const,
|
||||
tokenBudget: null, tokensUsed: 100, timeUsedSeconds: 1,
|
||||
createdAt: 1, updatedAt: 2 };
|
||||
const completeGoal = { ...activeGoal, status: "complete" as const };
|
||||
let release!: () => void, reached!: () => void;
|
||||
const held = new Promise<void>(resolve => { release = resolve; });
|
||||
const failureObserved = new Promise<void>(resolve => { reached = resolve; });
|
||||
const close = vi.fn(async () => { release(); });
|
||||
const session: NativeSession = {
|
||||
identity: () => identity, async capabilities() { return capabilities; },
|
||||
async *events() {
|
||||
yield runnerEvent(1, "session.goal.updated", { goal: activeGoal, workingNow: true });
|
||||
yield runnerEvent(2, "session.failed", { code: "provider_exited", provider: "codex", recoverable: true,
|
||||
expected: false, completedByExitedProcess: false, previousTurnCompleted: false, exitCode: 1, processGeneration: 1 });
|
||||
reached(); await held;
|
||||
yield runnerEvent(3, "session.goal.updated", { goal: activeGoal, workingNow: false });
|
||||
yield runnerEvent(4, "turn.completed");
|
||||
yield runnerEvent(5, "session.goal.snapshot", { goal: completeGoal, workingNow: false });
|
||||
},
|
||||
async startTurn() { return { turnId: "turn-recovery" }; },
|
||||
async result() { return null; },
|
||||
async goal() { return completeGoal; },
|
||||
async snapshot() { return { backendKind: "mock", sessionId: "driver-recovery", identity,
|
||||
providerSessionId: "provider-recovery", cursor: null, activeTurnId: null, pendingRuntimeRequests: [], goal: activeGoal, lineage: [] }; },
|
||||
close,
|
||||
};
|
||||
const completeRun = vi.fn();
|
||||
const execution = executeNativeSession({ input, backend: {
|
||||
async descriptor() { return { kind: "mock", name: "recoverable-open-exit", version: "1", capabilities }; },
|
||||
async openSession() { return session; },
|
||||
}, controlPlane: {
|
||||
async openRun() {}, async checkpointSession() {},
|
||||
async appendEvent(event) {
|
||||
if (!("sourceSeq" in event)) throw new Error("Expected provider PRP event");
|
||||
return { cursor: event.sourceSeq, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" }; },
|
||||
async replayEvents() { return { events: [], highestContiguousSourceSeq: 0 }; }, completeRun,
|
||||
}, runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery" });
|
||||
let settled = false;
|
||||
const observed = execution.then(value => { settled = true; return value; }, error => { settled = true; throw error; });
|
||||
try {
|
||||
await failureObserved;
|
||||
expect(settled).toBe(false); expect(close).not.toHaveBeenCalled(); expect(completeRun).not.toHaveBeenCalled();
|
||||
release();
|
||||
expect(await observed).toMatchObject({ result: { reportedWorkDisposition: "done" } });
|
||||
expect(close).toHaveBeenCalledOnce(); expect(completeRun).toHaveBeenCalledOnce();
|
||||
} finally { release(); await observed.catch(() => undefined); }
|
||||
});
|
||||
|
||||
it.each([false, true])("preserves a durable session failure when its stream closes (throws=%s)", async throws => {
|
||||
const capabilities = { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true };
|
||||
const session: NativeSession = {
|
||||
|
||||
@@ -1159,6 +1159,12 @@ async function consumeTurn(
|
||||
typeof failure.message === "string" ? failure.message : undefined,
|
||||
readProviderFailure(failure.providerFailure),
|
||||
);
|
||||
if ((receipt.disposition === "committed" || receipt.disposition === "duplicate") && !providerFailure.recoverable && failure.code === "provider_exited"
|
||||
&& failure.expected === false && failure.completedByExitedProcess === false
|
||||
&& failure.previousTurnCompleted === false && failure.provider === input.provider.kind
|
||||
&& event.runId === input.binding.runId && event.normalizedSessionId === session.identity().sessionId) {
|
||||
throw providerFailure;
|
||||
}
|
||||
}
|
||||
const request =
|
||||
payload.request &&
|
||||
|
||||
Reference in new issue
Block a user