Add v5 agent self-update runtime proof

Extend the unified-agent RC rehearsal with an optional local v5-to-v6 process-swap check.

Record the v5.1.34 to v6.0.0-rc.6 proof result for the upgrade-continuity gate.

docs/release-control/v6/internal/records/unified-agent-v5-upgrade-continuity-2026-03-12.md

@@ -14,6 +14,17 @@
 - `python3 scripts/release_control/unified_agent_rc_rehearsal.py --base-url http://127.0.0.1:7655 --expected-version 6.0.0-rc.1 --release-base-url file:///tmp/pulse-private-release-assets --arch linux-amd64 --api-token <redacted> --expected-active-agents 3 --expected-agent-name delly --expected-online-agents 3`
 - Result: pass
 
+## Automated Local Runtime Proof
+
+- Date: `2026-06-03`
+- Host architecture: `darwin-arm64`
+- Starting binary: `v5.1.34` built from `repos/pulse-5.1.x` tag `v5.1.34`
+- Target binary: `v6.0.0-rc.6` built from `pulse/v6-release`
+- Command: `PYTHONPATH=scripts/release_control/internal python3 scripts/release_control/internal/unified_agent_rc_rehearsal.py --base-url http://127.0.0.1:1 --expected-version 6.0.0-rc.6 --release-base-url http://127.0.0.1:1/releases --skip-asset-checks --runtime-v5-agent /tmp/pulse-v5-v6-runtime-proof/bin/pulse-agent-v5.1.34-darwin-arm64 --runtime-v6-agent /tmp/pulse-v5-v6-runtime-proof/bin/pulse-agent-v6.0.0-rc.6-darwin-arm64 --runtime-expected-from v5.1.34 --runtime-expected-to v6.0.0-rc.6 --runtime-timeout 90 --json`
+- Result: pass
+- Observed result: the v5 process reported `v5.1.34`, downloaded v6 binary checksum `407d803270388e36ddc6c851c0dd04bdb2ff9e4d7c08cda9d4cf5839df790227`, exec'd into `v6.0.0-rc.6`, and reported `updated_from=v5.1.34`.
+- Scope note: this local proof covers the in-place process swap, checksum-gated download, v5 host-report compatibility, v6 agent-report compatibility, and first-report `updated_from`. Live active-agent accounting still needs the authenticated Pulse API rehearsal above when proving a full server environment.
+
 ## Manual Crossover Exercise
 
 1. Built a real `linux-amd64` v5 agent from `main` with version `5.1.23`.

scripts/release_control/internal/unified_agent_rc_rehearsal.py

@@ -4,9 +4,17 @@
 from __future__ import annotations
 
 import argparse
+import gzip
 import hashlib
 import json
+import os
+import shutil
+import subprocess
+import tempfile
+import threading
+import time
 from dataclasses import asdict, dataclass
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 from typing import Iterable
 from urllib import error, parse, request
@@ -22,6 +30,104 @@ class CheckResult:
     detail: str
 
 
