From 8a0e7ab5117e3b5b856137dd48600dbeb3168be7 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 4 Nov 2026 15:07:06 -0600 Subject: [PATCH] fix(runner): preserve ordinary append for unscoped governed terminals (#19044) ## Thinking Path > - Paperclip manages work done by AI agents. > - A native turn can park on a saved human question or approval. > - Its terminal event can return a committed receipt before observers finish. > - That option requires an explicit turn ID at the event port. > - The protocol also permits events without a turn ID. > - The runner requested the option for those events, which the port rejected. > - This change keeps those events on the ordinary append path. ## Linked Issues or Issue Description Refs: #19025. **What happened?** The native governed-wait observation can accept a schema-valid `item.completed` without a turn ID. If the later terminal also has no turn ID, the runner compares two absent IDs and requests `deferPostCommitObservers`. The event port rejects that option. The run then fails with `governed_wait_provider_terminal_unproven`. **Expected behavior** Only a terminal with an explicit turn ID can request the early receipt. An unscoped terminal must keep the ordinary append behavior and await its observers. The event port must retain its turn identity check. **Steps to reproduce** 1. Create a pending human-only interaction from the run. 2. Pass a schema-valid completed tool event without a turn ID through the real governed-wait observation. 3. Supply an interrupted terminal without a turn ID during the cancellation drain. 4. Run the actual native runtime with the PostgreSQL event port. The parent code rejects the deferral request and reports a failed terminal. **Paperclip version or commit** Reproduced on `4ea53f9eb5a24de10db3dfda18456132987c4af8`. The follow-up was identified by a review published after #19025 merged. No observed provider emission or original-task recovery is claimed. I checked current master and related PRs before this correction. **Deployment mode** Built from source. The regression uses real PostgreSQL and a deterministic provider boundary; it makes no model call. ## What Changed - Require a truthy terminal turn ID before the runner requests deferred observers. - Keep ordinary append behavior, the port identity check and all timeouts unchanged. - Exercise the production governed-wait observation with a persisted pending interaction and the real event port. - Retain exact-turn observer and checkpoint controls. Add a direct rejection control for unscoped deferral. - Document that unscoped terminals await their observers. ## Verification - First meaningful red: the actual observation, native runtime and PostgreSQL port fail on the unmodified parent runtime. The new case reports `NativeProviderTerminalFailure` in 143ms; this is not a setup or deadline failure. - Focused regression and authority controls: 12 passed, 69 filtered. Full runtime suite: 203 passed. Full PostgreSQL port and timing suites: 84 passed. - The tests leave the human-only interaction pending with continuation policy `none`, accept one result, and start only one turn. Existing scope, invalid evidence, checkpoint, abort, duplicate and restart controls remain. - Exact candidate: `8840ea2c9c308e5a60423e67fb3612d24029ba84`. Repository-wide `pnpm -r typecheck`, `pnpm build` and `pnpm check:module-boundaries` passed on this exact head. `pnpm test:run` is still in progress and is not reported as passing. Current-head CI passed: 55 success, 3 skipped and 1 neutral, including required `ci / verify` and `ci / e2e`. Greptile and Ponytail both reviewed this exact head at 5/5 with no blocking findings. The native review was source-only. No unresolved review threads remain. - No UI, database schema, agent permission or mapped user entry point changes. No deployment, live retry or provider replay was performed. ## Risks Unscoped terminals retain the existing observer wait and can still fail if it exceeds the existing drain deadline. They do not gain an early receipt. The change does not supply a missing turn ID or weaken source, company, run, session, cancellation or transaction checks. A PRP terminal is not physical-stop or accounting-completion proof. ## Model Used OpenAI `gpt-6.1-sol`, with code and shell tools. The host does not expose the reasoning mode or context-window size. ## 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 --- doc/execution-semantics.md | 8 ++- .../src/native-session-runtime.ts | 2 +- .../paperclip-control-plane-port.test.ts | 64 +++++++++++++------ 3 files changed, 49 insertions(+), 25 deletions(-) diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 2eafb2edc9b..2b0bb888ca0 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2722,9 +2722,11 @@ is retained; complete token counters alone cannot certify a stopped provider. A timeout, missing accounting, or an unproven provider terminal remains a failure; the saved wait does not certify successful provider execution. -An exact governed-cancellation terminal may return its durable event receipt -before postcommit accounting and output observers finish. The receipt requires -the ordinary company, run, session, source and transaction fences; it is not a +An exact governed-cancellation terminal with an explicit turn ID may return its +durable event receipt before postcommit accounting and output observers finish. +Schema-valid terminals without a turn ID keep the ordinary append path and await +their observers; they do not request the turn-bound early-receipt option. The +receipt requires the ordinary company, run, session, source and transaction fences; it is not a physical-stop or accounting-completion receipt. The control plane retains the observer debt and joins it before accepting a checkpoint or result. The existing checkpoint deadline bounds that join, while the provider-terminal drain keeps its diff --git a/packages/paperclip-runner/src/native-session-runtime.ts b/packages/paperclip-runner/src/native-session-runtime.ts index 3f0b6551bb2..13f222af23e 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/packages/paperclip-runner/src/native-session-runtime.ts @@ -1004,7 +1004,7 @@ async function consumeTurn( if (trailing.turnId === event.turnId && (usage || terminal || cancellation)) { const receipt = await controlPlane.appendEvent(trailing, { signal: drainAbort.signal, - ...(terminal && trailing.eventType !== "turn.failed" && trailing.payload.providerTerminalObserved !== false + ...(terminal && trailing.turnId && trailing.eventType !== "turn.failed" && trailing.payload.providerTerminalObserved !== false ? { deferPostCommitObservers: true as const } : {}), }); if (drainAbort.signal.aborted) return; diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts b/server/src/services/native-runtime/paperclip-control-plane-port.test.ts index 95d2b43cf45..f1a88c8378a 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts +++ b/server/src/services/native-runtime/paperclip-control-plane-port.test.ts @@ -3,7 +3,7 @@ import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external import { plugins } from "@paperclipai/db"; import { readTaskPanelActivations } from "../plugin-task-panels.js"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; -import { asc, eq, sql } from "drizzle-orm"; +import { and, asc, eq, sql } from "drizzle-orm"; import { generateKeyPairSync, randomUUID } from "node:crypto"; import { renderPaperclipWakePrompt } from "@paperclipai/adapter-utils/server-utils"; import { ONBOARDING_FIRST_TASK_ORIGIN_KIND } from "@paperclipai/shared"; @@ -64,7 +64,7 @@ import { assertCurrentWakeCommentsRead } from "./current-wake-comments.js"; import { buildExecutionContinuation } from "../execution-continuation.js"; import { childReviewOutcomes } from "./child-review-outcomes.js"; import { createPostgresRunDispatchAdapter } from "../../modules/run-dispatch/adapters/postgres.js"; -import { materializeRuntimeQuestionFallback, nativeGovernedWaitResult } from "./native-session-executor.js"; +import { createGovernedWaitEventObservation, materializeRuntimeQuestionFallback, nativeGovernedWaitResult } from "./native-session-executor.js"; import { subscribeAllCompanyLiveEvents, subscribeCompanyLiveEvents, @@ -314,7 +314,7 @@ describe("PaperclipControlPlanePort conformance", () => { } }); - it.each(["settles", "checkpoint timeout"] as const)("retains a durably recorded governed terminal while its postcommit observer is pending: %s", async mode => { + it.each(["settles", "checkpoint timeout", "unscoped terminal"] as const)("retains a durably recorded governed terminal while its postcommit observer is pending: %s", async mode => { const identity = { ...CONTROL_PLANE_CONFORMANCE_OPEN.identity, issueId: randomUUID(), runId: randomUUID(), sessionId: randomUUID() }; const runnerId = randomUUID(); const fixtureContractId = randomUUID(); @@ -337,10 +337,20 @@ describe("PaperclipControlPlanePort conformance", () => { const recorded = new Promise(resolve => { terminalRecorded = resolve; }); let observerFinished!: () => void; const observerDone = new Promise(resolve => { observerFinished = resolve; }); + const observation = createGovernedWaitEventObservation(async () => { + const [pending] = await db.select().from(issueThreadInteractions).where(and( + eq(issueThreadInteractions.id, interaction.id), eq(issueThreadInteractions.companyId, identity.companyId), + eq(issueThreadInteractions.sourceRunId, identity.runId), eq(issueThreadInteractions.status, "pending"), + )); + return pending?.status === "pending" && pending.sourceRunId === identity.runId + ? nativeGovernedWaitResult({ interaction: pending, completionContract: contract }) : null; + }); + let triggerConsumed = false; const port = new PaperclipControlPlanePort(db, { ...identity, completionContractId: fixtureContractId, completionContractSha256: contractSha, sourceInstanceId: runnerId, controlPlaneSourceInstanceId: "park-control" }, { async onCommittedEvent(event) { + await observation.observe(event, event.eventType === "item.completed"); if (event.eventType === "turn.interrupted") { terminalWasRecorded = true; terminalRecorded(); try { await observerGate; } finally { observerFinished(); } } }, }); @@ -349,7 +359,7 @@ describe("PaperclipControlPlanePort conformance", () => { title: interaction.title ?? null, summary: interaction.summary ?? null }, completionContract: contract }); const event = (sourceSeq: number, eventType: PrpEvent["eventType"], payload: Record): PrpEvent => ({ schema: "paperclip.prp.event.v1", sourceEventId: `${runnerId}:${sourceSeq}`, sourceInstanceId: runnerId, sourceSeq, - sourceKind: "runner", runId: identity.runId, normalizedSessionId: identity.sessionId, turnId: "park-turn", + sourceKind: "runner", runId: identity.runId, normalizedSessionId: identity.sessionId, ...(mode === "unscoped terminal" ? {} : { turnId: "park-turn" }), eventType, schemaVersion: 1, priority: 0, emittedAt: new Date().toISOString(), payload, }); let starts = 0, terminalEmitted = false; @@ -376,30 +386,41 @@ describe("PaperclipControlPlanePort conformance", () => { session: { normalizedSessionId: identity.sessionId, driverKind: "codex_app_server", protocolVersion: 1 }, provider: { kind: "codex", model: null }, interactionResponses: [], credentialBindings: [], completionContract: { id: fixtureContractId, sha256: contractSha, schemaVersion: "paperclip.completion-contract.v1", contract } }), - resolveGovernedWait: ({ event }: { event: PrpEvent }) => event.eventType === "item.completed" ? yielded : null, + resolveGovernedWait: ({ event }: { event: PrpEvent }) => { + const result = observation.consume(event); + triggerConsumed ||= result !== null; + return result; + }, }; + if (mode === "unscoped terminal") releaseObserver(); const execution = executeNativeSession(options); const outcome = execution.then(value => ({ value }), error => ({ error })); try { await Promise.race([recorded, outcome.then(result => { if ("error" in result) throw result.error; throw new Error("run settled before recording terminal"); })]); const rows = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, identity.runId)); expect(rows.some(row => row.eventType === "turn.interrupted")).toBe(true); - await new Promise(resolve => setTimeout(resolve, 700)); - expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, identity.runId))).toEqual([]); - releaseObserver(); - await observerDone; - if (mode === "settles") expect(await outcome).not.toHaveProperty("error"); - else { - expect(await outcome).toMatchObject({ error: { message: "native session checkpoint timed out after 100ms" } }); + expect(triggerConsumed).toBe(true); + if (mode !== "unscoped terminal") { + await new Promise(resolve => setTimeout(resolve, 700)); expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, identity.runId))).toEqual([]); - const restored = new PaperclipControlPlanePort(db, { ...identity, completionContractId: fixtureContractId, - completionContractSha256: contractSha, sourceInstanceId: runnerId, controlPlaneSourceInstanceId: "park-control" }); - const checkpoint = await restored.loadSessionCheckpoint(); - expect(checkpoint?.governedWait?.sourceEvent).toMatchObject({ sourceSeq: 1, turnId: "park-turn" }); - const session = await backend.openSession({ identity }); - const recovering = { ...backend, async recoverSession() { return { recovered: true as const, session }; } }; - await expect(executeNativeSession({ ...options, checkpointTimeoutMs: 1_000, controlPlane: restored, - backend: recovering, persistedSession: checkpoint })).resolves.toMatchObject({ result: yielded }); + releaseObserver(); + await observerDone; + if (mode === "settles") expect(await outcome).not.toHaveProperty("error"); + else { + expect(await outcome).toMatchObject({ error: { message: "native session checkpoint timed out after 100ms" } }); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, identity.runId))).toEqual([]); + const restored = new PaperclipControlPlanePort(db, { ...identity, completionContractId: fixtureContractId, + completionContractSha256: contractSha, sourceInstanceId: runnerId, controlPlaneSourceInstanceId: "park-control" }); + const checkpoint = await restored.loadSessionCheckpoint(); + expect(checkpoint?.governedWait?.sourceEvent).toMatchObject({ sourceSeq: 1, turnId: "park-turn" }); + const session = await backend.openSession({ identity }); + const recovering = { ...backend, async recoverSession() { return { recovered: true as const, session }; } }; + await expect(executeNativeSession({ ...options, checkpointTimeoutMs: 1_000, controlPlane: restored, + backend: recovering, persistedSession: checkpoint })).resolves.toMatchObject({ result: yielded }); + } + } else { + expect(await outcome).not.toHaveProperty("error"); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, identity.runId))).toHaveLength(1); } expect(starts).toBe(1); expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, interaction.id)))[0]).toMatchObject({ @@ -553,7 +574,7 @@ describe("PaperclipControlPlanePort conformance", () => { } finally { release(); await duplicate; } }); - it.each(["company", "agent", "run", "session", "source", "failed", "unproven", "nonterminal"] as const)("does not grant deferred terminal authority to %s evidence", async fault => { + it.each(["company", "agent", "run", "session", "source", "failed", "unproven", "nonterminal", "unscoped"] as const)("does not grant deferred terminal authority to %s evidence", async fault => { const f = await terminalObserverFixture(); const event = { ...f.event, ...(fault === "run" ? { runId: randomUUID() } : {}), @@ -562,6 +583,7 @@ describe("PaperclipControlPlanePort conformance", () => { ...(fault === "failed" ? { eventType: "turn.failed" as const } : {}), ...(fault === "unproven" ? { payload: { providerTerminalObserved: false } } : {}), ...(fault === "nonterminal" ? { eventType: "item.completed" as const } : {}), + ...(fault === "unscoped" ? { turnId: undefined } : {}), }; const port = fault === "company" || fault === "agent" ? new PaperclipControlPlanePort(db, { ...f.binding, ...(fault === "company" ? { companyId: randomUUID() } : { agentId: randomUUID() }) }) : f.port;