Add disposable native acceptance for owned demo recovery

Exercise actual systemd and Caddy with a changed executable, distinct persistent data, delayed synthetic failure, observer loss and TERM during full rollback. Keep 300-second clocks, exact-source and cleanup receipts; no production credential or worker system privilege is added. Update the connected recovery contract and policy assertions, and retain bounded service identity observations.

Change-source: pulse-maintainer

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

@@ -224,7 +224,7 @@ class Host:
 
     def check_journal(self, cursors):
         for service, cursor in cursors.items():
-            text = self.command(["journalctl", "-u", service, "--after-cursor", cursor, "--output=json", "--no-pager"])
+            text = self.command(["journalctl", "-u", service, "--after-cursor", cursor, "--output=json", "--quiet", "--no-pager"])
             for line in text.splitlines():
                 try:
                     entry = json.loads(line)
@@ -335,13 +335,16 @@ class Transaction:
         baseline = self.host.state()
         started = self.host.now()
         evidence = self.receipt[phase]
-        evidence.update({"required_seconds": WINDOW, "samples": 0, "elapsed_seconds": 0})
+        evidence.update({"required_seconds": WINDOW, "samples": 0, "elapsed_seconds": 0,
+                         "initial_services": baseline})
         while True:
             self.host.check_journal(cursors)
             state = self.host.state()
+            evidence["last_services"] = state
             if state != baseline or not self.healthy(state):
                 raise Failure("sustained-health")
-            if self.host.version(self.request["local_url"]) != expected_version:
+            evidence["observed_version"] = self.host.version(self.request["local_url"])
+            if evidence["observed_version"] != expected_version:
                 raise Failure("runtime-version")
             evidence["samples"] += 1
             elapsed = self.host.now() - started
@@ -411,6 +414,8 @@ class Transaction:
             elif self.request["mode"] == "update":
                 raise Failure("update-baseline-unhealthy")
             self.receipt["healthy_baseline"] = healthy_baseline
+            self.receipt["baseline_services"] = states
+            self.receipt["baseline_version"] = original_version
             old_handlers = {sig: signal.getsignal(sig) for sig in (signal.SIGTERM, signal.SIGINT, signal.SIGHUP)}
             for sig in old_handlers:
                 signal.signal(sig, cancel)

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

@@ -38,7 +38,8 @@ if not attempt.exists():
             os.chmod(attempt / name, 0o600)
             stream.write(data); stream.flush(); os.fsync(stream.fileno())
     # Intent is durable before starting the unit. An uncertain launch is never replayed.
