Bind pre-1 qualification tooling without changing product source

scripts/pre1_harness_binding.py

@@ -13,14 +13,21 @@
 
 
 TOOL_ONLY_PATHS = frozenset({
+    "qualification/pre1-cases.json",
+    "scripts/pre1_comparative_topology.py",
+    "scripts/pre1_installed_discovery.py",
     "scripts/run_pre1_qualification_case.py",
     "scripts/test_pre1_qualification_case.py",
     "scripts/pre1_harness_binding.py",
     "scripts/pre1_environment_contract.py",
     "scripts/pre1_package_execution.py",
     "scripts/test_pre1_package_execution.py",
+    "scripts/test_pre1_prepared_home.py",
+    "scripts/test_pre1_comparative_topology.py",
+    "scripts/test_pre1_installed_discovery.py",
     "scripts/run_php83_local_ci.py",
     "scripts/test_php83_local_ci.py",
+    "scripts/prepare_pre1_minimum_runtime.py",
 })
 
 
@@ -32,6 +39,26 @@
 })
 
 
+# This is an exact append-only qualification note, not a documentation wildcard.
+# Both the released prefix and the reviewed addition remain immutable here.
+REVIEWED_DOCUMENT_APPENDICES = {
+    "OPERATIONS.md": {
+        "base_sha256": "79602ed2ddc66383fe7777415e9cb125b29134e3bc4a553edce5f3aab796ea45",
+        "append_sha256": "bc0a713ae89a67c8cb244d25edc15c7b6fd384ead6b0c54f4d39e59ffdc0f8b7",
+    },
+}
+
+
+def validate_document_appendix(root: Path, path: str, product: str, harness: str) -> None:
+    expected = REVIEWED_DOCUMENT_APPENDICES[path]
+    before = git(root, "show", f"{product}:{path}")
+    after = git(root, "show", f"{harness}:{path}")
+    if (hashlib.sha256(before).hexdigest() != expected["base_sha256"]
+            or not after.startswith(before)
+            or hashlib.sha256(after[len(before):]).hexdigest() != expected["append_sha256"]):
+        raise ValueError("qualification documentation differs from reviewed append-only note")
+
+
 def git(root: Path, *argv: str) -> bytes:
     return subprocess.check_output(
         ["git", *argv], cwd=root, stderr=subprocess.PIPE, timeout=30
@@ -75,5 +102,9 @@ def validate_harness_source(
         root, "diff", "--no-renames", "--name-only", "-z",
         product_source_commit, binding["harness_source_commit"], "--"
     ).decode().split("\0")
-    if set(filter(None, changes)) - (TOOL_ONLY_PATHS | CI_ONLY_PATHS):
+    remaining = set(filter(None, changes)) - (TOOL_ONLY_PATHS | CI_ONLY_PATHS)
+    for path in remaining & REVIEWED_DOCUMENT_APPENDICES.keys():
+        validate_document_appendix(root, path, product_source_commit, binding["harness_source_commit"])
+        remaining.remove(path)
+    if remaining:
         raise ValueError("qualification tooling revision changes frozen product source")

scripts/pre1_package_execution.py

@@ -341,6 +341,13 @@ def validate_binding_context(value: dict, context: dict) -> None:
     if any(value.get(k) != context[k] for k in ("component", "runtime", "target")) or value.get("bindings") != {k: context[k] for k in BINDINGS}:
         raise ValueError("package execution candidate/environment/runtime binding differs")
 
+def prepared_package_home() -> Path:
+    """Keep the immutable preparation boundary separate from per-case HOME."""
+    return safe_path(
+        Path(os.environ.get("IICP_PRE1_PREPARED_PACKAGE_HOME", os.environ["HOME"]))
+    )
+
+
 def validate_binding(value: dict, context: dict, artifact: Path, root: Path,
                      vendor_artifact: Path | None = None) -> Path:
     if context["component"] == "management":
@@ -350,7 +357,7 @@ def validate_binding(value: dict, context: dict, artifact: Path, root: Path,
     validate_binding_identity(value)
     validate_binding_context(value, context)
     workspace = safe_path(Path(value["workspace"]))
-    home = safe_path(Path(os.environ["HOME"]))
+    home = prepared_package_home()
     installed = safe_path(Path(value["installed_package"]))
     validate_workspace_boundary(workspace, home, installed, root)
     if value["artifact_sha256"] != file_digest(artifact) or value["installed_payload_sha256"] != digest(installed_payload(artifact, installed, context["component"])):
@@ -505,7 +512,7 @@ def validate_summary_identity(summary: dict) -> None:
 def write_case_proof(value: dict) -> Path:
     """Publish a complete sidecar atomically, without overwriting earlier evidence."""
     path = Path(os.environ["IICP_PRE1_CASE_PROOF_OUTPUT"])
-    home = safe_path(Path(os.environ["HOME"]))
+    home = prepared_package_home()
     parent = safe_path(path.parent)
     if not path.is_absolute() or not parent.is_relative_to(home) or path.exists() or path.is_symlink():
         raise ValueError("case proof output is unsafe or already exists")
@@ -597,7 +604,7 @@ def rust_fixtures(root: Path, installed: Path) -> dict[str, str]:
 
 def create_rust_binding(root, workspace, installed, artifact, runtime, target, bindings, vendor_artifact):
     validate_immutable_bindings(bindings)
-    validate_workspace_boundary(safe_path(workspace), safe_path(Path(os.environ["HOME"])),
+    validate_workspace_boundary(safe_path(workspace), prepared_package_home(),
                                safe_path(installed), root)
     payload, deps = verify_rust_payload(workspace, installed, artifact, vendor_artifact)
     value = {"schema": SCHEMA, "component": "client-rust", "runtime": runtime,
@@ -618,7 +625,7 @@ def validate_rust_binding(value, context, artifact, root, vendor_artifact):
     validate_binding_context(value, context)
     workspace = safe_path(Path(value["workspace"]))
     installed = safe_path(Path(value["installed_package"]))
-    validate_workspace_boundary(workspace, safe_path(Path(os.environ["HOME"])), installed, root)
+    validate_workspace_boundary(workspace, prepared_package_home(), installed, root)
     payload, deps = verify_rust_payload(workspace, installed, artifact, vendor_artifact)
     expected = {"artifact_sha256": file_digest(artifact),
                 "vendor_artifact_sha256": file_digest(vendor_artifact),
@@ -796,7 +803,7 @@ def management_assertions(name, source):
 def stage_management_consumer(root: Path, artifact: Path, workspace: Path) -> dict:
     """Prepare only; the caller owns dependency acquisition and isolation."""
     root, artifact, workspace = safe_path(root), safe_path(artifact), safe_path(workspace)
-    home = safe_path(Path(os.environ["HOME"]))
+    home = prepared_package_home()
     if workspace == home or not workspace.is_relative_to(home) or workspace.is_relative_to(root) or any(workspace.iterdir()):
         raise ValueError("Management consumer workspace is not empty and run-isolated")
     expected = rust_archive_files(artifact, artifact.stem + "/")
@@ -820,7 +827,7 @@ def stage_management_consumer(root: Path, artifact: Path, workspace: Path) -> di
 def validate_management_consumer(root: Path, artifact: Path, workspace: Path, binding: dict) -> Path:
     """Recheck artifact and reviewed assertions, even after a forged rehash."""
     workspace = safe_path(workspace)
-    home = safe_path(Path(os.environ["HOME"]))
+    home = prepared_package_home()
     if workspace == home or not workspace.is_relative_to(home) or workspace.is_relative_to(root.resolve()):
         raise ValueError("Management consumer workspace is not run-isolated")
     payload = safe_path(workspace / "payload")

scripts/run_pre1_qualification_case.py

@@ -433,7 +433,7 @@ def validate_runtime(runtime: str, runtime_row: dict, manifest: dict) -> None:
 
 
 def command_environment(runtime_row: dict, runtime: str) -> dict[str, str]:
-    allowed = {"PATH", "HOME", "IICP_HOME", "LANG", "LC_ALL", "SYSTEMROOT", "WINDIR", "COMSPEC", "PATHEXT", "TEMP", "TMP", "TMPDIR", "RUSTUP_HOME"}
+    allowed = {"PATH", "HOME", "IICP_HOME", "LANG", "LC_ALL", "SYSTEMROOT", "WINDIR", "COMSPEC", "PATHEXT", "TEMP", "TMP", "TMPDIR", "RUSTUP_HOME", "IICP_PRE1_PREPARED_PACKAGE_HOME"}
     env = {key: value for key, value in os.environ.items() if key.upper() in allowed}
     for name in (
         "IICP_PRE1_CELL_ID",

scripts/test_pre1_prepared_home.py

@@ -0,0 +1,81 @@
+"""Preparation ownership must not inherit the private execution HOME."""
+import ast
+import json
+import os
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+import pre1_package_execution as adapter
+
+
+class PreparedHomeTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.root = Path(self.tmp.name).resolve()
+        self.prepared = self.root / "prepared"
+        self.case = self.root / "case"
+        self.prepared.mkdir(mode=0o700)
+        self.case.mkdir(mode=0o700)
+
+    def test_legacy_preparation_defaults_to_home(self):
+        with patch.dict(os.environ, {"HOME": str(self.prepared)}, clear=True):
+            self.assertEqual(adapter.prepared_package_home(), self.prepared)
+
+    def test_execution_home_does_not_replace_preparation_owner(self):
+        workspace = self.prepared / "workspace"
+        installed = workspace / "payload"
+        installed.mkdir(parents=True)
+        source = self.root / "source"
+        source.mkdir()
+        env = {"HOME": str(self.case), "IICP_PRE1_PREPARED_PACKAGE_HOME": str(self.prepared)}
+        with patch.dict(os.environ, env, clear=True):
+            self.assertEqual(adapter.prepared_package_home(), self.prepared)
+            adapter.validate_workspace_boundary(workspace, adapter.prepared_package_home(), installed, source)
+            with self.assertRaises(ValueError):
+                adapter.validate_workspace_boundary(workspace, self.case, installed, source)
+
+    def test_wrong_explicit_preparation_boundary_is_rejected(self):
+        workspace = self.prepared / "workspace"
+        installed = workspace / "payload"
+        installed.mkdir(parents=True)
+        with patch.dict(os.environ, {"HOME": str(self.prepared),
+                "IICP_PRE1_PREPARED_PACKAGE_HOME": str(self.case)}, clear=True):
+            with self.assertRaises(ValueError):
+                adapter.validate_workspace_boundary(workspace, adapter.prepared_package_home(), installed, self.root / "source")
+
+    def test_symlink_preparation_boundary_is_rejected(self):
+        alias = self.root / "alias"
+        alias.symlink_to(self.prepared, target_is_directory=True)
+        with patch.dict(os.environ, {"HOME": str(self.case),
+                "IICP_PRE1_PREPARED_PACKAGE_HOME": str(alias)}, clear=True):
+            with self.assertRaises(ValueError):
+                adapter.prepared_package_home()
+
+    def test_driver_preserves_preparation_boundary(self):
+        # Execute the actual owning function without importing its driver (which
+        # performs unrelated qualification-map initialization at import time).
+        path = Path(adapter.__file__).with_name("run_pre1_qualification_case.py")
+        tree = ast.parse(path.read_text())
+        function = next(n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name == "command_environment")
+        namespace = {"os": os, "Path": Path, "json": json, "COMPONENT": "fixture"}
+        exec(compile(ast.Module(body=[function], type_ignores=[]), str(path), "exec"), namespace)
+        manifest = self.root / "candidate.json"
+        manifest.write_text(json.dumps({"components": [{"id": "fixture", "artifacts": [{"kind": "release-manifest", "name": "release.json"}]}]}))
+        env = {"HOME": str(self.case), "IICP_HOME": str(self.case),
+               "IICP_PRE1_PREPARED_PACKAGE_HOME": str(self.prepared)}
+        for node in ast.walk(function):
+            if isinstance(node, ast.For) and isinstance(node.iter, ast.Tuple):
+                for value in node.iter.elts:
+                    if isinstance(value, ast.Constant) and isinstance(value.value, str):
+                        env[value.value] = "diagnostic-fixture"
+        env.update(IICP_PRE1_CANDIDATE_MANIFEST=str(manifest), IICP_PRE1_ARTIFACT_ROOT=str(self.root))
+        with patch.dict(os.environ, env, clear=True):
+            result = namespace["command_environment"]({"env": {}, "programs": {name: "/runtime/" + name for name in ("python", "node", "cargo", "rustc", "php")}}, "fixture")
+        self.assertEqual(result["HOME"], str(self.case))
+        self.assertEqual(result["IICP_PRE1_PREPARED_PACKAGE_HOME"], str(self.prepared))
+
+
+if __name__ == "__main__":
+    unittest.main()