Prepare published cold Worker timer-redrive acceptance

.github/workflows/ci.yml

@@ -23,6 +23,10 @@ on:
         description: Run the reviewed published SDK/Server activity identity tuple
         type: boolean
         default: false
+      published_timer_redrive:
+        description: Run the reviewed published SDK/native timer-redrive tuple
+        type: boolean
+        default: false
 
 permissions:
   contents: read
@@ -32,6 +36,31 @@ concurrency:
   cancel-in-progress: ${{ github.event_name == 'pull_request' }}
 
 jobs:
+  published-timer-redrive:
+    if: github.event_name == 'workflow_dispatch' && inputs.published_timer_redrive
+    needs: [action-policy, lint]
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803  # v6
+        with:
+          persist-credentials: false
+      - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1  # v6
+        with:
+          python-version: "3.12.16"
+      - name: Bootstrap patched pip trust
+        uses: durable-workflow/.github/.github/actions/bootstrap-patched-pip@75587fc33ff40c34f32ef777293dad56bb6f48c7  # clean-trust
+        with:
+          clients: pip,python
+      - name: Run the published cold Worker timer-redrive case
+        run: python scripts/ci/published-activity-acceptance.py --scenario timer-redrive
+      - name: Retain published tuple and scenario evidence
+        if: always()
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a  # v7
+        with:
+          name: published-timer-redrive-${{ github.sha }}
+          path: published-activity-results/
+          retention-days: 7
   published-activity-acceptance:
     if: github.event_name == 'workflow_dispatch' && inputs.published_activity_acceptance
     needs: [action-policy, lint]

scripts/ci/activity_acceptance_consumer.py

@@ -1,4 +1,4 @@
-"""Install a closed wheel set, then await admission to the ordinary retry case."""
+"""Install a closed wheel set, then await admission to one fixed reviewed case."""
 
 from __future__ import annotations
 
@@ -23,6 +23,11 @@
 LIMIT = 64 * 1024 * 1024
 JUNIT = DATA / "identity.xml"
 CAPTURED_VERIFIER = ""
+CAPTURED_ENTRY = ""
+CASE_FILES = {
+    "activity-identity": {"test_activity_identity.py"},
+    "timer-redrive": {"test_native_timer_redrive.py", "redrive_timer_clock_workflow.py"},
+}
 
 
 def receive(maximum):
@@ -53,9 +58,11 @@ def run(arguments, environment, input_data=None):
 
 
 def validate_setup(packet):
-    if set(packet) != {"files", "wheels", "wheel_name", "wheel_sha256", "sdk_version"}:
+    if set(packet) != {"files", "wheels", "wheel_name", "wheel_sha256", "sdk_version", "scenario"}:
         raise ValueError("consumer setup shape differs")
-    expected_files = {"requirements.txt", "verify_activity_install.py", "test_activity_identity.py", "conftest.py"}
+    if packet["scenario"] not in CASE_FILES:
+        raise ValueError("consumer scenario differs")
+    expected_files = {"requirements.txt", "verify_activity_install.py", "conftest.py"} | CASE_FILES[packet["scenario"]]
     if set(packet["files"]) != expected_files or not 1 <= len(packet["wheels"]) <= 30:
         raise ValueError("consumer setup membership differs")
     if packet["wheel_name"] not in packet["wheels"]:
@@ -132,28 +139,27 @@ def main():
     verifier = retained_verifier()
     environment, python = prepare_environment()
     before = inventory(verifier, python, packet)
+    entry = captured_entry()
+    installed = entry["snapshot"](ENVIRONMENT)
+    site = ENVIRONMENT / "lib" / ("python" + str(sys.version_info.major) + "." + str(sys.version_info.minor))
+    distributions = sorted(
+        (item.metadata["Name"], item.version)
+        for item in importlib.metadata.distributions(path=[str(site / "site-packages")])
+    )
     ready = {
         "sdk_inventory": before,
         "python": sys.version,
         "environment": str(ENVIRONMENT),
-        "distributions": run(
-            [
-                str(python),
-                "-I",
-                "-B",
-                "-c",
-                "import importlib.metadata,json; print(json.dumps(sorted((d.metadata['Name'],"
-                "d.version) for d in importlib.metadata.distributions())))",
-            ],
-            environment,
-        ),
+        "distributions": json.dumps(distributions),
+        "environment_inventory": installed,
+        "environment_inventory_sha256": entry["digest"](installed),
     }
     print("ACTIVITY_CONSUMER_READY=" + json.dumps(ready, sort_keys=True), flush=True)
     admission = receive(8192)
     if set(admission) != {"runtime_ipv4", "token"}:
         raise ValueError("consumer admission shape differs")
     admission["server_url"] = await_runtime(admission["runtime_ipv4"])
-    output = execute_case(verifier, python, packet, before, environment, admission)
+    output = execute_case(verifier, python, packet, before, environment, admission, installed)
     print("ACTIVITY_CONSUMER_RESULT=" + json.dumps(output, sort_keys=True), flush=True)
     # Remain attached while the controller rechecks the same two IDs and IP.
     if receive(8192) != {"release": True}:
@@ -217,7 +223,31 @@ def prepare_environment():
     return environment, python
 
 
-def execute_case(verifier, python, packet, before, environment, admission):
+def captured_entry():
+    if not CAPTURED_ENTRY:
+        raise ValueError("controller did not bind the Git-captured entry")
+    namespace = {"__name__": "captured_entry"}
+    exec(compile(CAPTURED_ENTRY, "<captured entry>", "exec"), namespace)
+    return namespace
+
+
+def entry_source(mode, case_name, case_sha, installed, worker_entry=""):
+    source = CAPTURED_ENTRY
+    values = {
+        "MODE": mode, "INVENTORY_SHA256": captured_entry()["digest"](installed),
+        "CASE_SHA256": case_sha, "CASE_NAME": case_name, "WORKER_ENTRY": worker_entry,
+    }
+    for name, value in values.items():
+        placeholder = name + ' = ""'
+        if source.count(placeholder) != 1:
+            raise ValueError("captured entry binding placeholder differs")
+        source = source.replace(placeholder, name + " = " + repr(value), 1)
+    return source
+
+
+def execute_case(verifier, python, packet, before, environment, admission, installed):
+    redrive = packet["scenario"] == "timer-redrive"
+    test_name = "test_native_timer_redrive.py" if redrive else "test_activity_identity.py"
     environment.update(
         {
             "DURABLE_WORKFLOW_SERVER_URL": admission["server_url"],
@@ -226,34 +256,30 @@ def execute_case(verifier, python, packet, before, environment, admission):
             "PYTEST_DISABLE_PLUGIN_AUTOLOAD": "1",
         }
     )
+    if redrive:
+        environment["DURABLE_WORKFLOW_REDRIVE_CELL"] = "1"
+    worker_entry = entry_source("worker", test_name, packet["files"][test_name]["sha256"], installed)
+    parent_entry = entry_source("pytest", test_name, packet["files"][test_name]["sha256"], installed, worker_entry)
     result = subprocess.run(
         [
-            str(python),
+            sys.executable,
             "-I",
+            "-S",
             "-B",
-            "-m",
-            "pytest",
             "-c",
-            "/dev/null",
-            "-p",
-            "pytest_asyncio.plugin",
-            "-p",
-            "no:cacheprovider",
-            "--asyncio-mode=auto",
-            "--import-mode=importlib",
-            "-v",
-            "--capture=tee-sys",
-            str(DATA / "test_activity_identity.py"),
-            "--junitxml=" + str(JUNIT),
+            parent_entry,
         ],
         env=environment,
+        input=json.dumps({"inventory": installed}),
         capture_output=True,
         text=True,
-        timeout=90,
+        timeout=180 if redrive else 90,
     )
     after = inventory(verifier, python, packet)
     if after != before:
         raise ValueError("installed SDK inventory changed")
