feat: drive isolated remote synthetic replay with namespaced pod restart

Adds runtime: remote to the synthetic pilot driver (external base URL and
disposable database, restart via the driver's namespaced ServiceAccount,
UTC phase windows for external cgroup sampling) and the target replay
profile pinned to the deployed release digest and 3 Gunicorn workers.
Five focused tests added; 22 tests pass, ruff clean.

For railiance-apps RAPPS-WP-0014-T04.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 8340@bnt-lap001
Assistant-Session: 2e192b3d-b3e8-4403-94d4-d55a2c8b476c

config/synthetic-pilot-target-remote.json

@@ -0,0 +1,53 @@
+{
+  "users": 2,
+  "workers": 3,
+  "iterations": 100000,
+  "tenders": 100,
+  "document_bytes": 1048576,
+  "think_seconds": 0.2,
+  "timeout_seconds": 30,
+  "ordinary_p95_seconds": 2.0,
+  "duration_seconds": 180,
+  "idle_seconds": 30,
+  "runtime": "remote",
+  "app_image": "forgejo.coulomb.social/coulomb/vergabe-teilnahme@sha256:a26444f59c259698159c69ccb96f73dc648a261ece4c86bb2037a9d977870d91",
+  "app_cpu_limit": 1.0,
+  "app_memory_limit": "1Gi",
+  "operations": [
+    "list",
+    "search",
+    "tender-create",
+    "detail",
+    "tender-edit",
+    "lot-create",
+    "lot-edit",
+    "task-create",
+    "task-edit",
+    "document-upload",
+    "document-download"
+  ],
+  "phases": [
+    {
+      "name": "ordinary",
+      "kind": "load",
+      "duration_seconds": 90,
+      "think_seconds": 0.2
+    },
+    {
+      "name": "burst",
+      "kind": "load",
+      "duration_seconds": 30,
+      "think_seconds": 0
+    },
+    {
+      "name": "restart",
+      "kind": "restart"
+    },
+    {
+      "name": "recovery",
+      "kind": "load",
+      "duration_seconds": 60,
+      "think_seconds": 0.2
+    }
+  ]
+}

docs/synthetic-pilot.md

@@ -157,3 +157,17 @@ hardware and shared-contention evidence belongs to receiving task
 `railiance-apps/workplans/RAPPS-WP-0014-vergabe-invited-pilot.md` T04, linked to
 CUST-WP-0071-T02 and RESOURCE-WP-0008-T02. Its routine read-only capture/preparation
 is authorized; customer datasets must not be used for benchmark writes.
+
+## Remote runtime for an isolated target replay
+
+`runtime: remote` drives an already running isolated synthetic deployment instead
+of starting Docker. `SYNTHETIC_BASE_URL` and `SYNTHETIC_DATABASE_URL` are required
+and must point at a disposable app/database, never a customer dataset. A restart
+phase deletes the single ready pod matched by `SYNTHETIC_RESTART_SELECTOR` in the
+driver's own namespace (namespaced ServiceAccount) and waits for a different ready
+replacement. Resource sampling is external in this mode; every phase and the idle
+window retain UTC start/finish so external cgroup samples can be aligned.
+`config/synthetic-pilot-target-remote.json` is the phased profile with the deployed
+release's image digest and Gunicorn worker count (image CMD: 3). The 2026-10-21
+replay on railiance01 and its manifests/evidence are retained in
+`railiance-apps/docs/evidence/2026-10-21-vergabe-target-replay/`.

scripts/synthetic_pilot.py

@@ -1,5 +1,13 @@
 #!/usr/bin/env python3
-"""Provisional loopback HTTP workload using Gunicorn and disposable PostgreSQL."""
+"""Provisional HTTP workload against Gunicorn and a disposable PostgreSQL.
+
+Runtimes: ``host`` and ``container`` start everything locally. ``remote`` drives
+an already running isolated synthetic deployment (for example a throwaway
+Kubernetes namespace): the base URL, database URL and optional restart pod
+selector come from ``SYNTHETIC_BASE_URL``, ``SYNTHETIC_DATABASE_URL`` and
+``SYNTHETIC_RESTART_SELECTOR``. Remote resource sampling is external; phase UTC
+windows are retained so samples can be aligned afterwards.
+"""
 
 import argparse
 import concurrent.futures
