fix(pi): separate promotion validation phases

.github/scripts/test_pi_promotion.py

@@ -169,6 +169,69 @@ def test_pending_candidate_validation_is_static_and_pr_stages_every_new_target(s
                 f"promotion PR does not stage declared target {path}",
             )
 
+    def test_runtime_contracts_use_reviewed_baseline_before_prospective_apply(self):
+        validate = validation_job(WORKFLOW_PATH.read_text(encoding="utf-8"))
+        lines = (line.strip().removeprefix("run: ") for line in validate.splitlines())
+        commands = [line for line in lines
+                    if line.startswith("python .github/scripts/pi_promotion.py contract ")]
+        prefix = 'python .github/scripts/pi_promotion.py contract --id '
+        state = ' --state "$CA_PI_RECEIPT_STATE" -- '
+        apply = prefix + "promotion-apply" + state + (
+            'python .github/scripts/pi_promotion.py apply --candidate "$CA_PI_CANDIDATE"'
+        )
+        self.assertEqual(commands.count(apply), 1)
+        boundary = commands.index(apply)
+        runtime = {
+            "adapter-suite": "npm --prefix plugins/ca-pi/tools test",
+            "security-contract": "python .github/scripts/test_pi_security.py",
+            "package-contract": "python .github/scripts/test_pi_package.py",
+            "platform-contract": "python .github/scripts/test_pi_platform_contract.py --fixtures-only",
+        }
+        for identifier, invocation in runtime.items():
+            with self.subTest(contract=identifier):
+                command = prefix + identifier + state + invocation
+                self.assertEqual(commands.count(command), 1, "preserve the full runtime contract")
+                self.assertLess(commands.index(command), boundary,
+                                "runtime contracts must use reviewed source before candidate apply")
+        install = prefix + "toolchain-install" + state + (
+            "npm --prefix plugins/ca-pi/tools ci --ignore-scripts"
+        )
+        self.assertEqual(commands.count(install), 1)
+        for invocation in ("python tools/build-surface.py", "npm --prefix plugins/ca-pi/tools run build"):
+            with self.subTest(baseline_build=invocation):
+                build = prefix + "generated-artifacts" + state + invocation
+                before = commands[:boundary]
+                self.assertEqual(before.count(build), 1, "build the reviewed baseline before runtime tests")
+                self.assertLess(before.index(install), before.index(build))
+                for identifier, runtime_invocation in runtime.items():
+                    self.assertLess(before.index(build), before.index(prefix + identifier + state + runtime_invocation))
+
+    def test_prospective_phase_preserves_static_source_contracts(self):
+        validate = validation_job(WORKFLOW_PATH.read_text(encoding="utf-8"))
+        prospective = validate.split(' -- python .github/scripts/pi_promotion.py apply --candidate ', 1)[1]
+        lines = (line.strip().removeprefix("run: ") for line in prospective.splitlines())
+        commands = [line.split(' --state "$CA_PI_RECEIPT_STATE" -- ', 1)[1]
+                    for line in lines
+                    if line.startswith("python .github/scripts/pi_promotion.py contract ")]
+        generation = ("python tools/build-surface.py", "npm --prefix plugins/ca-pi/tools run build")
+        static = (
+            "npm --prefix plugins/ca-pi/tools run typecheck",
+            "python .github/scripts/test_host_descriptors.py",
+            "python .github/scripts/test_pi_parity.py",
+            "python tools/build-surface.py --check",
+            "python tools/build-host-packages.py --check",
+            "python .github/scripts/test_pi_security.py --contract-only",
+            "python .github/scripts/check_docs_contract.py",
+        )
+        for invocation in (*generation, *static):
+            with self.subTest(command=invocation):
+                self.assertEqual(commands.count(invocation), 1,
+                                 "preserve prospective generated, fixture and static validation")
+                if invocation in static:
+                    for build in generation:
+                        self.assertLess(commands.index(build), commands.index(invocation))
+        self.assertNotIn("pi_host_locks.py install", prospective)
+
     def test_checked_in_recipe_cannot_name_an_unapproved_runtime_write_path(self):
         module = load_module()
         with tempfile.TemporaryDirectory() as raw:

.github/workflows/pi-promotion.yml

@@ -115,26 +115,38 @@ jobs:
         run: |
           python .github/scripts/pi_host_locks.py capture --version ${{ needs.resolve.outputs.candidate }} --output "$RUNNER_TEMP/pi-candidate-lock"
           # The PENDING_REVIEW graph is data only. Dependency review is required before candidate code executes.
-      - name: Apply the declared prospective promotion in this disposable checkout
-        run: python .github/scripts/pi_promotion.py contract --id promotion-apply --state "$CA_PI_RECEIPT_STATE" -- python .github/scripts/pi_promotion.py apply --candidate "$CA_PI_CANDIDATE"
-        shell: bash
       - name: Install reviewed adapter toolchain
         run: python .github/scripts/pi_promotion.py contract --id toolchain-install --state "$CA_PI_RECEIPT_STATE" -- npm --prefix plugins/ca-pi/tools ci --ignore-scripts
         shell: bash
-      - name: Rebuild declared generated artifacts
+      - name: Build reviewed baseline artifacts
         shell: bash
         run: |
           python .github/scripts/pi_promotion.py contract --id generated-artifacts --state "$CA_PI_RECEIPT_STATE" -- python tools/build-surface.py
           python .github/scripts/pi_promotion.py contract --id generated-artifacts --state "$CA_PI_RECEIPT_STATE" -- npm --prefix plugins/ca-pi/tools run build
-      - name: Validate adapter, package, security, and documentation contracts
+      - name: Validate runtime contracts against the reviewed baseline
         shell: bash
         run: |