+class _RuntimeProofHandler(BaseHTTPRequestHandler):
+    server: "_RuntimeProofServer"
+
+    def do_GET(self) -> None:  # noqa: N802
+        self.server.record_request("GET", self.path)
+        parsed = parse.urlparse(self.path)
+        path = parsed.path
+
+        if path == "/api/agent/version":
+            self._write_json({"version": self.server.expected_version})
+            return
+
+        if path == "/download/pulse-agent":
+            self.send_response(200)
+            self.send_header("Content-Type", "application/octet-stream")
+            self.send_header("X-Checksum-Sha256", self.server.v6_checksum)
+            self.send_header("Content-Length", str(len(self.server.v6_binary)))
+            self.end_headers()
+            self.wfile.write(self.server.v6_binary)
+            return
+
+        if path in {"/api/agents/host/lookup", "/api/agents/agent/lookup"}:
+            if path.endswith("/host/lookup"):
+                self._write_json({"success": True, "host": {"id": self.server.agent_id}})
+            else:
+                self._write_json({"success": True, "agent": {"id": self.server.agent_id}})
+            return
+
+        if path.startswith("/api/agents/host/") and path.endswith("/config"):
+            self._write_json({"success": True, "hostId": self.server.agent_id, "config": {}})
+            return
+
+        if path.startswith("/api/agents/agent/") and path.endswith("/config"):
+            self._write_json({"success": True, "agentId": self.server.agent_id, "config": {}})
+            return
+
+        self.send_response(404)
+        self.end_headers()
+        self.wfile.write(b"missing")
+
+    def do_POST(self) -> None:  # noqa: N802
+        self.server.record_request("POST", self.path)
+        body = self.rfile.read(int(self.headers.get("Content-Length", "0") or "0"))
+        if self.headers.get("Content-Encoding", "").lower() == "gzip":
+            body = gzip.decompress(body)
+
+        if self.path in {"/api/agents/host/report", "/api/agents/agent/report"}:
+            try:
+                payload = json.loads(body.decode("utf-8"))
+            except json.JSONDecodeError:
+                self.send_response(400)
+                self.end_headers()
+                self.wfile.write(b"bad report json")
+                return
+            self.server.record_report(self.path, payload)
+            if self.path.endswith("/agent/report"):
+                self._write_json({"success": True, "agentId": self.server.agent_id, "config": {}})
+            else:
+                self._write_json({"success": True, "hostId": self.server.agent_id, "config": {}})
+            return
+
+        self.send_response(404)
+        self.end_headers()
+        self.wfile.write(b"missing")
+
+    def log_message(self, format: str, *args: object) -> None:  # noqa: A003
+        return
+
+    def _write_json(self, payload: dict[str, object]) -> None:
+        body = json.dumps(payload).encode("utf-8")
+        self.send_response(200)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+
+class _RuntimeProofServer(ThreadingHTTPServer):
+    def __init__(self, *, expected_version: str, v6_binary: bytes, agent_id: str) -> None:
+        super().__init__(("127.0.0.1", 0), _RuntimeProofHandler)
+        self.expected_version = expected_version
+        self.v6_binary = v6_binary
+        self.v6_checksum = sha256_hex(v6_binary)
+        self.agent_id = agent_id
+        self.requests: list[tuple[str, str]] = []
+        self.reports: list[tuple[str, dict[str, object]]] = []
+
+    @property
+    def base_url(self) -> str:
+        return f"http://127.0.0.1:{self.server_port}"
+
+    def record_request(self, method: str, path: str) -> None:
+        self.requests.append((method, path))
+
+    def record_report(self, path: str, payload: dict[str, object]) -> None:
+        self.reports.append((path, payload))
+
+
 def safe_check(name: str, fn) -> CheckResult:
     try:
         return fn()
@@ -118,6 +224,46 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
         default="Unified Agent RC Rehearsal",
         help="Markdown title used when writing --report-out",
     )
+    parser.add_argument(
+        "--skip-asset-checks",
+        action="store_true",
+        help=(
+            "Skip live Pulse version/install/download asset checks; intended for isolated "
+            "local runtime self-update proof runs"
+        ),
+    )
+    parser.add_argument(
+        "--runtime-v5-agent",
+        help="Optional path to a real v5 pulse-agent binary to exercise in-place self-update",
+    )
+    parser.add_argument(
+        "--runtime-v6-agent",
+        help="Optional path to the v6 pulse-agent binary served by the local runtime proof server",
+    )
+    parser.add_argument(
+        "--runtime-expected-from",
+        default="v5.1.34",
+        help="Expected version reported by --runtime-v5-agent before the swap",
+    )
+    parser.add_argument(
+        "--runtime-expected-to",
+        help="Expected version after the swap; defaults to --expected-version",
+    )
+    parser.add_argument(
+        "--runtime-timeout",
+        type=float,
+        default=60.0,
+        help="Seconds to wait for the real v5 process to self-update and report as v6",
+    )
+    parser.add_argument(
+        "--runtime-work-dir",
+        help="Optional existing directory for runtime proof artifacts; defaults to a temporary directory",
+    )
+    parser.add_argument(
+        "--runtime-keep-work-dir",
+        action="store_true",
+        help="Keep runtime proof artifacts instead of deleting the temporary directory",
+    )
     return parser.parse_args(argv)
 
 
