Reject failover backups behind the accepted event cursor (#19613)

A verified failover backup can still trail events already accepted by the controller. Compare the backup produced cursor with the registered controller state, or the supported local journal before registration, and reject stale files before remote sync. Keep the permanent replay-conflict classification ahead of generic Runner errors.

Validation: 1,192 native session and execution-driver tests and 10 workflow tests pass locally, together with the maintained low-memory Runner test project typecheck and fresh server declarations. All applicable checks on cd8c9cd07c are green (56 success, three skipped); exact-head Apex review is 5/5 with no unresolved findings. One e2e shard was retried after GitHub reported lost self-hosted runner communication, and the same-head rerun passed. No acknowledgement was lowered, provider replay authorized, or recovery files discarded.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Paperclip <Paperclip-Paperclip@users.noreply.github.com>
Co-Authored-By: Paperclip <noreply@paperclip.ing>

server/src/services/native-runtime/native-session-executor.ts

@@ -7057,6 +7057,35 @@ export function verifyNativeHarnessBackup(input: {
   return null;
 }
 
+export function nativeHarnessBackupCursorConflict(input: {
+  backupRoot: string;
+  stateRoot: string;
+  controllerState?: { schema: string; identity: { runnerInstanceId: string }; ackedSourceSeq: number };
+}): string | null {
+  const read = (path: string, maxBytes: number) => {
+    try {
+      return existsSync(path)
+        ? record(JSON.parse(readBoundedNativeFile(path, maxBytes, "runner_durable_identity_too_large").toString("utf8")))
+        : null;
+    } catch {
+      return null;
+    }
+  };
+  const control = input.controllerState ?? read(resolve(input.stateRoot, "control-plane", "control-plane-state.json"),
+    NATIVE_CONTROL_PLANE_STATE_MAX_BYTES);
+  if (!control || !RUNNERD_CONTROL_PLANE_STATE_SCHEMAS.has(String(control.schema))) return null;
+  const runner = read(resolve(input.backupRoot, "runner", "runner-state.json"), NATIVE_RUNNER_STATE_MAX_BYTES);
+  if (!runner) return null;
+  const acked = control.ackedSourceSeq;
+  const next = runner.nextSourceSeq;
+  if (!Number.isSafeInteger(acked) || !Number.isSafeInteger(next) || (next as number) < 1 ||
+      record(control.identity).runnerInstanceId !== runner.runnerInstanceId) return null;
+  const produced = (next as number) - 1;
+  return (acked as number) > produced
+    ? `native_event_replay_conflict: failover backup is behind the controller cursor (backup produced ${produced}, controller acknowledged ${acked as number})`
+    : null;
+}
+
 function nativeSessionCheckpointDirectory(): string {
   const directory = resolve(
     resolvePaperclipInstanceRoot(),
@@ -7464,6 +7493,9 @@ export function nativeSessionFailureSourceCode(
   ) {
     return "provider_transport_failed";
   }
+  if (/native_event_replay_conflict/i.test(message)) {
+    return "native_event_replay_conflict";
+  }
   if (
     /native_runner_process_exited|runnerd exited|runner process failed/i.test(
       message,
@@ -7474,9 +7506,6 @@ export function nativeSessionFailureSourceCode(
   if (/planning_mode_unsupported/i.test(message)) {
     return "planning_mode_unsupported";
   }
-  if (/native_event_replay_conflict/i.test(message)) {
-    return "native_event_replay_conflict";
-  }
   if (message.toLowerCase().includes(NATIVE_CURRENT_WAKE_COMMENTS_CHANGED_AFTER_READ.toLowerCase())) {
     return NATIVE_CURRENT_WAKE_COMMENTS_CHANGED_AFTER_READ;
   }
@@ -14433,6 +14462,9 @@ async function createRunnerdBackendWithinSessionClaim(
       runnerInstanceId: input.runnerInstanceId,
     });
     if (!backup) throw new Error("runner_harness_backup_unavailable");
+    const cursorConflict = nativeHarnessBackupCursorConflict({ backupRoot: backup.root, stateRoot: root,
+      controllerState: runnerControlPlane?.store.state });
+    if (cursorConflict) throw new Error(cursorConflict);
     await measureNativeRunnerSpan(
       input.trace,
       "harness_state.failover_restore",