Fix qualified PyPI registry readback

.github/workflows/forge-production-release.yml

@@ -238,10 +238,15 @@ jobs:
             *) echo "Unexpected PyPI identity lookup status: $status_code" >&2; exit 1 ;;
           esac
           python3 - "$wheel" "$sdist" release-input/dist/SHA256SUMS pypi.json <<'PY'
-          import json, sys
-          expected = {line.split()[1]: line.split()[0] for line in open(sys.argv[3])}
+          import json
+          from pathlib import Path
+          import sys
+
+          from scripts.pypi_distribution_readback import expected_digests_from_sha256sums
+
+          wanted = tuple(sys.argv[1:3])
+          expected = expected_digests_from_sha256sums(Path(sys.argv[3]), wanted)
           files = {item['filename']: item['digests']['sha256'] for item in json.load(open(sys.argv[4]))['urls']}
-          wanted = sys.argv[1:3]
           if any(name in files and files[name] != expected[name] for name in wanted): raise SystemExit('PUBLICATION_IDENTITY_CONFLICT')
           if all(name in files and files[name] == expected[name] for name in wanted): raise SystemExit(0)
           raise SystemExit('partial PyPI release is a publication identity conflict')
@@ -299,8 +304,35 @@ jobs:
           PY
             )"
             curl --fail --silent --show-error --location "$artifact_url" -o "registry-readback/$artifact"
-            (cd registry-readback && grep " $artifact$" ../release-input/dist/SHA256SUMS | sha256sum --check)
           done
+          python3 - "$VERSION" release-input/dist/SHA256SUMS registry-readback <<'PY'
+          from hashlib import sha256
+          from pathlib import Path
+          import sys
+
+          from scripts.pypi_distribution_readback import (
+              distribution_filenames,
+              expected_digests_from_sha256sums,
+          )
+
+          version, sums, root = sys.argv[1], Path(sys.argv[2]), Path(sys.argv[3])
+          filenames = distribution_filenames(version)
+          expected = expected_digests_from_sha256sums(sums, filenames)
+          if root.is_symlink() or not root.is_dir():
+              raise SystemExit("registry download root is unavailable or unsafe")
+          for filename in filenames:
+              artifact = root / filename
+              if artifact.is_symlink() or not artifact.is_file():
+                  raise SystemExit(f"registry download is unavailable or unsafe: {filename}")
+              digest = sha256()
+              with artifact.open("rb") as stream:
+                  for chunk in iter(lambda: stream.read(1024 * 1024), b""):
+                      digest.update(chunk)
+              if digest.hexdigest() != expected[filename]:
+                  raise SystemExit(
+                      f"downloaded PyPI artifact conflicts with qualified bytes: {filename}"
+                  )
+          PY
           python3 - registry-readback "$wheel" "$sdist" > registry-readback-digests.json <<'PY'
           from hashlib import sha256
           import json

tests/test_production_release_workflow.py

@@ -1,6 +1,7 @@
 """Guard the durable release-state ordering expected by production delivery."""
 
 import importlib.util
+from hashlib import sha256
 import json
 import os
 from pathlib import Path
@@ -51,6 +52,9 @@ def test_prepublication_operation_binds_exact_artifacts_and_terminal_states(self
             'gh release download "$TAG" --pattern "$QUALIFIED" --dir "$RUNNER_TEMP/forge-qualified-readback"',
             publish_job,
         )
+        self.assertIn("expected_digests_from_sha256sums(Path(sys.argv[3]), wanted)", publish_job)
+        self.assertIn("echo 'already_published=true' >> \"$GITHUB_OUTPUT\"", publish_job)
+        self.assertIn("if: steps.existing.outputs.already_published != 'true'", publish_job)
         self.assertIn("forge-release-published-$VERSION-$SOURCE_SHA.json", workflow)
         self.assertIn("needs: [release-context, build-and-qualify, publish-pypi, registry-readback-and-published-evidence]", workflow)
         self.assertIn("gh release download \"$TAG\" --pattern \"$PUBLISHED_RECEIPT\" --dir published-readback", workflow)
@@ -75,6 +79,15 @@ def test_prepublication_operation_binds_exact_artifacts_and_terminal_states(self
             registry_job.index("python3 scripts/pypi_distribution_readback.py"),
             registry_job.index('for artifact in "$wheel" "$sdist"; do'),
         )
