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 <noreply@paperclip.ing>

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

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;

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<void>(resolve => { terminalRecorded = resolve; });
     let observerFinished!: () => void;
     const observerDone = new Promise<void>(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<string, unknown>): 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;