+    if captured_entry()["snapshot"](ENVIRONMENT) != installed:
+        raise ValueError("private install changed during the case")
     return {
         "complete": result.returncode == 0,
         "sdk_inventory": after,

scripts/ci/activity_acceptance_entry.py

@@ -0,0 +1,160 @@
+"""Captured entry: verify source and the private install before enabling imports."""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import json
+import os
+import stat
+import sys
+import types
+from pathlib import Path
+
+MODE = ""
+INVENTORY_SHA256 = ""
+CASE_SHA256 = ""
+CASE_NAME = ""
+WORKER_ENTRY = ""
+ENVIRONMENT = Path("/run/acceptance-env")
+DATA = Path("/run/acceptance-data")
+MAXIMUM_FILE = 64 * 1024 * 1024
+
+
+def digest(value):
+    return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
+
+
+def metadata(path):
+    value = path.lstat()
+    return [value.st_dev, value.st_ino, value.st_uid, value.st_gid, value.st_mode, value.st_size, value.st_mtime_ns]
+
+
+def captured_file(path, maximum=MAXIMUM_FILE):
+    descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW)
+    try:
+        before = os.fstat(descriptor)
+        if not stat.S_ISREG(before.st_mode) or before.st_size > maximum:
+            raise ValueError("captured file shape or size differs")
+        chunks, size = [], 0
+        while block := os.read(descriptor, min(1024 * 1024, maximum + 1 - size)):
+            chunks.append(block)
+            size += len(block)
+            if size > maximum:
+                raise ValueError("captured file exceeds its bound")
+        after = os.fstat(descriptor)
+        if (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns) != (
+            after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns,
+        ):
+            raise ValueError("captured file changed during its read")
+        return b"".join(chunks)
+    finally:
+        os.close(descriptor)
+
+
+def snapshot(root):
+    before = metadata(root)
+    if not stat.S_ISDIR(before[4]):
+        raise ValueError("private install root is not a directory")
+    entries = {}
+    for path in sorted(root.rglob("*")):
+        info = metadata(path)
+        item = {"metadata": info}
+        if stat.S_ISREG(info[4]):
+            item["sha256"] = hashlib.sha256(captured_file(path)).hexdigest()
+        elif stat.S_ISLNK(info[4]):
+            item["target"] = os.readlink(path)
+        elif not stat.S_ISDIR(info[4]):
+            raise ValueError("private install contains an unsupported entry")
+        entries[path.relative_to(root).as_posix()] = item
+    if before != metadata(root):
+        raise ValueError("private install root changed during capture")
+    return {"root": before, "entries": entries}
+
+
+def guarded_source(path, expected):
+    source = captured_file(path, 256 * 1024)
+    if hashlib.sha256(source).hexdigest() != expected:
+        raise ValueError("captured case source differs")
+    return source
+
+
+def source_module(source):
+    module = types.ModuleType("captured_published_case")
+    module.__file__ = str(DATA / CASE_NAME)
+    sys.modules[module.__name__] = module
+    exec(compile(source, module.__file__, "exec", optimize=0), module.__dict__)
+    return module
+
+
+def run_pytest(source, inventory):
+    # pytest never opens the writable test or conftest as executable code.
+    import pytest
+
+    class CapturedModule(pytest.Module):
+        def _getobj(self):
+            module = source_module(source)
+            module._COLD_BOOTSTRAP = WORKER_ENTRY
+            module._COLD_INVENTORY = inventory
+            module._COLD_PYTHON = sys.executable
+            return module
+
+    class CapturedPlugin:
+        @pytest.hookimpl(tryfirst=True)
+        def pytest_pycollect_makemodule(self, module_path, parent):
+            if module_path != DATA / CASE_NAME:
+                raise ValueError("pytest requested an uncaptured module")
+            return CapturedModule.from_parent(parent, path=module_path)
+
+        @pytest.fixture(scope="session")
+        def server_url(self):
+            return os.environ["DURABLE_WORKFLOW_SERVER_URL"]
+
+        @pytest.fixture(scope="session")
+        def server_token(self):
+            return os.environ["DURABLE_WORKFLOW_AUTH_TOKEN"]
+
+    return pytest.main([
+        "-c", "/dev/null", "--noconftest", "-p", "pytest_asyncio.plugin", "-p", "no:cacheprovider",
+        "-o", "asyncio_default_fixture_loop_scope=function",
+        "--asyncio-mode=auto", "--import-mode=importlib", "-v", "--capture=tee-sys", str(DATA / CASE_NAME),
+        "--junitxml=" + str(DATA / "identity.xml"),
+    ], plugins=[CapturedPlugin()])
+
+
+def main():
+    if not (sys.flags.isolated and sys.flags.no_site and sys.dont_write_bytecode) or sys.flags.optimize:
+        raise ValueError("captured entry requires isolated no-site Python with active assertions")
+    payload = sys.stdin.buffer.read(4 * 1024 * 1024 + 1)
+    if len(payload) > 4 * 1024 * 1024:
+        raise ValueError("captured entry input exceeds its bound")
+    packet = json.loads(payload)
+    if set(packet) != ({"inventory", "worker"} if MODE == "worker" else {"inventory"}):
+        raise ValueError("captured entry input shape differs")
+    inventory = packet["inventory"]
+    if digest(inventory) != INVENTORY_SHA256:
+        raise ValueError("private install commitment differs")
+    source = guarded_source(DATA / CASE_NAME, CASE_SHA256)
+    if snapshot(ENVIRONMENT) != inventory:
+        raise ValueError("private install inventory changed")
+    # No site.addsitedir or site.main: .pth and sitecustomize are never run.
+    site = ENVIRONMENT / "lib" / ("python" + str(sys.version_info.major) + "." + str(sys.version_info.minor))
+    site /= "site-packages"
+    if not site.is_dir() or site.is_symlink():
+        raise ValueError("verified site directory differs")
+    sys.path.insert(0, str(site))
+    sys.prefix = sys.exec_prefix = str(ENVIRONMENT)
+    if MODE == "worker":
+        asyncio.run(source_module(source).cold_worker(packet["worker"]))
+        result = 0
+    elif MODE == "pytest":
+        result = run_pytest(source, inventory)
+    else:
+        raise ValueError("captured entry mode differs")
+    if snapshot(ENVIRONMENT) != inventory:
+        raise ValueError("private install changed during execution")
+    raise SystemExit(result)
+
+
+if __name__ == "__main__":
+    main()

scripts/ci/published-activity-acceptance.md

@@ -19,6 +19,11 @@ SDK wheel member, complete distribution metadata and installed RECORD, and uses
 isolated Python with explicit pytest plugins. Installer-created metadata and
 console scripts have exact expected bytes. The checker stays in the trusted
 bootstrap's memory through the case. The checkout is never installed as the SDK.
+The complete private installed-file inventory is captured before SDK admission.
+Parent pytest uses the read-only base interpreter with `-I -S -B`; its literal
+entry verifies captured test bytes and the installed inventory before enabling
+the exact private site directory. A captured collector and fixed fixtures avoid
+executing mutable test/conftest paths, and no `.pth` startup is processed.
 
 The published native runtime and official Python consumer share one owned
 internal network. Neither publishes a port or receives a host mount. The