@@ -97,8 +105,8 @@ def phase_plan(config):
             raise ValueError("phase names must be unique and not idle/startup")
         names.add(name)
         if item.get("kind") == "restart":
-            if config.get("runtime") != "container":
-                raise ValueError("restart phase requires container runtime")
+            if config.get("runtime") not in ("container", "remote"):
+                raise ValueError("restart phase requires container or remote runtime")
         elif item.get("kind") == "load":
             for key in ("duration_seconds", "think_seconds"):
                 value = item.get(key)
@@ -144,6 +152,70 @@ def container_url(name):
     return f"http://127.0.0.1:{port}"
 
 
+KUBE_SA = Path("/var/run/secrets/kubernetes.io/serviceaccount")
+
+
+def kube_request(method, path, sa_dir=KUBE_SA):
+    """Minimal in-cluster API call using the pod's own namespaced ServiceAccount."""
+    host = os.environ["KUBERNETES_SERVICE_HOST"]
+    port = os.environ.get("KUBERNETES_SERVICE_PORT", "443")
+    import ssl
+
+    context = ssl.create_default_context(cafile=str(sa_dir / "ca.crt"))
+    req = urllib.request.Request(
+        f"https://{host}:{port}{path}",
+        method=method,
+        headers={"Authorization": "Bearer " + (sa_dir / "token").read_text().strip()},
+    )
+    with urllib.request.urlopen(req, timeout=10, context=context) as response:
+        return json.loads(response.read())
+
+
+def ready_pods(listing):
+    return [
+        item["metadata"]["name"]
+        for item in listing.get("items", [])
+        if not item["metadata"].get("deletionTimestamp")
+        and any(
+            c["type"] == "Ready" and c["status"] == "True"
+            for c in item.get("status", {}).get("conditions", [])
+        )
+    ]
+
+
+def restart_kubernetes_pod(base_url, selector, timeout=180, sa_dir=KUBE_SA, request=None):
+    """Delete the single selected pod in this namespace and wait for a new ready one."""
+    request = request or (lambda method, path: kube_request(method, path, sa_dir))
+    namespace = (sa_dir / "namespace").read_text().strip()
+    query = urllib.parse.urlencode({"labelSelector": selector})
+    pods = f"/api/v1/namespaces/{namespace}/pods"
+    before = ready_pods(request("GET", f"{pods}?{query}"))
+    if len(before) != 1:
+        raise RuntimeError("restart requires exactly one ready selected pod")
+    request("DELETE", f"{pods}/{before[0]}")
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        listing = request("GET", f"{pods}?{query}")
+        names = [i["metadata"]["name"] for i in listing.get("items", [])]
+        if before[0] not in names and ready_pods(listing):
+            wait_ready(base_url, timeout=max(1, deadline - time.monotonic()))
+            return base_url
+        time.sleep(0.5)
+    raise RuntimeError("replacement pod readiness timeout")
+
+
+def settings_text(media, issue_db, allowed_hosts=("127.0.0.1",)):
+    return (
+        "from vergabe_teilnahme.settings.base import *\n"
+        f"DEBUG = False\nALLOWED_HOSTS = {list(allowed_hosts)!r}\nNETKINGDOM_ENABLED = False\n"
+        f"MEDIA_ROOT = {media!r}\n"
+        f"ISSUE_FACADE_LOCAL_DB = {issue_db!r}\n"
+        'ISSUE_FACADE_GITEA = None\nFORCE_SCRIPT_NAME = None\nMEDIA_URL = "/media/"\n'
+        'EMAIL_BACKEND = "django.core.mail.backends.locmem.EmailBackend"\n'
+        "SESSION_COOKIE_SECURE = False\nCSRF_COOKIE_SECURE = False\n"
+    )
+
+
 def restart_container(name):
     subprocess.run(
         ["docker", "restart", "--time", "5", name],
@@ -182,7 +254,7 @@ def validate(config):
         raise ValueError("duration must be positive and idle nonnegative")
     if config["document_bytes"] < 8:
         raise ValueError("document_bytes must be at least 8")
-    if config.get("runtime", "host") not in ("host", "container"):
+    if config.get("runtime", "host") not in ("host", "container", "remote"):
         raise ValueError("unsupported runtime")
     if config.get("runtime") == "container":
         if (
@@ -221,8 +293,10 @@ def guarded_sampler(function, errors):
         errors.append({"sampler": function.__name__, "error_type": type(exc).__name__})
 
 
-def sampling_gaps(samples, db_samples, errors):
+def sampling_gaps(samples, db_samples, errors, remote=False):
     gaps = list(errors)
+    if remote:
+        return gaps
     if any(s.get("sample_error") for s in samples):
         gaps.append({"sampler": "application", "reason": "sample failed"})
     good = [s for s in samples if "monotonic_seconds" in s]
@@ -243,17 +317,10 @@ def run(config, root, failure_output=None):
     os.environ["SECRET_KEY"] = "synthetic-disposable-not-a-deployment-key"
     os.environ["DJANGO_SETTINGS_MODULE"] = "synthetic_settings"
     container_mode = config.get("runtime", "host") == "container"
+    remote_mode = config.get("runtime") == "remote"
     media = "/tmp/synthetic-media" if container_mode else str(root / "media")
     issue_db = "/tmp/synthetic-issues.db" if container_mode else str(root / "issues.db")
-    (root / "synthetic_settings.py").write_text(
-        "from vergabe_teilnahme.settings.base import *\n"
-        'DEBUG = False\nALLOWED_HOSTS = ["127.0.0.1"]\nNETKINGDOM_ENABLED = False\n'
-        f"MEDIA_ROOT = {media!r}\n"
-        f"ISSUE_FACADE_LOCAL_DB = {issue_db!r}\n"
-        'ISSUE_FACADE_GITEA = None\nFORCE_SCRIPT_NAME = None\nMEDIA_URL = "/media/"\n'
-        'EMAIL_BACKEND = "django.core.mail.backends.locmem.EmailBackend"\n'
-        "SESSION_COOKIE_SECURE = False\nCSRF_COOKIE_SECURE = False\n"
-    )
+    (root / "synthetic_settings.py").write_text(settings_text(media, issue_db))
     sys.path.insert(0, str(root))
     import django
 
@@ -289,7 +356,9 @@ def run(config, root, failure_output=None):
     sampling_lock = threading.Lock()
     try:
         startup_started = time.monotonic()
-        if container_mode:
+        if remote_mode:
+            base_url = os.environ["SYNTHETIC_BASE_URL"].rstrip("/")
+        elif container_mode:
             app_container = os.environ["SYNTHETIC_CONTAINER"] + "-app"
             root.chmod(0o755)
             subprocess.run(
@@ -466,13 +535,16 @@ def run(config, root, failure_output=None):
         db_sampler = threading.Thread(
             target=guarded_sampler, args=(sample_db, sampler_errors), daemon=True
         )
-        sampler_threads.extend([sampler, db_sampler])
-        sampler.start()
-        db_sampler.start()
+        if not remote_mode:
+            sampler_threads.extend([sampler, db_sampler])
+            sampler.start()
+            db_sampler.start()
         wait_ready(base_url)
         startup_seconds = time.monotonic() - startup_started
         phase.update(name="idle", kind="idle")
+        idle_window = {"name": "idle", "kind": "idle", "started_at": datetime.now(UTC).isoformat()}
         time.sleep(config["idle_seconds"])
+        idle_window["finished_at"] = datetime.now(UTC).isoformat()
         plan = phase_plan(config)
         active_plan = []
         phase_results = []
@@ -678,10 +750,16 @@ def run(config, root, failure_output=None):
         for active_phase in plan:
             phase.update(name=active_phase["name"], kind=active_phase["kind"])
             phase_started = time.monotonic()
+            phase_started_at = datetime.now(UTC).isoformat()
             if active_phase["kind"] == "restart":
                 old_url = base_url
                 with sampling_lock:
-                    base_url = restart_container(app_container)
+                    if remote_mode:
+                        base_url = restart_kubernetes_pod(
+                            base_url, os.environ["SYNTHETIC_RESTART_SELECTOR"]
+                        )
+                    else:
+                        base_url = restart_container(app_container)
                 transport_rebinds.append(
                     {"phase": active_phase["name"], "before": old_url, "after": base_url}
                 )
@@ -690,6 +768,8 @@ def run(config, root, failure_output=None):
                         "name": active_phase["name"],
                         "kind": "restart",
                         "seconds": time.monotonic() - phase_started,
+                        "started_at": phase_started_at,
+                        "finished_at": datetime.now(UTC).isoformat(),
                         "ready": True,
                     }
                 )
@@ -711,6 +791,8 @@ def run(config, root, failure_output=None):
                     "name": active_phase["name"],
                     "kind": "load",
                     "seconds": time.monotonic() - phase_started,
+                    "started_at": phase_started_at,
+                    "finished_at": datetime.now(UTC).isoformat(),
                     "operations": active_plan,
                     "expected_requests": expected,
                     "actual_requests": len(phase_rows),
@@ -726,6 +808,7 @@ def run(config, root, failure_output=None):
         result = {
             "schema_version": 2,
             "startup_seconds": startup_seconds,
+            "idle_window": idle_window,
             "transport_rebinds": transport_rebinds,
             "phase_results": phase_results,
             "started_at": started,
@@ -736,8 +819,12 @@ def run(config, root, failure_output=None):
                 "python": platform.python_version(),
                 "django": django.get_version(),
                 "platform": platform.platform(),
-                "database": "disposable PostgreSQL 16 container",
-                "transport": "loopback HTTP Gunicorn",
+                "database": os.environ.get(
+                    "SYNTHETIC_DATABASE_DESCRIPTION", "disposable PostgreSQL 16 container"
+                ),
+                "transport": os.environ.get(
+                    "SYNTHETIC_TRANSPORT_DESCRIPTION", "loopback HTTP Gunicorn"
+                ),
                 "workers": config["workers"],
             },
             "ordinary": ordinary,
@@ -764,7 +851,9 @@ def run(config, root, failure_output=None):
         sampling_done.set()
         for thread in sampler_threads:
             thread.join(timeout=7)
-        gaps = sampling_gaps(samples, db_samples, sampler_errors)
+        gaps = sampling_gaps(samples, db_samples, sampler_errors, remote=remote_mode)
+        if remote_mode:
+            result["resource_sampling"] = "external; align with phase UTC windows"
         if any(thread.is_alive() for thread in sampler_threads):
             gaps.append({"sampler": "cleanup", "reason": "thread did not stop"})
         result["measurement_gaps"] = gaps
@@ -825,6 +914,45 @@ def run(config, root, failure_output=None):
         connections.close_all()
 
 
+def host_facts(config):
+    return {
+        "logical_cpus": os.cpu_count(),
+        "affinity_cpus": len(os.sched_getaffinity(0)),
+        "cpu_model": next(
+            (
+                x.split(":", 1)[1].strip()
+                for x in Path("/proc/cpuinfo").read_text().splitlines()
+                if x.startswith("model name")
+            ),
+            "unknown",
+        ),
+        "memory_total": next(
+            x for x in Path("/proc/meminfo").read_text().splitlines() if x.startswith("MemTotal:")
+        ),
+        "cgroup_cpu_max": Path("/sys/fs/cgroup/cpu.max").read_text().strip()
+        if Path("/sys/fs/cgroup/cpu.max").exists()
+        else "unavailable",
+        "app_cpu_quota": config.get("app_cpu_limit", "not imposed by harness"),
+    }
+
+
+def main_remote(config, args):
+    """Drive an existing isolated synthetic deployment; never a shared database."""
+    for name in ("SYNTHETIC_BASE_URL", "SYNTHETIC_DATABASE_URL"):
+        if not os.environ.get(name):
+            raise SystemExit(f"{name} is required for remote runtime")
+    with tempfile.TemporaryDirectory(prefix="vergabe-synthetic-") as directory:
+        result = run(config, Path(directory), args.output)
+    result["source_revision"] = os.environ.get("SYNTHETIC_SOURCE_REVISION", "unrecorded")
+    result["driver_sha256"] = hashlib.sha256(Path(__file__).read_bytes()).hexdigest()
+    result["host"] = host_facts(config)
+    result["config_sha256"] = hashlib.sha256(args.config.read_bytes()).hexdigest()
+    args.output.write_text(json.dumps(result, indent=2) + "\n")
+    keys = ("ordinary", "document_transfer", "complete", "provisional_pass")
+    print(json.dumps({k: result[k] for k in keys}))
+    return 0 if result["provisional_pass"] and result["measurement_complete"] else 1
+
+
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--config", type=Path, required=True)
@@ -832,6 +960,8 @@ if __name__ == "__main__":
     args = parser.parse_args()
     config = validate(json.loads(args.config.read_text()))
     sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+    if config.get("runtime") == "remote":
+        sys.exit(main_remote(config, args))
     name = "vergabe-synthetic-" + uuid.uuid4().hex[:12]
     network = name + "-network"
     network_args = []

scripts/test_synthetic_pilot.py

@@ -239,5 +239,72 @@ class SizingTests(unittest.TestCase):
         )
 
 
