Keep pending goal commands off incompatible runner leases

Co-Authored-By: Paperclip <noreply@paperclip.ing>

packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts

@@ -1055,6 +1055,60 @@ it.each([
   },
 );
 
+it.each([false, true])("preserves pending v2 goals across an incompatible reconnect (v1 prefix: %s)", async (withPrefix) => {
+  const root = mkdtempSync(resolve(tmpdir(), "runner-goal-downgrade-test-"));
+  const core = new DurablePrpControlPlane({ stateDirectory: root, identity, expectedRunnerVersion, expectedRunnerDigest });
+  const clients: AuthenticatedClient[] = [];
+  const connectVersion = async (version: number) => {
+    const client = await authenticate(core, core.issueBootstrapTicket(), identity,
+      expectedRunnerDigest, undefined, false, version);
+    if (client) clients.push(client);
+    return client;
+  };
+  try {
+    await core.start();
+    const original = (await connectVersion(2))!;
+    const prefix = withPrefix ? core.queueCommand("run.prepare") : null;
+    const goal = core.queueCommand("session.goal.get", {}, "pending-v2-goal");
+    const suspend = core.queueCommand("runner.suspend");
+    const originalGoal = structuredClone(goal);
+    original.socket.destroy();
+    await vi.waitFor(() => expect(core.activeRunnerConnectionCount()).toBe(0));
+
+    const oldRunner = await connectVersion(1);
+    if (prefix) {
+      expect(oldRunner).not.toBeNull();
+      expect(oldRunner!.welcome.payload).toMatchObject({ pendingCommands: [expect.objectContaining({ commandId: prefix.commandId })] });
+      sendSecure(oldRunner!, {
+        protocol: "paperclip.runner", version: 1, kind: "command_result",
+        payload: { commandId: prefix.commandId, commandType: prefix.type,
+          controllerSeq: prefix.controllerSeq, status: "completed", result: {} },
+      });
+      await expect(receiveSecure(oldRunner!)).resolves.toBeNull();
+    } else {
+      expect(oldRunner).toBeNull();
+    }
+    expect(core.store.state.commands.find((command) => command.commandId === goal.commandId)).toEqual(originalGoal);
+    expect(core.store.state.commandDeliveryCounts[goal.commandId] ?? 0).toBe(0);
+
+    const compatible = (await connectVersion(2))!;
+    expect(compatible.welcome.payload).toMatchObject({ pendingCommands: [expect.objectContaining({ commandId: goal.commandId, schema: "paperclip.prp.command.v2" })] });
+    sendSecure(compatible, {
+      protocol: "paperclip.runner", version: 2, kind: "command_result",
+      payload: { commandId: goal.commandId, commandType: goal.type,
+        controllerSeq: goal.controllerSeq, status: "completed", result: {} },
+    });
+    await expect(receiveSecure(compatible)).resolves.toMatchObject({
+      kind: "command", payload: { commandId: suspend.commandId, schema: "paperclip.prp.command.v1" },
+    });
+    expect(core.commandOutcome(goal.commandId)?.status).toBe("completed");
+  } finally {
+    for (const client of clients) client.socket.destroy();
+    await core.stop();
+    rmSync(root, { recursive: true, force: true });
+  }
+});
+
 it.each([1, 2])("only journals session goals after negotiating PRP v2 (version %s)", async (version) => {
   const root = mkdtempSync(resolve(tmpdir(), "runner-goal-version-test-"));
   const core = new DurablePrpControlPlane({

packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts

@@ -2706,7 +2706,10 @@ export class DurablePrpControlPlane {
     this.#store.state.lastLeaseId = lease.leaseId;
     this.#store.state.lastLeaseExpiresAt = lease.expiresAt;
 
-    const pending = connection.replayOnly ? [] : this.#nextPendingCommand();
+    const pending = connection.replayOnly
+      ? []
+      : this.#nextPendingCommand(connection);
+    if (pending === null) return;
     const [pendingCommand] = pending;
     connection.terminalLifecycleCommandId =
       pendingCommand && this.#isTerminalLifecycleCommand(pendingCommand)
@@ -2776,11 +2779,23 @@ export class DurablePrpControlPlane {
     return wire;
   }
 
-  #nextPendingCommand(): DurableRecoveryCoreCommand[] {
+  #nextPendingCommand(
+    connection: AuthorityConnection,
+  ): DurableRecoveryCoreCommand[] | null {
     if (this.#store.state.warmTransition) return [];
     const command = this.#store.state.commands.find(
       (candidate) => candidate.status === "pending",
     );
+    if (
+      command?.schema === "paperclip.prp.command.v2" &&
+      connection.lease?.protocolVersion !== 2
+    ) {
+      // Queue-time negotiation does not authorize delivery after a reconnect.
+      // Refuse this incompatible connection without consuming or skipping the
+      // durable command: a compatible runner must resume it before later work.
+      connection.close();
+      return null;
+    }
     return command === undefined ? [] : [command];
   }
 
@@ -2820,7 +2835,9 @@ export class DurablePrpControlPlane {
       this.#store.state.warmTransition
     )
       return;
-    const [command] = this.#nextPendingCommand();
+    const pending = this.#nextPendingCommand(connection);
+    if (pending === null) return;
+    const [command] = pending;
     if (command === undefined) return;
     if (this.#isTerminalLifecycleCommand(command)) {
       connection.terminalLifecycleCommandId = command.commandId;