fix(runner-e2e): honor governed wait boundaries

tests/runner-e2e/run-observations.ts

@@ -3,6 +3,24 @@ export interface ObservableRunState {
   errorCode?: string | null;
 }
 
+export interface ObservableRunEvent {
+  eventType?: string;
+  payload?: Record<string, unknown> | null;
+}
+
+export interface ObservableProviderSessionRun {
+  id: string;
+  sessionIdBefore?: string | null;
+  sessionIdAfter?: string | null;
+  contextSnapshot?: Record<string, unknown> | null;
+}
+
+function record(value: unknown): Record<string, unknown> {
+  return value && typeof value === "object" && !Array.isArray(value)
+    ? (value as Record<string, unknown>)
+    : {};
+}
+
 export function isNonExecutingReviewFenceRun(run: ObservableRunState) {
   return (
     run.status === "cancelled" &&
@@ -10,6 +28,80 @@ export function isNonExecutingReviewFenceRun(run: ObservableRunState) {
   );
 }
 
+export function isControlPlaneGovernedResponseWait(
+  events: readonly ObservableRunEvent[],
+) {
+  const accepted = events.filter(
+    (event) => event.eventType === "run.result.accepted",
+  );
+  if (accepted.length !== 1) return false;
+  const envelope = record(accepted[0]?.payload?.prpEvent);
+  const result = record(record(envelope.payload).result);
+  return (
+    envelope.schema === "paperclip.prp.event.v1" &&
+    envelope.eventType === "run.result.accepted" &&
+    envelope.sourceKind === "control_plane" &&
+    result.schema === "paperclip.run_result.v1" &&
+    result.reportedWorkDisposition === "yielded" &&
+    record(result.continuation).kind === "response_wake"
+  );
+}
+
+export function providerSessionContinuityFailures(
+  provider: "codex" | "opencode",
+  runs: readonly ObservableProviderSessionRun[],
+): string[] {
+  const failures: string[] = [];
+  for (let index = 0; index < runs.length; index += 1) {
+    const current = runs[index]!;
+    const currentSessionId = current.sessionIdAfter;
+    if (!currentSessionId) {
+      failures.push(
+        `expected ${provider} run ${current.id} to record provider session identity`,
+      );
+      continue;
+    }
+    if (index === 0) continue;
+
+    const previousSessionId = runs[index - 1]?.sessionIdAfter;
+    const context = record(current.contextSnapshot);
+    const acceptedPlanReset =
+      context.forceFreshSession === true &&
+      context.workspaceRefreshReason === "accepted_plan_confirmation" &&
+      context.source === "issue.interaction.accept" &&
+      context.interactionStatus === "accepted";
+    if (acceptedPlanReset) {
+      if (current.sessionIdBefore) {
+        failures.push(
+          `expected accepted Plan run ${current.id} to start without a prior provider session`,
+        );
+      }
+      if (previousSessionId && currentSessionId === previousSessionId) {
+        failures.push(
+          `expected accepted Plan run ${current.id} to rotate the ${provider} provider session`,
+        );
+      }
+      continue;
+    }
+
+    if (!previousSessionId || currentSessionId !== previousSessionId) {
+      failures.push(
+        `expected ${provider} to preserve its provider session for run ${current.id}`,
+      );
+      continue;
+    }
+    if (
+      current.sessionIdBefore &&
+      current.sessionIdBefore !== previousSessionId
+    ) {
+      failures.push(
+        `expected ${provider} run ${current.id} to resume provider session ${previousSessionId}`,
+      );
+    }
+  }
+  return failures;
+}
+
 export function numberedPlanStepCount(body: string | null | undefined) {
   return (body ?? "").split(/\r?\n/).filter((line) => {
     const normalized = line

tests/runner-e2e/runner.spec.ts

@@ -10,8 +10,10 @@ import { runnerE2EServerControlPaths } from "./harness-env.js";
 import { setupLiveFixtures, type LiveFixtureValues } from "./live-fixtures.js";
 import { evaluateMatcher, type MatcherResult } from "./matchers.js";
 import {
+  isControlPlaneGovernedResponseWait,
   isNonExecutingReviewFenceRun,
   numberedPlanStepCount,
+  providerSessionContinuityFailures,
 } from "./run-observations.js";
 import {
   assertSecretFree,
@@ -452,9 +454,12 @@ function nativeRunEventIntegrityFailures(
     }
   }
 
+  const runnerResultCount = runnerEventTypes.filter(
+    (value) => value === "run.result.proposed",
+  ).length;
   if (
-    runnerEventTypes.filter((value) => value === "run.result.proposed")
-      .length !== 1
+    runnerResultCount !== 1 &&
+    !(runnerResultCount === 0 && isControlPlaneGovernedResponseWait(events))
   ) {
     failures.push(
       `run ${run.id} must persist exactly one runner semantic result`,
@@ -1458,17 +1463,12 @@ for (const execution of executions) {
             execution.profile.provider === "opencode") &&
           selectedRuns.length > 1
         ) {
-          const providerSessions = selectedRuns.map(
-            (candidate) => candidate.sessionIdAfter,
+          invariantFailures.push(
+            ...providerSessionContinuityFailures(
+              execution.profile.provider,
+              selectedRuns,
+            ),
           );
-          if (
-            providerSessions.some((sessionId) => !sessionId) ||
-            new Set(providerSessions).size !== 1
-          ) {
-            invariantFailures.push(
-              `expected ${execution.profile.provider} to preserve one provider session across all heartbeat runs`,
-            );
-          }
         } else if (
           execution.profile.provider === "acpx" &&
           selectedRuns.length > 1

tests/runner-e2e/support.test.ts

@@ -27,8 +27,10 @@ import {
 } from "./redaction.js";
 import { parseDarwinSharedMemory } from "./shared-memory.js";
 import {
+  isControlPlaneGovernedResponseWait,
   isNonExecutingReviewFenceRun,
   numberedPlanStepCount,
+  providerSessionContinuityFailures,
 } from "./run-observations.js";
 import { runnerE2EWebServerCommand } from "./web-server-command.js";
 
@@ -299,6 +301,93 @@ describe("runner E2E run observations", () => {
       }),
     ).toBe(false);
   });
+
+  it("recognizes only authoritative control-plane governed response waits", () => {
+    const event = {
+      eventType: "run.result.accepted",
+      payload: {
+        prpEvent: {
+          schema: "paperclip.prp.event.v1",
+          eventType: "run.result.accepted",
+          sourceKind: "control_plane",
+          payload: {
+            result: {
+              schema: "paperclip.run_result.v1",
+              reportedWorkDisposition: "yielded",
+              continuation: { kind: "response_wake" },
+            },
+          },
+        },
+      },
+    };
+    expect(isControlPlaneGovernedResponseWait([event])).toBe(true);
+    expect(
+      isControlPlaneGovernedResponseWait([
+        {
+          ...event,
+          payload: {
+            prpEvent: {
+              ...event.payload.prpEvent,
+              sourceKind: "runner",
+            },
+          },
+        },
+      ]),
+    ).toBe(false);
+    expect(
+      isControlPlaneGovernedResponseWait([
+        event,
+        { ...event, payload: structuredClone(event.payload) },
+      ]),
+    ).toBe(false);
+  });
+
+  it("requires continuity except for an explicit accepted-Plan reset", () => {
+    const initial = {
+      id: "initial",
+      sessionIdBefore: null,
+      sessionIdAfter: "session-one",
+    };
+    const resumed = {
+      id: "resumed",
+      sessionIdBefore: "session-one",
+      sessionIdAfter: "session-one",
+    };
+    const acceptedPlan = {
+      id: "accepted-plan",
+      sessionIdBefore: null,
+      sessionIdAfter: "session-two",
+      contextSnapshot: {
+        forceFreshSession: true,
+        workspaceRefreshReason: "accepted_plan_confirmation",
+        source: "issue.interaction.accept",
+        interactionStatus: "accepted",
+      },
+    };
+    expect(
+      providerSessionContinuityFailures("codex", [
+        initial,
+        resumed,
+        acceptedPlan,
+      ]),
+    ).toEqual([]);
+    expect(
+      providerSessionContinuityFailures("codex", [
+        initial,
+        { ...acceptedPlan, sessionIdAfter: "session-one" },
+      ]),
+    ).toEqual([
+      "expected accepted Plan run accepted-plan to rotate the codex provider session",
+    ]);
+    expect(
+      providerSessionContinuityFailures("codex", [
+        initial,
+        { ...resumed, sessionIdAfter: "session-two" },
+      ]),
+    ).toEqual([
+      "expected codex to preserve its provider session for run resumed",
+    ]);
+  });
 });
 
 describe("runner E2E failure policy", () => {