Limit guarded demo updates to owned signed runtime files

Retain the rejected VERSION rollback result. Replace general installer execution with pinned-signature archive selection and atomic binary/VERSION swaps, preserve unrelated distribution assets, and fail closed on missing matching native acceptance before production mutation.

Change-source: pulse-maintainer

.github/scripts/demo-runtime-transaction.py

@@ -14,6 +14,7 @@ import socket
 import stat
 import subprocess
 import sys
+import tarfile
 import time
 from urllib.parse import urlsplit
 
@@ -24,6 +25,7 @@ SERVICES = ("pulse", "pulse-relay", "caddy")
 ROOT = Path("/var/lib/pulse-deploy/demo")
 LOCK = Path("/var/lib/pulse-deploy/relay/deploy.lock")
 PATHS = {"binary": Path("/opt/pulse/bin/pulse"),
+         "version": Path("/opt/pulse/VERSION"),
          "unit": Path("/etc/systemd/system/pulse.service"),
          "dropins": Path("/etc/systemd/system/pulse.service.d"),
          "data": Path("/etc/pulse")}
@@ -40,6 +42,7 @@ DEMO_HISTORY = ("ai_incidents.json", "ai_incidents.json.tmp",
                 "alerts/events.db", "alerts/events.db-shm", "alerts/events.db-wal")
 CRASH = re.compile(r"\bpanic\b|\bfatal error\b|segmentation fault|core dumped|watchdog timeout", re.I)
 TERMINAL = {"committed", "healthy_noop", "rolled_back", "refused"}
+MAX_BINARY = 128 * 1024 * 1024
 
 
 class Failure(Exception):
@@ -62,7 +65,7 @@ def atomic_json(path, value):
 
 
 def validate(request):
-    if set(request) != {"mode", "hostname", "local_url", "public_url", "version", "profile", "control_sha", "run_id", "run_attempt", "installer_sha256"}:
+    if set(request) != {"mode", "hostname", "local_url", "public_url", "version", "profile", "control_sha", "run_id", "run_attempt", "binary_sha256", "version_sha256"}:
         raise Failure("request-shape")
     if request["mode"] not in {"update", "recover"}:
         raise Failure("request-mode")
@@ -92,9 +95,47 @@ def validate(request):
         pattern = r"[1-9][0-9]{0,3}" if name in COUNT_KEYS else r"[1-9][0-9]{0,3}[smhd]"
         if not isinstance(value, str) or not re.fullmatch(pattern, value):
             raise Failure("request-profile")
-    digest = request["installer_sha256"]
-    if (request["mode"] == "update" and not re.fullmatch(r"[0-9a-f]{64}", digest)) or (request["mode"] == "recover" and digest != ""):
-        raise Failure("request-installer")
+    for name in ("binary_sha256", "version_sha256"):
+        digest = request[name]
+        if (not isinstance(digest, str)
+                or (request["mode"] == "update" and not re.fullmatch(r"[0-9a-f]{64}", digest))
+                or (request["mode"] == "recover" and digest != "")):
+            raise Failure("request-runtime")
+
+
+def runtime_members(archive, version):
+    """Select two regular files from an already signature-verified archive.
+
+    Never extract archive paths or execute its installer. This demo-only route
+    does not update bundled agents, scripts, helpers, units or timer assets.
+    """
+    files = {}
+    count = 0
+    with tarfile.open(archive, "r:gz") as stream:
+        for member in stream:
+            count += 1
+            if count > 4096:
+                raise Failure("archive-member-bound")
+            name = member.name.removeprefix("./")
+            if name not in {"pulse", "bin/pulse", "VERSION"}:
+                continue
+            label = "version" if name == "VERSION" else "binary"
+            if (label in files or not member.isfile() or member.issparse()
+                    or not 0 < member.size <= (128 if label == "version" else MAX_BINARY)):
+                raise Failure("archive-runtime-shape")
+            with stream.extractfile(member) as source:
+                files[label] = source.read(member.size + 1)
+            if len(files[label]) != member.size:
+                raise Failure("archive-runtime-size")
+    if set(files) != {"binary", "version"}:
+        raise Failure("archive-runtime-missing")
+    if files["version"].strip() not in {version.encode(), version.removeprefix("v").encode()}:
+        raise Failure("archive-runtime-version")
+    # The demo estate is the existing Linux/amd64 server, not a new platform.
+    data = files["binary"]
+    if len(data) < 64 or data[:6] != b"\x7fELF\x02\x01" or data[18:20] != b"\x3e\x00":
+        raise Failure("archive-runtime-architecture")
+    return files
 
 
 def copy_path(src, dst):
@@ -175,6 +216,8 @@ class Host:
                 raise Failure("estate-identity")
         if PATHS["dropins"].is_symlink():
             raise Failure("dropin-identity")
+        if PATHS["version"].is_symlink() or any(p.is_symlink() for p in PATHS["binary"].parents):
+            raise Failure("distribution-identity")
         for path in fields.get("DropInPaths", "").split():
             if Path(path).parent != PATHS["dropins"] or Path(path).is_symlink():
                 raise Failure("unsupported-dropin")