+        verifier = 'python3 - "$VERSION" release-input/dist/SHA256SUMS registry-readback'
+        self.assertIn(verifier, registry_job)
+        self.assertIn("expected_digests_from_sha256sums", registry_job)
+        self.assertNotIn('grep " $artifact$"', registry_job)
+        self.assertNotIn("sha256sum --check", registry_job)
+        self.assertLess(
+            registry_job.index('for artifact in "$wheel" "$sdist"; do'),
+            registry_job.index(verifier),
+        )
         self.assertLess(
             workflow.index("python3 scripts/pypi_distribution_readback.py"),
             workflow.index("--mark-published"),
@@ -104,6 +117,104 @@ def _cleanup_step() -> str:
         script_end = workflow.index("      - uses: actions/upload-artifact", script_start)
         return textwrap.dedent(workflow[script_start:script_end])
 
+    @staticmethod
+    def _registry_download_verifier() -> str:
+        workflow = Path(".github/workflows/forge-production-release.yml").read_text(encoding="utf-8")
+        marker = '          python3 - "$VERSION" release-input/dist/SHA256SUMS registry-readback <<\'PY\'\n'
+        script_start = workflow.index(marker) + len(marker)
+        script_end = workflow.index(
+            "          PY\n          python3 - registry-readback",
+            script_start,
+        )
+        return textwrap.dedent(workflow[script_start:script_end])
+
+    @staticmethod
+    def _existing_publication_verifier() -> str:
+        workflow = Path(".github/workflows/forge-production-release.yml").read_text(encoding="utf-8")
+        marker = '          python3 - "$wheel" "$sdist" release-input/dist/SHA256SUMS pypi.json <<\'PY\'\n'
+        script_start = workflow.index(marker) + len(marker)
+        script_end = workflow.index(
+            "          PY\n          echo 'already_published=true'",
+            script_start,
+        )
+        return textwrap.dedent(workflow[script_start:script_end])
+
+    def test_existing_publication_verifier_normalizes_qualified_paths_before_skip(self) -> None:
+        wheel = f"forge_autonomy-{self.version}-py3-none-any.whl"
+        sdist = f"forge_autonomy-{self.version}.tar.gz"
+        wheel_digest = sha256(b"qualified Forge wheel").hexdigest()
+        sdist_digest = sha256(b"qualified Forge source distribution").hexdigest()
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            sums = root / "SHA256SUMS"
+            sums.write_text(
+                f"{wheel_digest}  dist/{wheel}\n{sdist_digest}  dist/{sdist}\n",
+                encoding="utf-8",
+            )
+            document = root / "pypi.json"
+            document.write_text(json.dumps({"urls": [
+                {"filename": wheel, "digests": {"sha256": wheel_digest}},
+                {"filename": sdist, "digests": {"sha256": sdist_digest}},
+            ]}), encoding="utf-8")
+            command = [sys.executable, "-", wheel, sdist, str(sums), str(document)]
+            script = self._existing_publication_verifier()
+
+            exact = subprocess.run(
+                command, input=script, check=False, capture_output=True, text=True,
+            )
+            self.assertEqual(0, exact.returncode, exact.stderr)
+
+            document.write_text(json.dumps({"urls": [
+                {"filename": wheel, "digests": {"sha256": "f" * 64}},
+                {"filename": sdist, "digests": {"sha256": sdist_digest}},
+            ]}), encoding="utf-8")
+            conflict = subprocess.run(
+                command, input=script, check=False, capture_output=True, text=True,
+            )
+            self.assertNotEqual(0, conflict.returncode)
+            self.assertIn("PUBLICATION_IDENTITY_CONFLICT", conflict.stderr)
+
+            document.write_text(json.dumps({"urls": [
+                {"filename": wheel, "digests": {"sha256": wheel_digest}},
+            ]}), encoding="utf-8")
+            partial = subprocess.run(
+                command, input=script, check=False, capture_output=True, text=True,
+            )
+            self.assertNotEqual(0, partial.returncode)
+            self.assertIn("partial PyPI release", partial.stderr)
+
+    def test_registry_download_verifier_uses_qualified_dist_paths_and_fails_closed(self) -> None:
+        wheel = f"forge_autonomy-{self.version}-py3-none-any.whl"
+        sdist = f"forge_autonomy-{self.version}.tar.gz"
+        wheel_bytes = b"qualified Forge wheel"
+        sdist_bytes = b"qualified Forge source distribution"
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            downloads = root / "registry-readback"
+            downloads.mkdir()
+            (downloads / wheel).write_bytes(wheel_bytes)
+            (downloads / sdist).write_bytes(sdist_bytes)
+            sums = root / "SHA256SUMS"
+            sums.write_text(
+                f"{sha256(wheel_bytes).hexdigest()}  dist/{wheel}\n"
+                f"{sha256(sdist_bytes).hexdigest()}  dist/{sdist}\n",
+                encoding="utf-8",
+            )
+            command = [sys.executable, "-", self.version, str(sums), str(downloads)]
+            script = self._registry_download_verifier()
+
+            verified = subprocess.run(
+                command, input=script, check=False, capture_output=True, text=True,
+            )
+            self.assertEqual(0, verified.returncode, verified.stderr)
+
+            (downloads / wheel).write_bytes(b"different wheel bytes")
+            conflict = subprocess.run(
+                command, input=script, check=False, capture_output=True, text=True,
+            )
+            self.assertNotEqual(0, conflict.returncode)
+            self.assertIn("conflicts with qualified bytes", conflict.stderr)
+
     def _fixture(
         self,
         root: Path,