-    dfd = os.open(attempt, os.O_DIRECTORY); os.fsync(dfd); os.close(dfd)
+    for directory in (attempt, root / 'attempts', root, root.parent):
+        dfd = os.open(directory, os.O_DIRECTORY); os.fsync(dfd); os.close(dfd)
     try:
         child = subprocess.run(['systemd-run', '--quiet', '--collect',
                                 '--unit=pulse-demo-' + identity[:32],
@@ -53,6 +54,7 @@ if not attempt.exists():
     with open(attempt / 'launch.json', 'x') as stream:
         os.chmod(attempt / 'launch.json', 0o600)
         json.dump(launch, stream); stream.flush(); os.fsync(stream.fileno())
+    dfd = os.open(attempt, os.O_DIRECTORY); os.fsync(dfd); os.close(dfd)
 os.close(fd)
 # Closing or losing this observer never stops the independently owned unit.
 deadline = time.monotonic() + 4000

.github/scripts/tests/demo_runtime_native.py

@@ -0,0 +1,218 @@
+#!/usr/bin/env python3
+"""Secret-free, fresh public CI only. Never run this fixture on a service host."""
+import base64
+import hashlib
+import importlib.util
+import json
+import os
+from pathlib import Path
+import pwd
+import shutil
+import socket
+import subprocess
+import time
+
+SCRIPTS = Path(__file__).resolve().parents[1]
+RESULT = Path.cwd() / "demo-native-result.json"
+ROOT = Path("/var/lib/pulse-deploy/demo")
+OWNED = [Path("/etc/pulse"), Path("/opt/pulse"), Path("/var/lib/pulse-deploy"),
+         Path("/etc/systemd/system/pulse.service"), Path("/etc/systemd/system/pulse-relay.service"),
+         Path("/usr/local/share/ca-certificates/pulse-demo-native.crt")]
+
+
+def command(argv, timeout=120):
+    result = subprocess.run(argv, capture_output=True, timeout=timeout)
+    if result.returncode:
+        raise RuntimeError("fixture-command-failed")
+    return result.stdout
+
+
+def binary(version, fail_after=0, port=17655):
+    return ('''#!/usr/bin/python3
+import http.server,json,os,threading,time
+VERSION = %r
+def fail():
+    time.sleep(%r)
+    print('panic: synthetic disposable service fault',flush=True)
+    os._exit(1)
+if %r: threading.Thread(target=fail,daemon=True).start()
+class Handler(http.server.BaseHTTPRequestHandler):
+    def do_GET(self):
+        body=json.dumps({'version':VERSION} if self.path=='/api/version' else {'status':'healthy'}).encode()
+        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)).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:
+        if predicate():
+            return
+        time.sleep(1)
+    raise RuntimeError("fixture-observation-timeout")
+
+
+def main():
+    result = {"schema_version": 1, "control_sha": os.environ.get("GITHUB_SHA"),
+              "run_id": os.environ.get("GITHUB_RUN_ID"), "run_attempt": os.environ.get("GITHUB_RUN_ATTEMPT"),
+              "cases": [], "cleanup_complete": False, "signed_published_installer_acceptance": False}
+    processes = []
+    admitted = False
+    original_caddy = None
+    caddyfile = Path("/etc/caddy/Caddyfile")
+    try:
+        result["stage"] = "disposable-runner-admission"
+        # Environment fields are context, not permission: combine them with a
+        # fresh empty estate, effective root, disposable hosted VM and exact job.
+        if (os.geteuid() != 0 or os.environ.get("GITHUB_ACTIONS") != "true"
+                or os.environ.get("RUNNER_ENVIRONMENT") != "github-hosted"
+                or socket.gethostname() in {"pulse-relay", "pulse-license", "pulse-dev"}
+                or any(path.exists() or path.is_symlink() for path in OWNED)):
+            raise RuntimeError("not-an-empty-disposable-public-ci-runner")
+        admitted = True
+        result["stage"] = "prepare-native-fixtures"
+        original_caddy = caddyfile.read_bytes()
+        command(["systemctl", "show", "--property=Version"])
+        if not subprocess.run(["id", "pulse"], capture_output=True).returncode == 0:
+            command(["useradd", "--system", "--user-group", "--no-create-home", "pulse"])
+        owner = pwd.getpwnam("pulse")
+        Path("/opt/pulse/bin").mkdir(parents=True)
+        Path("/etc/pulse").mkdir(mode=0o755)
+        os.chown("/etc/pulse", owner.pw_uid, owner.pw_gid)
+        Path("/etc/pulse/.env").write_text("DEMO_MODE=true\n")
+        Path("/etc/pulse/billing.json").write_text('{"capabilities":[]}')
+        Path("/etc/pulse/persistent-marker").write_text("original data")
+        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/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))
+        command(["systemctl", "daemon-reload"])
+        command(["systemctl", "enable", "--now", "pulse", "pulse-relay"])
+        cert = Path("/etc/caddy/demo-native.crt"); key = Path("/etc/caddy/demo-native.key")
+        command(["openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-days", "1",
+                 "-subj", "/CN=localhost", "-addext", "subjectAltName=IP:127.0.0.1",
+                 "-addext", "basicConstraints=critical,CA:TRUE",
+                 "-keyout", str(key), "-out", str(cert)])
+        key.chmod(0o640); shutil.chown(key, group="caddy")
+        shutil.copyfile(cert, OWNED[-1]); command(["update-ca-certificates"])
+        caddyfile.write_text("{\n admin off\n auto_https off\n}\n" +
+                             f"https://127.0.0.1:18443 {{\n tls {cert} {key}\n reverse_proxy 127.0.0.1:17655\n}}\n" +
+                             f"https://127.0.0.1:18444 {{\n tls {cert} {key}\n reverse_proxy 127.0.0.1:17656\n}}\n")
+        command(["systemctl", "restart", "caddy"])
+        spec = importlib.util.spec_from_file_location("dispatcher", SCRIPTS / "dispatch-demo-runtime.py")
+        dispatcher = importlib.util.module_from_spec(spec); spec.loader.exec_module(dispatcher)
+        original = (SCRIPTS / "demo-runtime-transaction.py").read_bytes()
+        # Only the cohost endpoint is rebound. Transaction, paths, locking,
+        # systemd, HTTP/TLS, estate, clocks and watches remain real and unchanged.
+        endpoint = b'https://relay.pulserelay.pro/health'
+        assert original.count(endpoint) == 1
+        source = original.replace(endpoint, b'https://127.0.0.1:18444/health')
+        result["original_engine_sha256"] = hashlib.sha256(original).hexdigest()
+        result["fixture_engine_sha256"] = hashlib.sha256(source).hexdigest()
+        result["bootstrap_sha256"] = hashlib.sha256(dispatcher.BOOTSTRAP.encode()).hexdigest()
+        result["driver_sha256"] = hashlib.sha256(Path(__file__).read_bytes()).hexdigest()
+        result["tools"] = {"systemd": command(["systemctl", "--version"]).decode().splitlines()[0],
+                           "caddy": command(["caddy", "version"]).decode().strip()}
+        profile = {name: "2" for name in ("nodes", "vms_per_node", "lxcs_per_node", "docker_hosts", "docker_containers", "generic_hosts", "k8s_clusters", "k8s_nodes", "k8s_pods", "k8s_deployments")}
+        profile.update(seed_duration="2h", sample_interval="5m", update_interval="15s")
+
+        def submit(version, fault=0):
+            install = fixture_installer(version, fault)
+            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()}
+            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()
+            return proc, ROOT / "attempts" / identity, payload
+
+        # Readiness is real, including authenticated TLS through actual Caddy.
+        result["stage"] = "native-listener-readiness"
+        wait(lambda: subprocess.run(["curl", "--disable", "--silent", "--fail", "https://127.0.0.1:18443/api/health"], capture_output=True).returncode == 0, 60)
+        proc, attempt, payload = submit("1.0.1")
+        result["stage"] = "changed-executable-forward"
+        proc.wait(timeout=900)
+        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"
+        # 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)
+        result["stage"] = "delayed-fault-observer-loss"
+        wait(lambda: (attempt / "receipt.json").exists() and json.loads((attempt / "receipt.json").read_text())["status"] == "applying", 60)
+        proc.terminate(); proc.wait(timeout=30)
+        command(["systemctl", "is-active", "pulse-demo-" + attempt.name[:32]])
+        wait(lambda: json.loads((attempt / "receipt.json").read_text())["status"] == "recovering", 180)
+        result["stage"] = "term-during-owned-recovery"
+        command(["systemctl", "kill", "--kill-whom=main", "--signal=TERM", "pulse-demo-" + attempt.name[:32]])
+        replay = subprocess.run(["python3", "-c", dispatcher.BOOTSTRAP], input=json.dumps(payload).encode(), capture_output=True, timeout=900)
+        receipt = json.loads((attempt / "receipt.json").read_text()); result["cases"].append(receipt)
+        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"
+        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"
+        result["passed"] = True
+    except Exception as error:
+        result["passed"] = False
+        result["failure_type"] = type(error).__name__
+        result["failure_code"] = str(error) if isinstance(error, RuntimeError) else "fixture-observation-or-assertion"
+    finally:
+        if admitted:
+            try:
+                for proc in processes:
+                    if proc.poll() is None:
+                        proc.terminate(); proc.wait(timeout=30)
+                # Stop only retained children belonging to this fresh fixture.
+                if (ROOT / "attempts").exists():
+                    for attempt in (ROOT / "attempts").iterdir():
+                        subprocess.run(["systemctl", "stop", "pulse-demo-" + attempt.name[:32]], capture_output=True, timeout=1300)
+                command(["systemctl", "disable", "--now", "pulse", "pulse-relay"])
+                command(["systemctl", "stop", "caddy"])
+                if original_caddy is not None:
+                    caddyfile.write_bytes(original_caddy)
+                for path in OWNED:
+                    if path.is_dir(): shutil.rmtree(path)
+                    elif path.exists(): path.unlink()
+                for path in (Path("/etc/caddy/demo-native.crt"), Path("/etc/caddy/demo-native.key")):
+                    if path.exists(): path.unlink()
+                command(["systemctl", "daemon-reload"]); command(["update-ca-certificates", "--fresh"])
+                result["cleanup_complete"] = not any(path.exists() for path in OWNED)
+            except Exception as cleanup:
+                result["cleanup_failure_type"] = type(cleanup).__name__
+                result["passed"] = False
+        RESULT.write_text(json.dumps(result, indent=2) + "\n")
+        RESULT.chmod(0o644)
+    return 0 if result.get("passed") and result["cleanup_complete"] else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

.github/scripts/tests/test_demo_runtime_transaction.py

@@ -22,6 +22,7 @@ def load(name, filename):
 
 engine = load("demo_transaction", "demo-runtime-transaction.py")
 dispatcher = load("demo_dispatcher", "dispatch-demo-runtime.py")
+native = load("demo_native", "tests/demo_runtime_native.py")
 
 
 def request(mode="recover"):
@@ -181,6 +182,13 @@ class TransactionTest(unittest.TestCase):
         self.assertEqual(receipt["status"], "committed")
         self.assertEqual(receipt["forward"]["elapsed_seconds"], 300)
         self.assertEqual(receipt["forward"]["samples"], 61)
+        self.assertEqual(receipt["baseline_services"]["pulse"]["MainPID"], "100")
+        self.assertEqual(receipt["forward"]["initial_services"], receipt["forward"]["last_services"])
+        self.assertEqual(receipt["forward"]["observed_version"], "6.4.5")
+        self.assertEqual(self.host.install_count, 0)
+        for name in ("binary", "unit", "dropins"):
+            self.assertEqual(engine.estate_hash({name: self.paths[name]}),
+                             engine.estate_hash({name: self.attempt / "snapshot" / name}))
         self.assertEqual((self.paths["data"] / "alerts/events.db").read_bytes(), b"synthetic opaque persistent file")
         self.assertTrue((self.attempt / "snapshot/data/persisted-history").exists())
         self.assertNotIn("not-logged", json.dumps(receipt))
@@ -387,6 +395,13 @@ class InputAndCommandTest(unittest.TestCase):
             host.check_journal(cursors)
         self.assertEqual(cursors, {"pulse": "d"})
 
+    def test_available_empty_journal_window_has_no_non_json_header(self):
+        host = engine.Host(); cursors = {"pulse": "c"}
+        with patch.object(host, "command", return_value="") as command:
+            host.check_journal(cursors)
+        self.assertIn("--quiet", command.call_args.args[0])
+        self.assertEqual(cursors, {"pulse": "c"})
+
     def test_no_journal_cursor_and_malformed_service_state_fail_closed(self):
         host = engine.Host()
         with patch.object(host, "command", return_value=""):
@@ -446,5 +461,40 @@ class DispatchLifecycleTest(unittest.TestCase):
                     self.assertEqual(json.loads((attempt / "launch.json").read_text())["state"], "uncertain")
 
 
+class NativeAdmissionTest(unittest.TestCase):
+    def test_native_driver_refuses_wrong_context_before_any_service_command(self):
+        with tempfile.TemporaryDirectory() as directory:
+            result_path = Path(directory) / "result.json"
+            existing = Path(directory) / "existing-estate"
+            existing.mkdir()
+            cases = [(1000, "true", "github-hosted", []),
+                     (0, "false", "github-hosted", []),
+                     (0, "true", "self-hosted", []),
+                     (0, "true", "github-hosted", [existing])]
+            for uid, actions, environment, estate in cases:
+                with self.subTest(uid=uid, actions=actions, environment=environment, occupied=bool(estate)), \
+                     patch.object(native, "RESULT", result_path), patch.object(native, "OWNED", estate), \
+                     patch.object(os, "geteuid", return_value=uid), \
+                     patch.dict(os.environ, {"GITHUB_ACTIONS": actions, "RUNNER_ENVIRONMENT": environment}), \
+                     patch.object(native.socket, "gethostname", return_value="disposable-fixture"), \
+                     patch.object(native, "command", side_effect=AssertionError("must not execute a service command")) as command:
+                    self.assertEqual(native.main(), 1)
+                    command.assert_not_called()
+                    result = json.loads(result_path.read_text())
+                    self.assertFalse(result["passed"])
+                    self.assertFalse(result["cleanup_complete"])
+                    self.assertFalse(result["signed_published_installer_acceptance"])
+                    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):
+        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"))
+
+
 if __name__ == "__main__":
     unittest.main()

.github/workflows/demo-runtime-native.yml

@@ -0,0 +1,43 @@
+name: Demo transaction native acceptance
+on:
+  pull_request:
+    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'
+  push:
+    branches: [main]
+    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'
+permissions:
+  contents: read
+concurrency:
+  group: demo-native-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+jobs:
+  native:
+    runs-on: ubuntu-24.04
+    timeout-minutes: 35
+    steps:
+      - name: Check out exact source without retained credentials
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+      - name: Prepare Caddy only in the disposable public CI runner
+        run: |
+          sudo apt-get update -qq
+          sudo apt-get install -y --no-install-recommends caddy openssl
+      - name: Exercise real systemd Caddy and retained child recovery
+        run: sudo --preserve-env=GITHUB_ACTIONS,RUNNER_ENVIRONMENT,GITHUB_SHA,GITHUB_RUN_ID,GITHUB_RUN_ATTEMPT python3 .github/scripts/tests/demo_runtime_native.py
+      - name: Retain sanitised terminal receipts and cleanup
+        if: always()
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: demo-native-${{ github.run_id }}
+          path: demo-native-result.json
+          if-no-files-found: error
+          retention-days: 14