@@ -244,11 +287,44 @@ class Host:
         self.command(["systemctl", "start", "pulse"], timeout=90)
 
     def install(self, attempt, request):
-        installer = attempt / "installer.sh"
-        if hashlib.sha256(installer.read_bytes()).hexdigest() != request["installer_sha256"]:
-            raise Failure("installer-identity")
-        self.command(["bash", str(installer), "--version", request["version"]], timeout=900,
-                     env={"PATH": "/usr/sbin:/usr/bin:/sbin:/bin", "HOME": "/root", "PULSE_SERVICE_NAME": "pulse"})
+        # Trust is checked against the pinned release key in CI, before these
+        # exact bytes enter the authenticated SSH envelope. Check the retained
+        # byte identities again before any swap; no arbitrary shell is run.
+        staged = {label: attempt / label for label in ("binary", "version")}
+        for label, path in staged.items():
+            if path.is_symlink() or not path.is_file():
+                raise Failure("runtime-payload-shape")
+            if not 0 < path.stat().st_size <= (128 if label == "version" else MAX_BINARY):
+                raise Failure("runtime-payload-bound")
+            if hashlib.sha256(path.read_bytes()).hexdigest() != request[label + "_sha256"]:
+                raise Failure("runtime-payload-identity")
+        if staged["version"].read_bytes().strip() not in {
+                request["version"].encode(), request["version"].removeprefix("v").encode()}:
+            raise Failure("runtime-payload-version")
+        for label, source in staged.items():
+            destination = PATHS[label]
+            # Sibling staging keeps replacement atomic; owned recovery removes
+            # the sibling if a failure interrupts either of the two swaps.
+            temporary = destination.with_name(destination.name + ".demo-new")
+            if temporary.exists() or temporary.is_symlink():
+                raise Failure("runtime-staging-exists")
+            try:
+                with open(source, "rb") as src, open(temporary, "xb") as dst:
+                    shutil.copyfileobj(src, dst)
+                    owner = PATHS["binary"].stat()
+                    os.fchown(dst.fileno(), owner.st_uid, owner.st_gid)
+                    os.fchmod(dst.fileno(), 0o755 if label == "binary" else 0o644)
+                    dst.flush()
+                    os.fsync(dst.fileno())
+                os.replace(temporary, destination)
+                fd = os.open(destination.parent, os.O_DIRECTORY)
+                try:
+                    os.fsync(fd)
+                finally:
+                    os.close(fd)
+            finally:
+                if temporary.is_file() and not temporary.is_symlink():
+                    temporary.unlink()
 
     def profile_values(self, request):
         values = {"DEMO_MODE": "true", "PULSE_MOCK_MODE": "true", "PULSE_MOCK_RANDOM_METRICS": "true",
@@ -308,7 +384,9 @@ class Transaction:
         self.receipt = {"schema_version": 2, "status": "intent", "mutated": False,
                         "control_sha": request["control_sha"], "run_id": request["run_id"],
                         "run_attempt": request["run_attempt"], "mode": request["mode"],
-                        "expected_version": request["version"], "installer_sha256": request["installer_sha256"],
+                        "expected_version": request["version"],
+                        "footprint": "server-binary-version-and-demo-data",
+                        "binary_sha256": request["binary_sha256"], "version_sha256": request["version_sha256"],
                         "forward": {}, "recovery": {}, "recovery_required": False}
 
     def save(self, status):
@@ -346,6 +424,11 @@ class Transaction:
             evidence["observed_version"] = self.host.version(self.request["local_url"])
             if evidence["observed_version"] != expected_version:
                 raise Failure("runtime-version")
+            runtime = estate_hash({name: PATHS[name] for name in ("binary", "version")})
+            expected = self.receipt.get("runtime_snapshot_sha256" if phase == "recovery" else "installed_runtime_sha256")
+            if expected and runtime != expected:
+                raise Failure("distribution-state-changed")
+            evidence["runtime_sha256"] = runtime
             evidence["samples"] += 1
             elapsed = self.host.now() - started
             evidence["elapsed_seconds"] = elapsed
@@ -358,8 +441,8 @@ class Transaction:
         size = sum(path.lstat().st_size for root in PATHS.values() if root.exists()
                    for path in ([root] + (list(root.rglob("*")) if root.is_dir() else []))
                    if path.is_file() and not path.is_symlink())
-        # Both our independent estate and the signed installer's own backup need
-        # space. Refuse shortage; never delete backups or databases for headroom.
+        # Keep room for the independent estate and the two atomic runtime swaps.
+        # Never run the installer's backup rotation or delete data for headroom.
         if any(shutil.disk_usage(parent).free < 2 * size + 64 * 1024 * 1024
                for parent in (self.attempt, PATHS["data"].parent)):
             raise Failure("snapshot-headroom")
@@ -372,6 +455,7 @@ class Transaction:
         atomic_json(snapshot / "paths.json", manifest)
         digest = estate_hash(PATHS)
         self.receipt["snapshot_sha256"] = digest
+        self.receipt["runtime_snapshot_sha256"] = estate_hash({name: PATHS[name] for name in ("binary", "version")})
         self.receipt["snapshot_retained"] = True
         return digest
 
@@ -433,8 +517,8 @@ class Transaction:
             cursors = self.host.cursors()
             if self.request["mode"] == "update":
                 self.host.install(self.attempt, self.request)
-                # The installer may start the unit; finish profile work quiescent.
-                self.host.stop()
+                self.receipt["installed_runtime_sha256"] = estate_hash({
+                    name: PATHS[name] for name in ("binary", "version")})
             self.host.profile(self.request, not healthy_baseline)
             self.host.start()
             self.ready(cursors)

.github/scripts/dispatch-demo-runtime.py

@@ -4,7 +4,10 @@ from __future__ import annotations
 
 import base64
 import hashlib
+import importlib.util
 import json
+import datetime
+import re
 import os
 from pathlib import Path
 import shlex
@@ -14,12 +17,14 @@ import sys
 BOOTSTRAP = r'''
 import base64, fcntl, hashlib, json, os, pathlib, subprocess, sys, time
 root = pathlib.Path('/var/lib/pulse-deploy/demo')
-payload = json.loads(sys.stdin.read(512 * 1024))
-assert os.geteuid() == 0 and set(payload) == {'request', 'source', 'installer'}
+payload = json.loads(sys.stdin.read(192 * 1024 * 1024 + 1))
+assert os.geteuid() == 0 and set(payload) == {'request', 'source', 'binary', 'version_file'}
 source = base64.b64decode(payload['source'], validate=True)
-installer = base64.b64decode(payload['installer'], validate=True)
-assert len(source) < 128 * 1024 and len(installer) < 256 * 1024
-assert (hashlib.sha256(installer).hexdigest() == payload['request']['installer_sha256']) if installer else (payload['request']['installer_sha256'] == '')
+binary = base64.b64decode(payload['binary'], validate=True)
+version_file = base64.b64decode(payload['version_file'], validate=True)
+assert len(source) < 128 * 1024 and len(binary) <= 128 * 1024 * 1024 and len(version_file) <= 128
+for name, data in [('binary', binary), ('version', version_file)]:
+    assert (hashlib.sha256(data).hexdigest() == payload['request'][name + '_sha256']) if data else (payload['request'][name + '_sha256'] == '')
 identity = hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(',', ':')).encode()).hexdigest()
 root.mkdir(mode=0o700, parents=True, exist_ok=True)
 assert not root.is_symlink() and root.stat().st_uid == 0
@@ -31,7 +36,7 @@ fcntl.flock(fd, fcntl.LOCK_EX)
 attempt = root / 'attempts' / identity
 if not attempt.exists():
     attempt.mkdir(mode=0o700)
-    for name, data in [('transaction.py', source), ('installer.sh', installer),
+    for name, data in [('transaction.py', source), ('binary', binary), ('version', version_file),
                        ('request.json', json.dumps(payload['request']).encode()),
                        ('intent.json', json.dumps({'schema_version': 1, 'identity': identity, 'state': 'intent'}).encode())]:
         with open(attempt / name, 'xb') as stream:
@@ -83,20 +88,103 @@ def payload(mode, environ=os.environ):
                ('nodes', 'vms_per_node', 'lxcs_per_node', 'docker_hosts', 'docker_containers',
                 'generic_hosts', 'k8s_clusters', 'k8s_nodes', 'k8s_pods', 'k8s_deployments',
                 'seed_duration', 'sample_interval', 'update_interval')}
-    installer = Path('/tmp/pulse-install.sh').read_bytes() if mode == 'update' else b''
+    binary = Path('/tmp/pulse-demo-binary').read_bytes() if mode == 'update' else b''
+    version_file = Path('/tmp/pulse-demo-VERSION').read_bytes() if mode == 'update' else b''
     request = {'mode': mode, 'hostname': environ['DEMO_EXPECTED_HOSTNAME'],
                'local_url': environ['DEMO_LOCAL_BASE_URL'], 'public_url': environ['DEMO_PUBLIC_HEALTH_URL'],
                'version': environ['EXPECTED_VERSION'], 'profile': profile,
                'control_sha': environ['GITHUB_SHA'], 'run_id': environ['GITHUB_RUN_ID'],
                'run_attempt': environ['GITHUB_RUN_ATTEMPT'],
-               'installer_sha256': hashlib.sha256(installer).hexdigest() if installer else ''}
+               'binary_sha256': hashlib.sha256(binary).hexdigest() if binary else '',
+               'version_sha256': hashlib.sha256(version_file).hexdigest() if version_file else ''}
     return {'request': request, 'source': base64.b64encode(Path(__file__).with_name('demo-runtime-transaction.py').read_bytes()).decode(),
-            'installer': base64.b64encode(installer).decode()}
+            'binary': base64.b64encode(binary).decode(), 'version_file': base64.b64encode(version_file).decode()}
+
+
+NATIVE_PATHS = ('.github/scripts/demo-runtime-transaction.py',
+                '.github/scripts/dispatch-demo-runtime.py',
+                '.github/scripts/tests/demo_runtime_native.py',
+                '.github/workflows/demo-runtime-native.yml')
+
+
+def require_native(environ=os.environ):
+    """Fail closed before SSH; a metadata/component pass is not native proof."""
+    sha, repository = environ['GITHUB_SHA'], environ['GITHUB_REPOSITORY']
+    if not re.fullmatch(r'[0-9a-f]{40}', sha) or repository != 'rcourtman/Pulse':
+        raise ValueError('native-source-identity')
+    since = (datetime.datetime.now(datetime.timezone.utc).date() - datetime.timedelta(days=6)).isoformat()
+    base = f'repos/{repository}/actions/workflows/demo-runtime-native.yml/runs?branch=main&event=push&per_page=5'
+
+    def read(query):
+        result = subprocess.run(['gh', 'api', query], capture_output=True, timeout=30, check=False)
+        if result.returncode:
+            raise ValueError('native-evidence-unavailable')
+        rows = json.loads(result.stdout)['workflow_runs']
+        if not isinstance(rows, list) or any(not isinstance(row, dict) for row in rows):
+            raise ValueError('native-evidence-malformed')
+        for row in rows:
+            if (type(row.get('id')) is not int or not re.fullmatch(r'[0-9a-f]{40}', str(row.get('head_sha')))
+                    or type(row.get('run_attempt')) is not int or not isinstance(row.get('updated_at'), str)
+                    or not isinstance(row.get('created_at'), str)):
+                raise ValueError('native-evidence-malformed')
+        return rows
+
+    rows = read(base + '&created=' + since + '..*')
+    if len(rows) < 5:
+        rows += read(base)
+    merged = {}
+    for row in rows:
+        old = merged.get(row['id'])
+        if old is None or (row['run_attempt'], row['updated_at']) > (old['run_attempt'], old['updated_at']):
+            merged[row['id']] = row
+    for row in sorted(merged.values(), key=lambda r: (r['created_at'], r['id']), reverse=True):
+        head = row['head_sha']
+        if (row.get('event') != 'push' or row.get('head_branch') != 'main'
+                or row.get('path') != '.github/workflows/demo-runtime-native.yml'):
+            raise ValueError('native-run-identity')
+        ancestor = subprocess.run(['git', 'merge-base', '--is-ancestor', head, sha], capture_output=True)
+        if ancestor.returncode != 0:
+            continue
+        matching = subprocess.run(['git', 'diff', '--quiet', head, sha, '--', *NATIVE_PATHS], capture_output=True)
+        if matching.returncode == 1:
+            continue
+        if matching.returncode != 0:
+            raise ValueError('native-source-unavailable')
+        # Never select an older green sample over a newer matching failure or
+        # pending run. The native driver exits success only with full cleanup.
+        if row.get('status') != 'completed' or row.get('conclusion') != 'success':
+            raise ValueError('native-acceptance-not-passed')
+        return {'run_id': row['id'], 'run_attempt': row['run_attempt'], 'source_sha': head}
+    raise ValueError('native-acceptance-missing')
+
+
+def prepare_runtime():
+    version = os.environ['EXPECTED_VERSION']
+    if not re.fullmatch(r'v[0-9]+\.[0-9]+\.[0-9]+', version):
+        raise ValueError('release-version')
+    spec = importlib.util.spec_from_file_location('demo_runtime', Path(__file__).with_name('demo-runtime-transaction.py'))
+    engine = importlib.util.module_from_spec(spec); spec.loader.exec_module(engine)
+    # Called only after mandatory pinned-key SSH signature verification in CI.
+    files = engine.runtime_members('/tmp/pulse-demo-release.tgz', version)
+    for label, name in (('binary', '/tmp/pulse-demo-binary'), ('version', '/tmp/pulse-demo-VERSION')):
+        with open(name, 'xb') as stream:
+            os.chmod(name, 0o600)
+            stream.write(files[label])
 
 
 def main():
-    if len(sys.argv) != 2 or sys.argv[1] not in {'update', 'recover'}:
+    if len(sys.argv) != 2 or sys.argv[1] not in {'prepare', 'update', 'recover'}:
         return 2
+    if sys.argv[1] == 'prepare':
+        prepare_runtime()
+        return 0
+    # Check exact reviewed native source before opening the production SSH path.
+    try:
+        require_native()
+    except (ValueError, KeyError, OSError, subprocess.TimeoutExpired):
+        print(json.dumps({'schema_version': 2, 'status': 'refused', 'mutated': False,
+                          'failure': 'native-acceptance-unavailable'}))
+        return 1
     # Same project CI SSH identity and pinned host verification as before. This
     # is not an installed worker dispatcher or a new project credential.
     command = ['ssh', '-i', str(Path.home() / '.ssh/id_ed25519'), '-o', 'IdentitiesOnly=yes',

.github/scripts/tests/demo_runtime_native.py

@@ -31,6 +31,9 @@ def binary(version, fail_after=0, port=17655):
     return ('''#!/usr/bin/python3
 import http.server,json,os,threading,time
 VERSION = %r
+if %r == 17655:
+    from pathlib import Path
+    Path('/etc/pulse/persistent-marker').write_text('candidate data ' + VERSION)
 def fail():
     time.sleep(%r)
     print('panic: synthetic disposable service fault',flush=True)
@@ -45,29 +48,13 @@ class Handler(http.server.BaseHTTPRequestHandler):
         self.send_response(200); self.end_headers(); self.wfile.write(body)
     def log_message(self,*args): pass
 http.server.HTTPServer(('127.0.0.1',%r),Handler).serve_forever()
-''' % (version, fail_after, fail_after, port, port)).encode()
+''' % (version, port, fail_after, fail_after, port, port)).encode()
 
 
 def unit(executable, user="pulse"):
     return f"[Unit]\nDescription=Disposable demo native fixture\n[Service]\nUser={user}\nGroup={user}\nExecStart={executable}\nRestart=no\n[Install]\nWantedBy=multi-user.target\n"
 
 