@@ -182,6 +328,35 @@ def sha256_hex(data: bytes) -> str:
     return hashlib.sha256(data).hexdigest()
 
 
+def normalize_version(value: str) -> str:
+    value = value.strip()
+    if value.startswith("v"):
+        value = value[1:]
+    return value
+
+
+def report_agent_info(report: dict[str, object]) -> dict[str, object]:
+    agent = report.get("agent")
+    if isinstance(agent, dict):
+        return agent
+    return {}
+
+
+def agent_report_version(report: dict[str, object]) -> str:
+    agent = report_agent_info(report)
+    return str(agent.get("version", "")).strip()
+
+
+def agent_report_updated_from(report: dict[str, object]) -> str:
+    agent = report_agent_info(report)
+    return str(agent.get("updatedFrom") or agent.get("updated_from") or "").strip()
+
+
+def agent_report_type(report: dict[str, object]) -> str:
+    agent = report_agent_info(report)
+    return str(agent.get("type", "")).strip()
+
+
 def release_asset_url(release_base_url: str, version: str, asset_name: str) -> str:
     return f"{normalize_base_url(release_base_url)}/v{version}/{asset_name}"
 
@@ -275,6 +450,180 @@ def check_update_info(update_info_dir: str, expected_updated_from: str) -> Check
     )
 
 