-          python .github/scripts/pi_promotion.py contract --id adapter-typecheck --state "$CA_PI_RECEIPT_STATE" -- npm --prefix plugins/ca-pi/tools run typecheck
           python .github/scripts/pi_promotion.py contract --id adapter-suite --state "$CA_PI_RECEIPT_STATE" -- npm --prefix plugins/ca-pi/tools test
           python .github/scripts/pi_promotion.py contract --id security-contract --state "$CA_PI_RECEIPT_STATE" -- python .github/scripts/test_pi_security.py
-          python .github/scripts/pi_promotion.py contract --id parity-contract --state "$CA_PI_RECEIPT_STATE" -- python .github/scripts/test_pi_parity.py
           python .github/scripts/pi_promotion.py contract --id package-contract --state "$CA_PI_RECEIPT_STATE" -- python .github/scripts/test_pi_package.py
           python .github/scripts/pi_promotion.py contract --id platform-contract --state "$CA_PI_RECEIPT_STATE" -- python .github/scripts/test_pi_platform_contract.py --fixtures-only
+      - name: Apply the declared prospective promotion in this disposable checkout
+        run: python .github/scripts/pi_promotion.py contract --id promotion-apply --state "$CA_PI_RECEIPT_STATE" -- python .github/scripts/pi_promotion.py apply --candidate "$CA_PI_CANDIDATE"
+        shell: bash
+      - name: Rebuild declared prospective artifacts without executing the candidate
+        shell: bash
+        run: |
+          python .github/scripts/pi_promotion.py contract --id generated-artifacts --state "$CA_PI_RECEIPT_STATE" -- python tools/build-surface.py
+          python .github/scripts/pi_promotion.py contract --id generated-artifacts --state "$CA_PI_RECEIPT_STATE" -- npm --prefix plugins/ca-pi/tools run build
+      - name: Validate prospective static and fixture contracts
+        shell: bash
+        run: |
+          python .github/scripts/pi_promotion.py contract --id adapter-typecheck --state "$CA_PI_RECEIPT_STATE" -- npm --prefix plugins/ca-pi/tools run typecheck
+          python .github/scripts/pi_promotion.py contract --id parity-contract --state "$CA_PI_RECEIPT_STATE" -- python .github/scripts/test_host_descriptors.py
+          python .github/scripts/pi_promotion.py contract --id parity-contract --state "$CA_PI_RECEIPT_STATE" -- python .github/scripts/test_pi_parity.py
+          python .github/scripts/pi_promotion.py contract --id generated-artifacts --state "$CA_PI_RECEIPT_STATE" -- python tools/build-surface.py --check
+          python .github/scripts/pi_promotion.py contract --id generated-artifacts --state "$CA_PI_RECEIPT_STATE" -- python tools/build-host-packages.py --check
+          python .github/scripts/pi_promotion.py contract --id security-contract --state "$CA_PI_RECEIPT_STATE" -- python .github/scripts/test_pi_security.py --contract-only
           python .github/scripts/pi_promotion.py contract --id docs-contract --state "$CA_PI_RECEIPT_STATE" -- python .github/scripts/check_docs_contract.py
       - name: Render the compact promotion receipt
         if: failure()

plugins/ca-pi/tools/test/inference-broker.test.ts

@@ -615,20 +615,29 @@ describe("#455 loopback inference broker", () => {
   // deleting the destruction loop stayed green. It is load-bearing — `server.close()` alone waits
   // for every idle keep-alive connection, so a child holding one open would keep an authorized
   // listener alive for as long as it liked.
-  test("close() destroys a still-open connection instead of waiting on it", async () => {
+  test.each(["ordinary close", "peer reset"])("close() destroys a still-open connection instead of waiting on it (%s)", async (outcome) => {
     const broker = await brokerFor(NONCE_A);
     const { hostname, port } = new URL(broker.baseUrl);
     const socket = connect({ host: hostname, port: Number(port) });
     openSockets.push(socket);
     await once(socket, "connect");
-    const peerClosed = once(socket, "close");
+    const socketErrors: NodeJS.ErrnoException[] = [];
+    socket.on("error", (error: NodeJS.ErrnoException) => socketErrors.push(error));
+    // A destructive peer shutdown may report ECONNRESET before close. Require the
+    // real close event even in that case, and reject every other error below.
+    const peerClosed = new Promise<void>((resolve) => socket.once("close", () => resolve()));
     // `server.close()` alone waits for every open connection, so without the destruction loop the
     // broker cannot finish closing while any child holds a socket open. The budget only has to
     // separate "destroys promptly" from "waits on the idle keep-alive forever" — a keep-alive
     // default is measured in minutes, so 15s keeps the discrimination while riding out a loaded
     // shared CI runner (this raced 3s and flaked three hosted macOS promotion cells in a row).
+    const closing = broker.close();
+    if (outcome === "peer reset") {
+      // Reproduce the socket error observed during destructive shutdown on macOS.
+      socket.emit("error", Object.assign(new Error("read ECONNRESET"), { code: "ECONNRESET" }));
+    }
     const closed = await Promise.race([
-      broker.close().then(() => "closed" as const),
+      closing.then(() => "closed" as const),
       new Promise<"hung">((resolve) => setTimeout(() => resolve("hung"), 15_000)),
     ]);
     expect(closed).toBe("closed");
@@ -637,5 +646,6 @@ describe("#455 loopback inference broker", () => {
       peerClosed,
       new Promise((_resolve, reject) => setTimeout(() => reject(new Error("connection survived close()")), 15_000)),
     ]);
+    for (const error of socketErrors) expect(error.code).toBe("ECONNRESET");
   });
 });