-def installer(version, fail_after=0):
-    source = binary(version, fail_after)
-    return ("#!/usr/bin/python3\nfrom pathlib import Path\n"
-            "import base64\n"
-            f"p=Path('/opt/pulse/bin/pulse');p.write_bytes(base64.b64decode({base64.b64encode(source).decode()!r}));p.chmod(0o755)\n"
-            "p=Path('/etc/systemd/system/pulse.service');p.write_text(p.read_text()+'# candidate fixture\\n')\n"
-            f"Path('/etc/pulse/persistent-marker').write_text({'candidate data ' + version!r})\n").encode()
-
-
-def fixture_installer(version, fail_after=0):
-    # The real engine invokes bash like the published installer. This fixture
-    # deliberately changes executable, unit and persistent data without download.
-    script = installer(version, fail_after)
-    return ("#!/bin/bash\nset -eu\npython3 - <<'PY'\n" + script.decode().split("\n", 1)[1] + "\nPY\n").encode()
-
-
 def wait(predicate, limit):
     deadline = time.monotonic() + limit
     while time.monotonic() < deadline:
@@ -110,6 +97,13 @@ def main():
         for path in Path("/etc/pulse").iterdir():
             os.chown(path, owner.pw_uid, owner.pw_gid)
         Path("/opt/pulse/bin/pulse").write_bytes(binary("1.0.0")); Path("/opt/pulse/bin/pulse").chmod(0o755)