+def check_local_runtime_self_update(
+    *,
+    v5_agent: str,
+    v6_agent: str,
+    expected_from: str,
+    expected_to: str,
+    timeout: float,
+    work_dir: str | None,
+    keep_work_dir: bool,
+) -> CheckResult:
+    v5_path = Path(v5_agent)
+    v6_path = Path(v6_agent)
+    if not v5_path.is_file():
+        return CheckResult("local-runtime-self-update", False, f"v5 agent not found: {v5_path}")
+    if not v6_path.is_file():
+        return CheckResult("local-runtime-self-update", False, f"v6 agent not found: {v6_path}")
+
+    expected_from_output = subprocess.check_output([str(v5_path), "--version"], text=True).strip()
+    if normalize_version(expected_from_output) != normalize_version(expected_from):
+        return CheckResult(
+            "local-runtime-self-update",
+            False,
+            f"v5 binary reports {expected_from_output!r}, expected {expected_from!r}",
+        )
+
+    expected_to_output = subprocess.check_output([str(v6_path), "--version"], text=True).strip()
+    if normalize_version(expected_to_output) != normalize_version(expected_to):
+        return CheckResult(
+            "local-runtime-self-update",
+            False,
+            f"v6 binary reports {expected_to_output!r}, expected {expected_to!r}",
+        )
+
+    v6_binary = v6_path.read_bytes()
+    agent_id = "agent-v5-to-v6-runtime-proof"
+    server = _RuntimeProofServer(
+        expected_version=normalize_version(expected_to),
+        v6_binary=v6_binary,
+        agent_id=agent_id,
+    )
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+
+    temp_root_created = False
+    if work_dir:
+        root = Path(work_dir)
+        root.mkdir(parents=True, exist_ok=True)
+    else:
+        root = Path(tempfile.mkdtemp(prefix="pulse-agent-v5-v6-proof-"))
+        temp_root_created = True
+
+    proc: subprocess.Popen[str] | None = None
+    try:
+        runtime_agent = root / "pulse-agent"
+        shutil.copy2(v5_path, runtime_agent)
+        runtime_agent.chmod(0o755)
+        token_file = root / "token"
+        token_file.write_text("runtime-proof-token\n", encoding="utf-8")
+        token_file.chmod(0o600)
+        agent_id_file = root / "agent-id"
+        state_dir = root / "state"
+        state_dir.mkdir(exist_ok=True)
+        log_path = root / "agent.log"
+
+        env = os.environ.copy()
+        env.update(
+            {
+                "PULSE_AGENT_CONFIG_SIGNATURE_REQUIRED": "false",
+                "PULSE_STATE_DIR": str(state_dir),
+            }
+        )
+        with log_path.open("w", encoding="utf-8") as log_file:
+            proc = subprocess.Popen(
+                [
+                    str(runtime_agent),
+                    "--url",
+                    server.base_url,
+                    "--token-file",
+                    str(token_file),
+                    "--agent-id",
+                    agent_id,
+                    "--agent-id-file",
+                    str(agent_id_file),
+                    "--hostname",
+                    "pulse-v5-v6-runtime-proof",
+                    "--interval",
+                    "2s",
+                    "--health-addr",
+                    "127.0.0.1:0",
+                    "--insecure",
+                    "--log-level",
+                    "debug",
+                ],
+                cwd=str(root),
+                env=env,
+                stdout=log_file,
+                stderr=subprocess.STDOUT,
+                text=True,
+            )
+
+            deadline = time.monotonic() + timeout
+            saw_v5 = False
+            saw_v6 = False
+            saw_updated_from = False
+            while time.monotonic() < deadline:
+                if proc.poll() is not None:
+                    break
+                for _path, report in server.reports:
+                    version = normalize_version(agent_report_version(report))
+                    if version == normalize_version(expected_from):
+                        saw_v5 = True
+                    if version == normalize_version(expected_to):
+                        saw_v6 = True
+                        if normalize_version(agent_report_updated_from(report)) == normalize_version(
+                            expected_from
+                        ):
+                            saw_updated_from = True
+                        if agent_report_type(report) != "unified":
+                            return CheckResult(
+                                "local-runtime-self-update",
+                                False,
+                                f"v6 report type={agent_report_type(report)!r}, expected 'unified'",
+                            )
+                if saw_v5 and saw_v6 and saw_updated_from:
+                    replaced_version = subprocess.check_output(
+                        [str(runtime_agent), "--version"], text=True
+                    ).strip()
+                    if normalize_version(replaced_version) != normalize_version(expected_to):
+                        return CheckResult(
+                            "local-runtime-self-update",
+                            False,
+                            f"replaced binary reports {replaced_version!r}, expected {expected_to!r}",
+                        )
+                    return CheckResult(
+                        "local-runtime-self-update",
+                        True,
+                        (
+                            f"v5 process reported {expected_from_output}, downloaded checksum "
+                            f"{server.v6_checksum}, exec'd v6 report {expected_to_output} with "
+                            f"updated_from={expected_from_output}; work_dir={root}"
+                        ),
+                    )
+                time.sleep(0.25)
+
+            log_excerpt = ""
+            if log_path.exists():
+                log_lines = log_path.read_text(encoding="utf-8", errors="replace").splitlines()
+                log_excerpt = "; log_tail=" + " | ".join(log_lines[-8:])
+            versions = [agent_report_version(report) for _path, report in server.reports]
+            return CheckResult(
+                "local-runtime-self-update",
+                False,
+                (
+                    f"timed out waiting for v5->v6 report sequence "
+                    f"(saw_v5={saw_v5}, saw_v6={saw_v6}, "
+                    f"saw_updated_from={saw_updated_from}, versions={versions!r}, "
+                    f"work_dir={root}){log_excerpt}"
+                ),
+            )
+    finally:
+        if proc is not None and proc.poll() is None:
+            proc.terminate()
+            try:
+                proc.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                proc.kill()
+                proc.wait(timeout=5)
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=5)
+        if temp_root_created and not keep_work_dir:
+            shutil.rmtree(root, ignore_errors=True)
+
+
 def check_active_agent_accounting(
     *,
     base_url: str,
@@ -433,44 +782,64 @@ def run_rehearsal(args: argparse.Namespace) -> list[CheckResult]:
     base_url = normalize_base_url(args.base_url)
     release_base_url = normalize_base_url(args.release_base_url)
     auth_headers = build_auth_headers(args)
-    results = [
-        safe_check(
-            "agent-version-endpoint",
-            lambda: check_version(base_url, args.expected_version, args.timeout),
-        ),
-        safe_check(
-            "install-sh-asset",
-            lambda: compare_asset(
-                name="install-sh-asset",
-                live_url=f"{base_url}/install.sh",
-                release_url=release_asset_url(release_base_url, args.expected_version, "install.sh"),
-                timeout=args.timeout,
-            ),
-        ),
-        safe_check(
-            "install-ps1-asset",
-            lambda: compare_asset(
-                name="install-ps1-asset",
-                live_url=f"{base_url}/install.ps1",
-                release_url=release_asset_url(release_base_url, args.expected_version, "install.ps1"),
-                timeout=args.timeout,
-            ),
-        ),
-    ]
+    results: list[CheckResult] = []
 
-    for arch in args.arch:
-        quoted_arch = parse.quote(arch, safe="")
+    if not args.skip_asset_checks:
+        results.extend(
+            [
+                safe_check(
+                    "agent-version-endpoint",
+                    lambda: check_version(base_url, args.expected_version, args.timeout),
+                ),
+                safe_check(
+                    "install-sh-asset",
+                    lambda: compare_asset(
+                        name="install-sh-asset",
+                        live_url=f"{base_url}/install.sh",
+                        release_url=release_asset_url(
+                            release_base_url, args.expected_version, "install.sh"
+                        ),
+                        timeout=args.timeout,
+                    ),
+                ),
+                safe_check(
+                    "install-ps1-asset",
+                    lambda: compare_asset(
+                        name="install-ps1-asset",
+                        live_url=f"{base_url}/install.ps1",
+                        release_url=release_asset_url(
+                            release_base_url, args.expected_version, "install.ps1"
+                        ),
+                        timeout=args.timeout,
+                    ),
+                ),
+            ]
+        )
+
+        for arch in args.arch:
+            quoted_arch = parse.quote(arch, safe="")
+            results.append(
+                safe_check(
+                    f"agent-binary-{arch}",
+                    lambda arch=arch, quoted_arch=quoted_arch: compare_asset(
+                        name=f"agent-binary-{arch}",
+                        live_url=f"{base_url}/download/pulse-agent?arch={quoted_arch}",
+                        release_url=release_asset_url(
+                            release_base_url, args.expected_version, agent_binary_asset_name(arch)
+                        ),
+                        timeout=args.timeout,
+                        expect_checksum_header=True,
+                    ),
+                )
+            )
+    elif args.arch:
         results.append(
             safe_check(
-                f"agent-binary-{arch}",
-                lambda arch=arch, quoted_arch=quoted_arch: compare_asset(
-                    name=f"agent-binary-{arch}",
-                    live_url=f"{base_url}/download/pulse-agent?arch={quoted_arch}",
-                    release_url=release_asset_url(
-                        release_base_url, args.expected_version, agent_binary_asset_name(arch)
-                    ),
-                    timeout=args.timeout,
-                    expect_checksum_header=True,
+                "asset-check-selection",
+                lambda: CheckResult(
+                    name="asset-check-selection",
+                    ok=False,
+                    detail="--arch cannot be combined with --skip-asset-checks",
                 ),
             )
         )
@@ -519,19 +888,60 @@ def run_rehearsal(args: argparse.Namespace) -> list[CheckResult]:
             )
         )
 