+class RemoteRuntimeTests(unittest.TestCase):
+    def config(self):
+        return SyntheticContractTests.config(None) | {
+            "runtime": "remote",
+            "phases": [
+                {"name": "ordinary", "kind": "load", "duration_seconds": 1, "think_seconds": 0},
+                {"name": "restart", "kind": "restart"},
+            ],
+        }
+
+    def test_remote_runtime_accepts_restart_and_host_rejects_it(self):
+        self.assertEqual(pilot.validate(self.config())["runtime"], "remote")
+        with self.assertRaises(ValueError):
+            pilot.validate(self.config() | {"runtime": "host"})
+
+    def test_remote_mode_skips_local_sampler_coverage(self):
+        self.assertEqual(pilot.sampling_gaps([], [], [], remote=True), [])
+        self.assertTrue(pilot.sampling_gaps([], [], []))
+
+    def test_settings_text_sets_allowed_hosts_and_disables_sso(self):
+        text = pilot.settings_text("/m", "/i.db", ["vergabe-app"])
+        self.assertIn("ALLOWED_HOSTS = ['vergabe-app']", text)
+        self.assertIn("NETKINGDOM_ENABLED = False", text)
+
+    def test_restart_deletes_single_ready_pod_and_waits_for_replacement(self):
+        ready = {"type": "Ready", "status": "True"}
+
+        def pod(name, conditions=(ready,), deleting=False):
+            meta = {"name": name} | ({"deletionTimestamp": "x"} if deleting else {})
+            return {"metadata": meta, "status": {"conditions": list(conditions)}}
+
+        listings = [
+            {"items": [pod("old")]},
+            {"items": [pod("old", deleting=True), pod("new", conditions=())]},
+            {"items": [pod("new")]},
+        ]
+        calls = []
+
+        def request(method, path):
+            calls.append((method, path))
+            return listings.pop(0) if method == "GET" else {}
+
+        with tempfile.TemporaryDirectory() as directory:
+            Path(directory, "namespace").write_text("bench\n")
+            with (
+                mock.patch.object(pilot, "wait_ready") as waited,
+                mock.patch.object(pilot.time, "sleep"),
+            ):
+                url = pilot.restart_kubernetes_pod(
+                    "http://app", "app=x", sa_dir=Path(directory), request=request
+                )
+        self.assertEqual(url, "http://app")
+        self.assertIn(("DELETE", "/api/v1/namespaces/bench/pods/old"), calls)
+        waited.assert_called_once()
+
+    def test_restart_refuses_ambiguous_selection(self):
+        with tempfile.TemporaryDirectory() as directory:
+            Path(directory, "namespace").write_text("bench")
+            with self.assertRaises(RuntimeError):
+                pilot.restart_kubernetes_pod(
+                    "http://app",
+                    "app=x",
+                    sa_dir=Path(directory),
+                    request=lambda method, path: {"items": []},
+                )
+
+
 if __name__ == "__main__":
     unittest.main()