+        Path("/opt/pulse/VERSION").write_text("1.0.0\n")
+        Path("/opt/pulse/scripts").mkdir()
+        untouched = {"agent": Path("/opt/pulse/bin/pulse-agent-fixture"),
+                     "script": Path("/opt/pulse/scripts/install-fixture.sh")}
+        for path in untouched.values():
+            path.write_bytes(b"unchanged bundled distribution asset")
+        untouched_before = {name: hashlib.sha256(path.read_bytes()).hexdigest() for name, path in untouched.items()}
         Path("/opt/pulse/bin/relay").write_bytes(binary("1.0.0", port=17656)); Path("/opt/pulse/bin/relay").chmod(0o755)
         for name, executable in (("pulse", "/opt/pulse/bin/pulse"), ("pulse-relay", "/opt/pulse/bin/relay")):
             Path(f"/etc/systemd/system/{name}.service").write_text(unit(executable))
@@ -144,13 +138,16 @@ def main():
         profile.update(seed_duration="2h", sample_interval="5m", update_interval="15s")
 
         def submit(version, fault=0):
-            install = fixture_installer(version, fault)
+            candidate = binary(version, fault)
+            metadata = (version + "\n").encode()
             request = {"mode": "update", "hostname": socket.gethostname(), "local_url": "http://127.0.0.1:17655",
                        "public_url": "https://127.0.0.1:18443/api/health", "version": "v" + version,
                        "profile": profile, "control_sha": os.environ["GITHUB_SHA"],
                        "run_id": os.environ["GITHUB_RUN_ID"], "run_attempt": os.environ["GITHUB_RUN_ATTEMPT"],
-                       "installer_sha256": hashlib.sha256(install).hexdigest()}
-            payload = {"request": request, "source": base64.b64encode(source).decode(), "installer": base64.b64encode(install).decode()}
+                       "binary_sha256": hashlib.sha256(candidate).hexdigest(),
+                       "version_sha256": hashlib.sha256(metadata).hexdigest()}
+            payload = {"request": request, "source": base64.b64encode(source).decode(),
+                       "binary": base64.b64encode(candidate).decode(), "version_file": base64.b64encode(metadata).decode()}
             identity = hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
             proc = subprocess.Popen(["python3", "-c", dispatcher.BOOTSTRAP], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
             processes.append(proc); proc.stdin.write(json.dumps(payload).encode()); proc.stdin.close()
@@ -165,6 +162,8 @@ def main():
         receipt = json.loads((attempt / "receipt.json").read_text()); result["cases"].append(receipt)
         assert proc.returncode == 0 and receipt["status"] == "committed" and receipt["forward"]["elapsed_seconds"] >= 300
         assert Path("/etc/pulse/persistent-marker").read_text() == "candidate data 1.0.1"
+        assert Path("/opt/pulse/VERSION").read_text() == "1.0.1\n"
+        assert {name: hashlib.sha256(path.read_bytes()).hexdigest() for name, path in untouched.items()} == untouched_before
         # A later candidate fails at 55s; lose the observer while the owned
         # systemd child continues, then TERM during actual full recovery.
         proc, attempt, payload = submit("1.0.2", 55)
@@ -180,6 +179,8 @@ def main():
         assert replay.returncode == 1 and receipt["status"] == "rolled_back" and receipt["recovery"]["elapsed_seconds"] >= 300
         assert receipt["failure"] in {"new-service-crash", "sustained-health"}
         assert Path("/etc/pulse/persistent-marker").read_text() == "candidate data 1.0.1"
+        assert Path("/opt/pulse/VERSION").read_text() == "1.0.1\n"
+        assert {name: hashlib.sha256(path.read_bytes()).hexdigest() for name, path in untouched.items()} == untouched_before
         version = json.loads(command(["curl", "--disable", "--silent", "--fail", "http://127.0.0.1:17655/api/version"]))["version"]
         assert version == "1.0.1"
         result["stage"] = "complete"

.github/scripts/tests/test_demo_runtime_transaction.py

@@ -1,5 +1,7 @@
 import hashlib
 import importlib.util
+import io
+import tarfile
 import json
 import os
 from pathlib import Path
@@ -29,7 +31,8 @@ def request(mode="recover"):
     return {"mode": mode, "hostname": "fixture", "local_url": "http://127.0.0.1:7655",
             "public_url": "https://demo.example/api/health", "version": "v6.4.5",
             "control_sha": "a" * 40, "run_id": "123", "run_attempt": "1",
-            "installer_sha256": hashlib.sha256(b"signed-installer-fixture").hexdigest() if mode == "update" else "",
+            "binary_sha256": hashlib.sha256(b"changed binary").hexdigest() if mode == "update" else "",
+            "version_sha256": hashlib.sha256(b"6.4.6\n").hexdigest() if mode == "update" else "",
             "profile": {**{name: "2" for name in engine.COUNT_KEYS},
                         "seed_duration": "2h", "sample_interval": "5m", "update_interval": "15s"}}
 
@@ -123,13 +126,10 @@ class FixtureHost(engine.Host):
             self.unhealthy = False
 
     def install(self, attempt, req):
-        if hashlib.sha256((attempt / "installer.sh").read_bytes()).hexdigest() != req["installer_sha256"]:
-            raise engine.Failure("installer-identity")
+        # Keep the clock/services virtual, but use the real narrow file swaps.
+        super().install(attempt, req)
         self.install_count += 1
         self.runtime_version = req["version"].lstrip("v")
-        engine.PATHS["binary"].write_bytes(b"new executable")
-        engine.PATHS["unit"].write_bytes(b"new unit")
-        (engine.PATHS["data"] / "persisted-history").write_bytes(b"forward history")
 
     def profile(self, req, unhealthy):
         super().profile(req, unhealthy)
@@ -159,7 +159,15 @@ class TransactionTest(unittest.TestCase):
         self.root = root / "control"
         self.attempt = self.root / "attempts" / ("a" * 64)
         self.attempt.mkdir(parents=True)
-        (self.attempt / "installer.sh").write_bytes(b"signed-installer-fixture")
+        self.paths["version"].write_text("6.4.5\n")
+        (self.attempt / "binary").write_bytes(b"changed binary")
+        (self.attempt / "version").write_bytes(b"6.4.6\n")
+        # Untouched distribution/service assets are outside the chosen operation.
+        self.untouched = {name: root / "unowned" / name for name in
+                          ("agent", "scripts", "helper", "auto-update", "timer", "symlink", "marker", "backup")}
+        for name, path in self.untouched.items():
+            path.parent.mkdir(parents=True, exist_ok=True); path.write_text("original " + name)
+        self.untouched_before = engine.estate_hash(self.untouched)
         self.addCleanup(patch.stopall)
         patch.object(engine, "ROOT", self.root).start()
         patch.object(engine, "PATHS", self.paths).start()
@@ -222,6 +230,34 @@ class TransactionTest(unittest.TestCase):
         self.assertEqual(engine.estate_hash(self.paths), self.before)
         self.assertEqual(self.host.runtime_version, "6.4.5")
 
+    def test_success_changes_only_runtime_and_profile_leaving_distribution_assets(self):
+        code, receipt = self.run_transaction("update", "v6.4.6")
+        self.assertEqual((code, receipt["status"]), (0, "committed"))
+        self.assertEqual(self.paths["binary"].read_bytes(), b"changed binary")
+        self.assertEqual(self.paths["version"].read_text(), "6.4.6\n")
+        for name in ("unit", "dropins"):
+            self.assertEqual(engine.estate_hash({name: self.paths[name]}),
+                             engine.estate_hash({name: self.attempt / "snapshot" / name}))
+        self.assertEqual(engine.estate_hash(self.untouched), self.untouched_before)
+        self.assertEqual(receipt["forward"]["elapsed_seconds"], 300)
+
+    def test_late_failure_restores_VERSION_and_runtime_with_untouched_helpers(self):
+        self.host.fail_at = 55
+        code, receipt = self.run_transaction("update", "v6.4.6")
+        self.assertEqual((code, receipt["status"]), (1, "rolled_back"))
+        self.assertEqual(self.paths["version"].read_text(), "6.4.5\n")
+        self.assertEqual(engine.estate_hash(self.paths), self.before)
+        self.assertEqual(engine.estate_hash(self.untouched), self.untouched_before)
+        self.assertEqual(receipt["recovery"]["runtime_sha256"], receipt["runtime_snapshot_sha256"])
+        self.assertEqual(receipt["recovery"]["elapsed_seconds"], 300)
+
+    def test_payload_version_mismatch_restores_without_installing(self):
+        (self.attempt / "version").write_text("6.4.7\n")
+        code, receipt = self.run_transaction("update", "v6.4.6")
+        self.assertEqual((code, receipt["status"]), (1, "rolled_back"))
+        self.assertEqual(receipt["failure"], "runtime-payload-identity")
+        self.assertEqual(engine.estate_hash(self.paths), self.before)
+
     def test_failure_at_299_seconds_still_collects_entire_recovery_and_cancellation(self):
         self.host.fail_at = 299
         self.host.cancel_in_recovery = True
@@ -359,7 +395,7 @@ class InputAndCommandTest(unittest.TestCase):
                  ("local_url", "http://other-host:7655"), ("public_url", "http://demo.example/api/health"),
                  ("public_url", "https://user:secret@demo.example/api/health"),
                  ("public_url", "https://demo.example/api/health?token=secret"),
-                 ("installer_sha256", "arbitrary")]
+                 ("binary_sha256", "arbitrary"), ("version_sha256", "arbitrary")]
         for name, value in cases:
             with self.subTest(name=name, value=value):
                 req = request(); req[name] = value