+    if args.runtime_v5_agent or args.runtime_v6_agent:
+        if not args.runtime_v5_agent or not args.runtime_v6_agent:
+            results.append(
+                CheckResult(
+                    name="local-runtime-self-update",
+                    ok=False,
+                    detail="both --runtime-v5-agent and --runtime-v6-agent are required together",
+                )
+            )
+        else:
+            results.append(
+                safe_check(
+                    "local-runtime-self-update",
+                    lambda: check_local_runtime_self_update(
+                        v5_agent=args.runtime_v5_agent,
+                        v6_agent=args.runtime_v6_agent,
+                        expected_from=args.runtime_expected_from,
+                        expected_to=args.runtime_expected_to or args.expected_version,
+                        timeout=args.runtime_timeout,
+                        work_dir=args.runtime_work_dir,
+                        keep_work_dir=args.runtime_keep_work_dir,
+                    ),
+                )
+            )
+
+    if not results:
+        results.append(
+            CheckResult(
+                name="check-selection",
+                ok=False,
+                detail="no rehearsal checks selected",
+            )
+        )
+
     return results
 
 
 def render_text(results: Iterable[CheckResult]) -> str:
+    results = list(results)
     lines: list[str] = []
     for result in results:
         prefix = "PASS" if result.ok else "FAIL"
         lines.append(f"[{prefix}] {result.name}: {result.detail}")