@@ -435,7 +471,7 @@ class DispatchLifecycleTest(unittest.TestCase):
                 source = source.replace("root.stat().st_uid == 0", "root.stat().st_uid == os.getuid()")
                 clock = [0]
                 calls = []
-                payload = {"request": request(), "source": "c291cmNl", "installer": ""}
+                payload = {"request": request(), "source": "c291cmNl", "binary": "", "version_file": ""}
 
                 def launch(argv, **kwargs):
                     calls.append(argv)
@@ -490,13 +526,103 @@ class NativeAdmissionTest(unittest.TestCase):
                     self.assertEqual(result["failure_code"], "not-an-empty-disposable-public-ci-runner")
             self.assertTrue(existing.is_dir())
 
-    def test_synthetic_changed_executable_and_data_are_distinct_between_steps(self):
+    def test_synthetic_changed_executables_have_actual_persistent_effects(self):
         compile(native.binary("1.0.1"), "fixture-runtime", "exec")
-        compile(native.installer("1.0.1"), "fixture-installer", "exec")
         self.assertNotEqual(native.binary("1.0.1"), native.binary("1.0.2", 55))
-        self.assertIn(b"candidate data 1.0.1", native.installer("1.0.1"))
-        self.assertIn(b"candidate data 1.0.2", native.installer("1.0.2", 55))
-        self.assertNotIn(b"curl", native.fixture_installer("1.0.1"))
+        self.assertIn(b"persistent-marker", native.binary("1.0.1"))
+        self.assertNotIn(b"curl", native.binary("1.0.1"))
+
+
+
+class RuntimeSelectionTest(unittest.TestCase):
+    def archive(self, directory, entries):
+        path = Path(directory) / 'release.tgz'
+        with tarfile.open(path, 'w:gz') as stream:
+            for name, content, kind in entries:
+                member = tarfile.TarInfo(name); member.size = len(content); member.type = kind
+                if kind == tarfile.SYMTYPE:
+                    member.linkname = '/outside'; member.size = 0
+                    stream.addfile(member)
+                else:
+                    stream.addfile(member, io.BytesIO(content))
+        return path
+
+    def elf(self):
+        body = bytearray(64); body[:6] = b'\x7fELF\x02\x01'; body[18:20] = b'\x3e\x00'
+        return bytes(body)
+
+    def test_selects_only_signed_archive_runtime_without_extracting_paths(self):
+        with tempfile.TemporaryDirectory() as directory:
+            entries = [('bin/pulse', self.elf(), tarfile.REGTYPE), ('./VERSION', b'6.4.6\n', tarfile.REGTYPE),
+                       ('../outside', b'untrusted path', tarfile.REGTYPE),
+                       ('scripts/install.sh', b'never executed', tarfile.REGTYPE),
+                       ('bin/pulse-agent', b'not installed', tarfile.REGTYPE)]
+            files = engine.runtime_members(self.archive(directory, entries), 'v6.4.6')
+            self.assertEqual(set(files), {'binary', 'version'})
+            self.assertEqual(files['binary'], self.elf())
+            self.assertEqual([p.name for p in Path(directory).iterdir()], ['release.tgz'])
+
+    def test_duplicate_link_wrong_version_or_architecture_is_rejected(self):
+        valid = [('bin/pulse', self.elf(), tarfile.REGTYPE), ('VERSION', b'6.4.6\n', tarfile.REGTYPE)]
+        cases = [valid + [('pulse', self.elf(), tarfile.REGTYPE)],
+                 [valid[0], ('VERSION', b'', tarfile.SYMTYPE)],
+                 [valid[0], ('VERSION', b'6.4.7\n', tarfile.REGTYPE)],
+                 [('bin/pulse', b'#!/bin/sh\n', tarfile.REGTYPE), valid[1]], [valid[0]]]
+        for entries in cases:
+            with self.subTest(entries=[x[0] for x in entries]), tempfile.TemporaryDirectory() as directory:
+                with self.assertRaises(engine.Failure):
+                    engine.runtime_members(self.archive(directory, entries), 'v6.4.6')
+
+
+class NativeGateTest(unittest.TestCase):
+    def row(self, run_id=10, **fields):
+        return {'id': run_id, 'head_sha': 'a' * 40, 'run_attempt': 1,
+                'created_at': '2026-10-03T01:00:00Z', 'updated_at': '2026-10-03T01:10:00Z',
+                'path': '.github/workflows/demo-runtime-native.yml', 'event': 'push', 'head_branch': 'main',
+                'status': 'completed', 'conclusion': 'success', **fields}
+
+    def check(self, rows, history=None, read_exit=0, same=True):
+        calls = []
+        def command(argv, **kwargs):
+            calls.append(argv)
+            if argv[0] == 'gh':
+                payload = rows if 'created=' in argv[-1] else (history if history is not None else rows)
+                return subprocess.CompletedProcess(argv, read_exit, json.dumps({'workflow_runs': payload}).encode())
+            return subprocess.CompletedProcess(argv, 0 if argv[1] == 'merge-base' or same else 1)
+        with patch.object(dispatcher.subprocess, 'run', side_effect=command):
+            result = dispatcher.require_native({'GITHUB_SHA': 'b' * 40, 'GITHUB_REPOSITORY': 'rcourtman/Pulse'})
+        return result, calls
+
+    def test_matching_reviewed_native_source_passes_with_current_first_inventory(self):
+        result, calls = self.check([self.row()])
+        self.assertEqual(result['run_id'], 10)
+        reads = [c for c in calls if c[0] == 'gh']
+        self.assertEqual(len(reads), 2)
+        self.assertIn('created=', reads[0][-1]); self.assertNotIn('created=', reads[1][-1])
+        self.assertEqual(set(calls[-1][6:]), set(dispatcher.NATIVE_PATHS))
+
+    def test_newer_matching_failure_or_pending_run_cannot_use_older_green(self):
+        for fields in ({'conclusion': 'failure'}, {'status': 'in_progress', 'conclusion': None}):
+            with self.subTest(fields=fields), self.assertRaisesRegex(ValueError, 'not-passed'):
+                self.check([self.row(11, **fields), self.row(10)])
+
+    def test_latest_duplicate_attempt_retains_failure(self):
+        with self.assertRaisesRegex(ValueError, 'not-passed'):
+            self.check([self.row()], [self.row(run_attempt=2, conclusion='failure')])
+
+    def test_missing_changed_or_misidentified_proof_cannot_enable_mutation(self):
+        for rows, same in (([], True), ([self.row()], False), ([self.row(event='workflow_dispatch')], True)):
+            with self.subTest(rows=rows, same=same), self.assertRaises(ValueError):
+                self.check(rows, same=same)
+
+    def test_refused_read_has_no_history_fallback_or_ssh(self):
+        calls = []
+        def denied(argv, **kwargs):
+            calls.append(argv); return subprocess.CompletedProcess(argv, 1, b'')
+        with patch.object(dispatcher.subprocess, 'run', side_effect=denied), self.assertRaises(ValueError):
+            dispatcher.require_native({'GITHUB_SHA': 'b' * 40, 'GITHUB_REPOSITORY': 'rcourtman/Pulse'})
+        self.assertEqual(len(calls), 1)
+        self.assertEqual(calls[0][0:2], ['gh', 'api'])
 
 
 if __name__ == "__main__":