-    lines.append(
-        "Manual follow-up still required: confirm the upgraded v5-installed agent "
-        "reconnects as one canonical v6 identity, surfaces `updated_from` exactly once, "
-        "and leaves user-visible active-agent counts aligned with runtime enforcement."
-    )
+    if any(result.name == "local-runtime-self-update" and result.ok for result in results):
+        lines.append(
+            "Runtime follow-up covered: local v5 pulse-agent performed an in-place "
+            "self-update, exec'd the v6 binary, and reported `updated_from` once."
+        )
+    else:
+        lines.append(
+            "Manual follow-up still required: confirm the upgraded v5-installed agent "
+            "reconnects as one canonical v6 identity, surfaces `updated_from` exactly once, "
+            "and leaves user-visible active-agent counts aligned with runtime enforcement."
+        )
     return "\n".join(lines)
 
 
@@ -543,6 +953,7 @@ def render_markdown_report(
     release_base_url: str,
     results: Iterable[CheckResult],
 ) -> str:
+    results = list(results)
     lines = [
         f"# {title}",
         "",
@@ -556,16 +967,29 @@ def render_markdown_report(
     for result in results:
         status = "PASS" if result.ok else "FAIL"
         lines.append(f"- `{status}` `{result.name}`: {result.detail}")
-    lines.extend(
-        [
-            "",
-            "## Manual Follow-up",
-            "",
-            "- Confirm the upgraded v5-installed agent reconnects as one canonical v6 identity.",
-            "- Confirm `updated_from` appears exactly once on the first canonical v6 report and clears on the next report.",
-            "- Confirm settings/billing active-agent counts still match runtime enforcement after the upgrade.",
-        ]
-    )
+    runtime_covered = any(result.name == "local-runtime-self-update" and result.ok for result in results)
+    if runtime_covered:
+        lines.extend(
+            [
+                "",
+                "## Runtime Proof",
+                "",
+                "- Local v5 `pulse-agent` performed the in-place self-update and exec'd the v6 binary.",
+                "- The first v6 report carried `updated_from` for the v5 source version.",
+                "- Active-agent accounting still requires a live Pulse API check when this proof is run outside a full server.",
+            ]
+        )
+    else:
+        lines.extend(
+            [
+                "",
+                "## Manual Follow-up",
+                "",
+                "- Confirm the upgraded v5-installed agent reconnects as one canonical v6 identity.",
+                "- Confirm `updated_from` appears exactly once on the first canonical v6 report and clears on the next report.",
+                "- Confirm settings/billing active-agent counts still match runtime enforcement after the upgrade.",
+            ]
+        )
     return "\n".join(lines)
 
 

scripts/release_control/internal/unified_agent_rc_rehearsal_test.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import contextlib
 import io
 import json
+import shutil
 import tempfile
 import threading
 import unittest
@@ -12,9 +13,11 @@ from pathlib import Path
 
 from unified_agent_rc_rehearsal import (
     agent_binary_asset_name,
+    check_local_runtime_self_update,
     check_update_info,
     main,
     render_markdown_report,
+    render_text,
     release_asset_url,
     run_rehearsal,
     summarize_http_error_body,
@@ -61,6 +64,101 @@ class UnifiedAgentRCRehearsalTest(unittest.TestCase):
         _FixtureHandler.routes = routes
         _FixtureHandler.request_headers = {}
 
+    def write_fake_runtime_agents(self, root: Path) -> tuple[Path, Path]:
+        v5 = root / "fake-v5-agent"
+        v6 = root / "fake-v6-agent"
+        v5.write_text(
+            """#!/usr/bin/env python3
+import json
+import os
+import sys
+import urllib.request
+
+def arg(name):
+    if name in sys.argv:
+        index = sys.argv.index(name)
+        if index + 1 < len(sys.argv):
+            return sys.argv[index + 1]
+    return ""
+
+if "--version" in sys.argv:
+    print("v5.1.34")
+    sys.exit(0)
+
+base_url = arg("--url").rstrip("/")
+agent_id = arg("--agent-id") or "agent-v5"
+
+def post(path, payload):
+    request = urllib.request.Request(
+        base_url + path,
+        data=json.dumps(payload).encode("utf-8"),
+        headers={"Content-Type": "application/json"},
+        method="POST",
+    )
+    urllib.request.urlopen(request, timeout=5).read()
+
+post("/api/agents/host/report", {"agent": {"id": agent_id, "type": "unified", "version": "v5.1.34"}})
+urllib.request.urlopen(base_url + "/api/agent/version", timeout=5).read()
+download = urllib.request.urlopen(base_url + "/download/pulse-agent?arch=darwin-arm64", timeout=5).read()
+with open(sys.argv[0], "wb") as handle:
+    handle.write(download)
+os.chmod(sys.argv[0], 0o755)
+with open(os.path.join(os.path.dirname(sys.argv[0]), ".pulse-update-info"), "w", encoding="utf-8") as handle:
+    handle.write("v5.1.34")
+os.execv(sys.argv[0], sys.argv)
+""",
+            encoding="utf-8",
+        )
+        v6.write_text(
+            """#!/usr/bin/env python3
+import json
+import os
+import sys
+import time
+import urllib.request
+
+def arg(name):
+    if name in sys.argv:
+        index = sys.argv.index(name)
+        if index + 1 < len(sys.argv):
+            return sys.argv[index + 1]
+    return ""
+
+if "--version" in sys.argv:
+    print("v6.0.0-rc.6")
+    sys.exit(0)
+
+base_url = arg("--url").rstrip("/")
+agent_id = arg("--agent-id") or "agent-v6"
+info_path = os.path.join(os.path.dirname(sys.argv[0]), ".pulse-update-info")
+updated_from = ""
+if os.path.exists(info_path):
+    with open(info_path, encoding="utf-8") as handle:
+        updated_from = handle.read().strip()
+    os.remove(info_path)
+payload = {
+    "agent": {
+        "id": agent_id,
+        "type": "unified",
+        "version": "v6.0.0-rc.6",
+        "updatedFrom": updated_from,
+    }
+}
+request = urllib.request.Request(
+    base_url + "/api/agents/agent/report",
+    data=json.dumps(payload).encode("utf-8"),
+    headers={"Content-Type": "application/json"},
+    method="POST",
+)
+urllib.request.urlopen(request, timeout=5).read()
+time.sleep(60)
+""",
+            encoding="utf-8",
+        )
+        v5.chmod(0o755)
+        v6.chmod(0o755)
+        return v5, v6
+
     def test_release_asset_url(self) -> None:
         got = release_asset_url("https://example.invalid/releases/download/", "6.0.0-rc.1", "install.sh")
         self.assertEqual(got, "https://example.invalid/releases/download/v6.0.0-rc.1/install.sh")
@@ -113,6 +211,14 @@ class UnifiedAgentRCRehearsalTest(unittest.TestCase):
                 "expected_agent_name": [],
                 "expected_online_agents": None,
                 "json": False,
+                "skip_asset_checks": False,
+                "runtime_v5_agent": None,
+                "runtime_v6_agent": None,
+                "runtime_expected_from": "v5.1.34",
+                "runtime_expected_to": None,
+                "runtime_timeout": 60.0,
+                "runtime_work_dir": None,
+                "runtime_keep_work_dir": False,
             },
         )()
         results = run_rehearsal(args)
@@ -217,6 +323,90 @@ class UnifiedAgentRCRehearsalTest(unittest.TestCase):
         self.assertIn("`FAIL` `agent-binary-linux-amd64`", report)
         self.assertIn("## Manual Follow-up", report)
 
+    def test_render_text_accepts_one_pass_iterable(self) -> None:
+        result = type(
+            "R",
+            (),
+            {
+                "name": "local-runtime-self-update",
+                "ok": True,
+                "detail": "runtime proof passed",
+            },
+        )()
+        rendered = render_text(item for item in [result])
+        self.assertIn("[PASS] local-runtime-self-update", rendered)
+        self.assertIn("Runtime follow-up covered", rendered)
+
+    def test_main_refuses_empty_runtime_only_selection(self) -> None:
+        exit_code = run_main(
+            [
+                "--base-url",
+                f"{self.base_url}/pulse",
+                "--expected-version",
+                "6.0.0-rc.1",
+                "--release-base-url",
+                f"{self.base_url}/releases",
+                "--skip-asset-checks",
+                "--json",
+            ]
+        )
+        self.assertEqual(exit_code, 1)
+
+    def test_main_refuses_arch_when_asset_checks_are_skipped(self) -> None:
+        exit_code = run_main(
+            [
+                "--base-url",
+                f"{self.base_url}/pulse",
+                "--expected-version",
+                "6.0.0-rc.1",
+                "--release-base-url",
+                f"{self.base_url}/releases",
+                "--skip-asset-checks",
+                "--arch",
+                "linux-amd64",
+                "--json",
+            ]
+        )
+        self.assertEqual(exit_code, 1)
+
+    def test_local_runtime_self_update_exercises_process_swap(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            v5, v6 = self.write_fake_runtime_agents(Path(tmp))
+
+            result = check_local_runtime_self_update(
+                v5_agent=str(v5),
+                v6_agent=str(v6),
+                expected_from="v5.1.34",
+                expected_to="v6.0.0-rc.6",
+                timeout=10,
+                work_dir=None,
+                keep_work_dir=False,
+            )
+
+        self.assertTrue(result.ok, result.detail)
+
+    def test_local_runtime_self_update_keeps_temp_work_dir_when_requested(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            v5, v6 = self.write_fake_runtime_agents(Path(tmp))
+
+            result = check_local_runtime_self_update(
+                v5_agent=str(v5),
+                v6_agent=str(v6),
+                expected_from="v5.1.34",
+                expected_to="v6.0.0-rc.6",
+                timeout=10,
+                work_dir=None,
+                keep_work_dir=True,
+            )
+
+        self.assertTrue(result.ok, result.detail)
+        work_dir = Path(result.detail.rsplit("work_dir=", maxsplit=1)[1])
+        try:
+            self.assertTrue(work_dir.exists())
+            self.assertTrue((work_dir / "pulse-agent").exists())
+        finally:
+            shutil.rmtree(work_dir, ignore_errors=True)
+
     def test_main_writes_report(self) -> None:
         install = b"#!/bin/sh\necho install\n"
         ps1 = b"Write-Output 'install'\n"