From 5e5f230e24696d3b1b43d7df4081a70a94f7776b Mon Sep 17 00:00:00 2001
From: Durable Workflow <support@durable-workflow.com>
Date: Wed, 9 Sep 2026 13:47:48 +0000
Subject: [PATCH 1/3] Exercise standard workflows and expose daily soak
 measurements

---
 .github/workflows/server-perf-soak.yml        |  11 +
 .../v1/bindings/php/capacity_adapter.php      |   4 +
 docs/perf-runner.md                           |  30 ++-
 scripts/perf/Dockerfile.sdk                   |  14 +
 scripts/perf/run-server-soak.sh               |  11 +-
 scripts/perf/run-vultr-soak.sh                |   1 +
 scripts/perf/server_soak.py                   | 239 +++++++++++++-----
 scripts/perf/standard-workflow.compose.yml    |  15 ++
 scripts/perf/standard_workflow_soak.php       |  81 ++++++
 tests/Unit/Support/server_soak_test.py        |  56 ++++
 10 files changed, 394 insertions(+), 68 deletions(-)
 create mode 100644 scripts/perf/Dockerfile.sdk
 create mode 100644 scripts/perf/standard-workflow.compose.yml
 create mode 100644 scripts/perf/standard_workflow_soak.php

diff --git a/.github/workflows/server-perf-soak.yml b/.github/workflows/server-perf-soak.yml
index 7fa1153e..5633d44b 100644
--- a/.github/workflows/server-perf-soak.yml
+++ b/.github/workflows/server-perf-soak.yml
@@ -50,6 +50,7 @@ jobs:
           DW_PERF_CONCURRENCY: ${{ github.event_name == 'workflow_dispatch' && inputs.concurrency || '24' }}
           DW_PERF_NAMESPACES: "8"
           DW_PERF_TASK_QUEUES: "16"
+          DW_PERF_STANDARD_WORKFLOWS: "true"
           DW_PERF_REDIS_CACHE_DB: "1"
           DW_PERF_MAX_SERVER_MEMORY_MB: "1024"
           DW_PERF_MAX_POLLING_KEYS: "2048"
@@ -66,6 +67,15 @@ jobs:
           DW_PERF_RUNNER_ENVIRONMENT: "self-hosted"
         run: scripts/perf/run-vultr-soak.sh
 
+      - name: Summarize measured workload
+        if: always()
+        run: |
+          if [ -f build/perf/summary.md ]; then
+            cat build/perf/summary.md >> "$GITHUB_STEP_SUMMARY"
+          else
+            printf '## Server endurance soak\n\nNo complete measurement summary was produced. Inspect setup, timeout, and provisioning logs; this is not a pass.\n' >> "$GITHUB_STEP_SUMMARY"
+          fi
+
       - name: Upload perf artifacts
         if: always()
         uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a  # v7
@@ -73,3 +83,4 @@ jobs:
           name: server-perf-soak
           path: build/perf/
           if-no-files-found: warn
+          retention-days: 14
diff --git a/benchmarks/capacity/v1/bindings/php/capacity_adapter.php b/benchmarks/capacity/v1/bindings/php/capacity_adapter.php
index bdc4738e..211d726f 100644
--- a/benchmarks/capacity/v1/bindings/php/capacity_adapter.php
+++ b/benchmarks/capacity/v1/bindings/php/capacity_adapter.php
@@ -623,6 +623,10 @@ function capacityConformanceEvidence(array $fixtures): array
     ];
 }
 
+if (realpath($_SERVER['SCRIPT_FILENAME'] ?? '') !== __FILE__) {
+    return;
+}
+
 $mode = $argv[1] ?? '';
 if ($mode === 'describe') {
     echo json_encode(capacityAdapterDescriptor(), JSON_THROW_ON_ERROR | JSON_UNESCAPED_SLASHES).PHP_EOL;
diff --git a/docs/perf-runner.md b/docs/perf-runner.md
index 324f8da9..6e9e7b9b 100644
--- a/docs/perf-runner.md
+++ b/docs/perf-runner.md
@@ -1,7 +1,35 @@
 # Server Perf Runner
 
 The server perf harness exercises the HTTP worker polling path for bounded
-memory growth and polling-cache cleanup.
+memory growth and polling-cache cleanup, with a low-rate PHP standard-workflow
+canary alongside it. The daily Action summary exposes both results directly.
+
+This is an endurance test, not a capacity claim: the canary uses the existing
+`capacity.v1.one_activity` binding (one 1-KiB echo activity), one workflow at a
+time, with at most one start every five seconds. It verifies each result,
+completed status, and ordered activity history. It skips missed start slots
+instead of accumulating a backlog. Its rate is not the capacity suite's
+concurrency/warmup measurement protocol. A slow correct completion is measured,
+not rejected against an invented throughput target; a result timeout is 30s.
+
+Health, readiness, and cluster-info probes run even without workflow-row growth.
+Artifacts include HTTP p50/p95/p99, backpressure separately from successful
+polls, sampled CPU/memory for Server, its queue worker/scheduler, MySQL, Redis,
+and the PHP SDK process container. The SDK client and worker share a 0.5-CPU,
+256-MiB container on the existing host. A 200 with an invalid health payload,
+missing probe results, or an entirely backpressured poll run fails validation.
+
+`DW_PERF_STANDARD_WORKFLOWS=false` disables only the SDK canary for a focused
+polling reproduction. The Compose wrapper enables it by default. Raw
+`server_soak.py` invocation does not build the SDK image; use the wrapper.
+
+The soak still does not exercise timer/signal/query execution, recovery fault
+injection, all SDK bindings, or HA. Those need the separate conformance and
+performance qualification experiments. Zero cache keys for an unexercised
+policy do not prove that feature works. Review failed setup and cleanup steps
+as well as workload results; neither is a successful experiment. Scheduled
+artifacts are retained for 14 days; record significant findings on the owning
+issue rather than committing generated run directories.
 
 ## Runner Shape
 
diff --git a/scripts/perf/Dockerfile.sdk b/scripts/perf/Dockerfile.sdk
new file mode 100644
index 00000000..a4e80691
--- /dev/null
+++ b/scripts/perf/Dockerfile.sdk
@@ -0,0 +1,14 @@
+FROM composer:2 AS dependencies
+WORKDIR /sdk
+COPY benchmarks/capacity/v1/bindings/php/composer.json benchmarks/capacity/v1/bindings/php/composer.lock ./
+RUN composer install --no-dev --no-interaction --no-progress --prefer-dist
+
+FROM php:8.3-cli
+RUN docker-php-ext-install pcntl
+WORKDIR /app
+COPY benchmarks/capacity/v1/bindings/php benchmarks/capacity/v1/bindings/php
+COPY --from=dependencies /sdk/vendor benchmarks/capacity/v1/bindings/php/vendor
+COPY scripts/perf/standard_workflow_soak.php scripts/perf/standard_workflow_soak.php
+USER 1000:1000
+ENTRYPOINT ["php"]
+CMD ["benchmarks/capacity/v1/bindings/php/capacity_adapter.php", "worker"]
diff --git a/scripts/perf/run-server-soak.sh b/scripts/perf/run-server-soak.sh
index 6c2a97c7..b4902602 100755
--- a/scripts/perf/run-server-soak.sh
+++ b/scripts/perf/run-server-soak.sh
@@ -57,6 +57,7 @@ export DW_WORKER_TOKEN="${DW_WORKER_TOKEN:-}"
 export DW_OPERATOR_TOKEN="${DW_OPERATOR_TOKEN:-}"
 export DW_ADMIN_TOKEN="${DW_ADMIN_TOKEN:-}"
 export DW_AUTH_BACKWARD_COMPATIBLE="${DW_AUTH_BACKWARD_COMPATIBLE:-true}"
+export DW_PERF_STANDARD_WORKFLOWS="${DW_PERF_STANDARD_WORKFLOWS:-true}"
 
 OVERRIDE_FILE="$ARTIFACT_DIR/docker-compose.perf.yml"
 cat > "$OVERRIDE_FILE" <<YAML
@@ -96,6 +97,8 @@ services:
     ports: !override []
 YAML
 
+compose=(docker compose -p "$PROJECT" -f "$ROOT_DIR/docker-compose.yml" -f "$OVERRIDE_FILE" -f "$ROOT_DIR/scripts/perf/standard-workflow.compose.yml")
+
 cleanup() {
   local status=$?
 
@@ -104,13 +107,14 @@ cleanup() {
   docker logs "${PROJECT}-scheduler-1" > "$ARTIFACT_DIR/scheduler.log" 2>&1 || true
   docker logs "${PROJECT}-mysql-1" > "$ARTIFACT_DIR/mysql.log" 2>&1 || true
   docker logs "${PROJECT}-redis-1" > "$ARTIFACT_DIR/redis.log" 2>&1 || true
+  docker logs "${PROJECT}-soak-sdk-1" > "$ARTIFACT_DIR/soak-sdk.log" 2>&1 || true
 
   docker rm -f "$PROMETHEUS_CONTAINER" >/dev/null 2>&1 || true
   if [ -n "$PROMETHEUS_CONFIG_DIR" ]; then
     rm -rf "$PROMETHEUS_CONFIG_DIR"
   fi
 
-  docker compose -p "$PROJECT" -f "$ROOT_DIR/docker-compose.yml" -f "$OVERRIDE_FILE" down -v --remove-orphans || true
+  "${compose[@]}" --profile standard-soak down -v --remove-orphans || true
   exit "$status"
 }
 trap cleanup EXIT
@@ -224,7 +228,10 @@ else
   echo "Starting perf stack with project ${PROJECT} on a dynamic host port"
 fi
 setup_status=0
-docker compose -p "$PROJECT" -f "$ROOT_DIR/docker-compose.yml" -f "$OVERRIDE_FILE" up -d --build --wait || setup_status=$?
+"${compose[@]}" up -d --build --wait || setup_status=$?
+if [ "$setup_status" -eq 0 ] && [ "$DW_PERF_STANDARD_WORKFLOWS" = true ]; then
+  "${compose[@]}" build soak-sdk || setup_status=$?
+fi
 if [ "$setup_status" -ne 0 ]; then
   echo "Perf environment setup failed before product smoke execution; docker compose could not build or start the stack." >&2
   write_environment_setup_failure "$setup_status" "docker_compose_up" "docker compose failed before server_soak.py started"
diff --git a/scripts/perf/run-vultr-soak.sh b/scripts/perf/run-vultr-soak.sh
index a372782a..101738ff 100755
--- a/scripts/perf/run-vultr-soak.sh
+++ b/scripts/perf/run-vultr-soak.sh
@@ -266,6 +266,7 @@ for name in \
   DW_PERF_CONCURRENCY \
   DW_PERF_NAMESPACES \
   DW_PERF_TASK_QUEUES \
+  DW_PERF_STANDARD_WORKFLOWS \
   DW_PERF_REDIS_CACHE_DB \
   DW_PERF_MAX_SERVER_MEMORY_MB \
   DW_PERF_MAX_POLLING_KEYS \
diff --git a/scripts/perf/server_soak.py b/scripts/perf/server_soak.py
index 716c66f7..57ee3552 100755
--- a/scripts/perf/server_soak.py
+++ b/scripts/perf/server_soak.py
@@ -173,7 +173,7 @@ def __init__(self) -> None:
         self.lock = threading.Lock()
         self.endpoints: dict[str, dict[str, Any]] = {}
 
-    def record(self, endpoint: str, status: int, latency: float, *, backpressured: bool = False) -> None:
+    def record(self, endpoint: str, status: int, latency: float, *, backpressured: bool = False, valid: bool = True) -> None:
         with self.lock:
             entry = self.endpoints.setdefault(
                 endpoint,
@@ -182,6 +182,8 @@ def record(self, endpoint: str, status: int, latency: float, *, backpressured: b
             entry["requests"] += 1
             entry["statuses"][str(status)] += 1
             entry["latencies"].append(latency)
+            if not valid or (status != 429 and not 200 <= status < 300):
+                entry["errors"] += 1
             if backpressured:
                 entry["backpressured"] = int(entry.get("backpressured", 0)) + 1
 
@@ -230,10 +232,57 @@ def snapshot(self) -> dict[str, Any]:
                             6,
                         ),
                         "max": 0.0 if not latencies else round(latencies[-1], 6),
+                        "p50": percentile(latencies, 0.50),
+                        "p99": percentile(latencies, 0.99),
                     },
                 }
 
-            return snapshot
+        return snapshot
+
+
+def percentile(values: list[float], fraction: float) -> float | None:
+    if not values:
+        return None
+    ordered = sorted(values)
+    return round(ordered[max(0, math.ceil(len(ordered) * fraction) - 1)], 6)
+
+
+def standard_workflow_loop(project: str, duration: int, artifact_dir: Path) -> dict[str, Any]:
+    command = ["docker", "exec", f"{project}-soak-sdk-1", "php", "scripts/perf/standard_workflow_soak.php", str(duration)]
+    try:
+        result = subprocess.run(command, capture_output=True, text=True, timeout=duration + 90, check=False)
+        (artifact_dir / "standard-workflows.jsonl").write_text(result.stdout, encoding="utf-8")
+        (artifact_dir / "standard-workflows.log").write_text(result.stderr, encoding="utf-8")
+        return evaluate_standard_workflows(result.stdout, result.returncode, duration)
+    except subprocess.TimeoutExpired:
+        return {"enabled": True, "failures": ["standard workflow client exceeded its bounded deadline"]}
+
+
+def evaluate_standard_workflows(output: str, exit_code: int, duration: int) -> dict[str, Any]:
+    try:
+        rows = [json.loads(line) for line in output.splitlines() if line.strip()]
+        starts = [row for row in rows if row.get("phase") == "started"]
+        finished = [row for row in rows if row.get("phase") == "finished"]
+        workflows = [row for row in rows if row.get("phase") == "workflow"]
+        completed = [row for row in workflows if row.get("completed") is True]
+        failures = []
+        if exit_code != 0 or len(workflows) != len(completed):
+            failures.append("standard workflow client failed result/history validation or execution")
+        if not workflows or len(starts) != 1 or len(finished) != 1 or float(finished[0]["elapsed_seconds"]) < duration:
+            failures.append("standard workflow measurement was empty, partial, or stopped early")
+        latencies = [float(row["latency_seconds"]) for row in completed]
+        return {
+            "enabled": True, "binding": "php", "workload": "DW Standard Workflow v1",
+            "started": len(workflows), "completed": len(completed), "errors": len(workflows) - len(completed),
+            "duration_seconds": duration, "interval_seconds": 5,
+            "sdk": starts[0].get("sdk") if starts else None,
+            "php": starts[0].get("php") if starts else None,
+            "completed_per_second": round(len(completed) / float(finished[0]["elapsed_seconds"]), 6) if finished else None,
+            "latency_seconds": {f"p{int(p * 100)}": percentile(latencies, p) for p in (0.5, 0.95, 0.99)},
+            "failures": failures,
+        }
+    except (ValueError, KeyError, TypeError, AttributeError):
+        return {"enabled": True, "failures": ["standard workflow client produced malformed evidence"]}
 
 
 class MetricsHandler(BaseHTTPRequestHandler):
@@ -553,7 +602,8 @@ def file_sha256(path: Path) -> str:
 
 
 def compose_command(project: str, *args: str) -> list[str]:
-    return ["docker", "compose", "-p", project, *args]
+    root = Path(__file__).resolve().parents[2]
+    return ["docker", "compose", "-p", project, "-f", str(root / "docker-compose.yml"), "-f", str(root / "scripts/perf/standard-workflow.compose.yml"), *args]
 
 
 def parse_bytes(value: str) -> int:
@@ -578,19 +628,23 @@ def parse_bytes(value: str) -> int:
     return int(amount * scale)
 
 
-def docker_stats(project: str) -> dict[str, int]:
+def docker_stats(project: str, include_sdk: bool = True) -> dict[str, Any]:
     ids_by_service: dict[str, str] = {}
-    for service in ("server", "mysql", "redis"):
+    services = ["server", "worker", "scheduler", "mysql", "redis"]
+    if include_sdk and os.environ.get("DW_PERF_STANDARD_WORKFLOWS") == "true":
+        services.append("soak-sdk")
+    for service in services:
         result = run_command(compose_command(project, "ps", "-q", service))
         container_id = result.stdout.strip()
         if container_id:
             ids_by_service[service] = container_id
 
-    if set(ids_by_service) != {"server", "mysql", "redis"}:
+    if set(ids_by_service) != set(services):
         return {"docker_stats_ok": 0}
 
     result = run_command(["docker", "stats", "--no-stream", "--format", "{{json .}}", *ids_by_service.values()])
     memory_by_id: dict[str, int] = {}
+    cpu_by_id: dict[str, float] = {}
     for line in result.stdout.splitlines():
         try:
             row = json.loads(line)
@@ -601,9 +655,16 @@ def docker_stats(project: str) -> dict[str, int]:
         memory = parse_bytes(mem_usage)
         memory_by_id[row_id] = memory
         memory_by_id[row_id[:12]] = memory
+        try:
+            cpu = float(str(row["CPUPerc"]).removesuffix("%"))
+            if math.isfinite(cpu) and cpu >= 0:
+                cpu_by_id[row_id[:12]] = cpu
+        except (KeyError, ValueError):
+            pass
 
     stats = {f"{service}_memory_bytes": memory_by_id.get(container_id[:12], 0) for service, container_id in ids_by_service.items()}
-    stats["docker_stats_ok"] = 1 if result.returncode == 0 and all(stats.values()) else 0
+    stats["docker_stats_ok"] = 1 if result.returncode == 0 and all(stats.values()) and len(cpu_by_id) == len(services) else 0
+    stats.update({f"{service}_cpu_percent": cpu_by_id.get(container_id[:12]) for service, container_id in ids_by_service.items()})
     health = command_output(
         ["docker", "inspect", "--format", "{{.State.Health.Status}}", ids_by_service["server"]],
     )
@@ -719,10 +780,10 @@ def mysql_counts(project: str) -> dict[str, int]:
     return {"mysql_sample_ok": 0}
 
 
-def sample(project: str) -> dict[str, Any]:
+def sample(project: str, include_sdk: bool = True) -> dict[str, Any]:
     row: dict[str, Any] = {"timestamp": time.time()}
     if project:
-        row.update(docker_stats(project))
+        row.update(docker_stats(project, include_sdk))
         row.update(redis_info(project))
         row.update(mysql_counts(project))
     return row
@@ -978,8 +1039,8 @@ def health_probe_loop(
                     {},
                     timeout_seconds=timeout_seconds,
                 )
-                endpoint_metrics.record(endpoint, status, time.monotonic() - started)
                 body_status = body.get("status") if isinstance(body, dict) else None
+                endpoint_metrics.record(endpoint, status, time.monotonic() - started, valid=body_status == expected_status)
                 if status != 200 or body_status != expected_status:
                     write_jsonl(
                         errors_path,
@@ -1011,7 +1072,7 @@ def cluster_info_probe_loop(
                 auth_headers(token, namespace),
                 timeout_seconds=timeout_seconds,
             )
-            endpoint_metrics.record("cluster_info", status, time.monotonic() - started)
+            endpoint_metrics.record("cluster_info", status, time.monotonic() - started, valid=isinstance(body, dict) and bool(body.get("version")))
             if status != 200 or not isinstance(body, dict) or not body.get("version"):
                 write_jsonl(
                     errors_path,
@@ -1246,6 +1307,80 @@ def github_actions_provenance_present(provenance: dict[str, Any]) -> bool:
     return all(str(provenance.get(field) or "").strip() for field in required_fields)
 
 
+def evaluate_availability(results: dict[str, Any], required: list[str], health_limit: float, control_limit: float) -> list[str]:
+    failures = []
+    for endpoint in required:
+        result = results.get(endpoint, {})
+        if not result.get("requests"):
+            failures.append(f"{endpoint} availability was not sampled")
+        elif result.get("availability", 0) < 1:
+            failures.append(f"{endpoint} availability fell below 1.0")
+        if result.get("errors", 0):
+            failures.append(f"{endpoint} returned request or payload errors")
+        limit = health_limit if endpoint in ("health", "ready") else control_limit if endpoint == "cluster_info" else None
+        if limit is not None and result.get("latency_seconds", {}).get("max", 0) > limit:
+            failures.append(f"{endpoint} latency exceeded {limit}s")
+    # Some backpressure is expected; rejecting every poll is not useful evidence.
+    poll = results.get("worker_poll", {})
+    if "worker_poll" in required and poll.get("requests", 0) <= poll.get("backpressured", 0):
+        failures.append("no non-backpressured worker poll completed")
+    return failures
+
+
+def resource_summary(samples: list[dict[str, Any]]) -> dict[str, Any]:
+    result = {}
+    for service in ("server", "worker", "scheduler", "mysql", "redis", "soak-sdk"):
+        memory = [int(row[f"{service}_memory_bytes"]) / 1048576 for row in samples if f"{service}_memory_bytes" in row]
+        cpu = [float(row[f"{service}_cpu_percent"]) for row in samples if row.get(f"{service}_cpu_percent") is not None]
+        if memory:
+            result[service] = {
+                "peak_memory_mib": round(max(memory), 2),
+                "final_memory_mib": round(memory[-1], 2),
+                "cpu_mean_percent": round(sum(cpu) / len(cpu), 2) if cpu else None,
+                "cpu_peak_percent": round(max(cpu), 2) if cpu else None,
+            }
+    return result
+
+
+def render_summary(summary: dict[str, Any]) -> str:
+    standard = summary.get("standard_workflows", {})
+    lines = [
+        "## Server endurance soak",
+        "",
+        "PASS" if not summary.get("failures") else "FAIL: " + "; ".join(summary["failures"]),
+        "",
+        "This is an endurance canary, not a maximum-capacity benchmark or a failover qualification.",
+        "Poll requests are not workflow completions. Backpressure is reported separately from successful polls.",
+        "",
+        f"Measured duration: {summary.get('duration_seconds')}s. Source: `{summary.get('evidence', {}).get('provenance', {}).get('sha', 'unknown')}`.",
+        f"Server memory slope: {summary.get('server_memory_slope_mb_hour')} MiB/h. Final Server cache keys: {summary.get('final_server_cache_keys')}.",
+        f"Sampling: {summary.get('periodic_sample_count')}/{summary.get('expected_periodic_samples')}; unhealthy samples: {summary.get('sampling_health', {}).get('unhealthy_samples')}.",
+        "",
+        "| HTTP endpoint | Requests | Successful | Backpressure | Errors | p50 s | p95 s | p99 s |",
+        "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: |",
+    ]
+    for name, endpoint in summary.get("request_availability", {}).items():
+        latency = endpoint.get("latency_seconds", {})
+        lines.append(f"| {name} | {endpoint['requests']} | {endpoint['successful']} | {endpoint.get('backpressured', 0)} | {endpoint['errors']} | {latency.get('p50')} | {latency.get('p95')} | {latency.get('p99')} |")
+    lines.extend(["", "### Standard workflows"])
+    if standard.get("enabled"):
+        lines.extend([
+            f"PHP SDK {standard.get('sdk')}, PHP {standard.get('php')}; one 1-KiB echo activity per workflow, closed-loop with a 5s start interval.",
+            f"Completed: {standard.get('completed')} / {standard.get('started')} attempts. Errors: {standard.get('errors')}. Observed completions/s: {standard.get('completed_per_second')}.",
+            f"Start-to-result latency (seconds): {standard.get('latency_seconds')}.",
+            "Every counted completion has its result, completed status, and ordered activity history checked. This is not the capacity suite's concurrency/warmup protocol.",
+        ])
+    else:
+        lines.append("Not exercised in this run.")
+    lines.extend(["", "| Component | Peak MiB | Final MiB | Mean CPU % | Peak CPU % |", "| --- | ---: | ---: | ---: | ---: |"])
+    for name, row in summary.get("resources", {}).items():
+        lines.append(f"| {name} | {row['peak_memory_mib']} | {row['final_memory_mib']} | {row['cpu_mean_percent']} | {row['cpu_peak_percent']} |")
+    lines.extend(["", "CPU uses Docker's scale: 100% is one logical CPU. Samples are not whole-host peak RSS.",
+                  "Not covered: timer/signal/query execution, backend failure injection, cross-SDK parity, or HA. Zero keys for an unexercised policy do not qualify that feature.",
+                  "Inspect the artifact's errors and provisioning log as well: workload PASS does not override setup, artifact-upload, or host-cleanup failure.", ""])
+    return "\n".join(lines)
+
+
 def main() -> int:
     args = parse_args()
 
@@ -1266,6 +1401,8 @@ def main() -> int:
     base_url = args.base_url.rstrip("/")
     started_at = datetime.now(timezone.utc)
     started_monotonic = time.monotonic()
+    standard_enabled = os.environ.get("DW_PERF_STANDARD_WORKFLOWS") == "true"
+    standard_result: dict[str, Any] = {"enabled": False}
 
     try:
         emit_progress(f"waiting for health at {base_url}")
@@ -1276,6 +1413,11 @@ def main() -> int:
         queues = [f"perf-queue-{index:03d}" for index in range(max(1, args.task_queues))]
         create_namespaces(base_url, args.token, namespaces)
         workers = register_workers(base_url, args.token, namespaces, queues)
+        if standard_enabled:
+            if not args.compose_project:
+                raise ValueError("Standard workflows require the isolated Compose fixture.")
+            create_namespaces(base_url, args.token, ["perf-standard"])
+            run_command(compose_command(args.compose_project, "up", "-d", "--no-deps", "soak-sdk"), timeout=60).check_returncode()
         resolved_artifact_versions = artifact_versions(base_url, args.token, namespaces[0])
         emit_progress(
             f"registered {len(workers)} workers across {len(namespaces)} namespaces and {len(queues)} task queues"
@@ -1292,8 +1434,14 @@ def main() -> int:
             f"workflow_runs={args.workflow_runs} and sample_interval={sample_interval}s"
         )
 
-        growth_workers = max(1, args.start_concurrency) + 3 if args.workflow_runs > 0 else 0
-        with ThreadPoolExecutor(max_workers=max(1, args.concurrency) + growth_workers) as executor:
+        growth_workers = max(1, args.start_concurrency) + 1 if args.workflow_runs > 0 else 0
+        with ThreadPoolExecutor(max_workers=max(1, args.concurrency) + growth_workers + 3) as executor:
+            standard_future = None
+            if standard_enabled:
+                standard_future = executor.submit(standard_workflow_loop, args.compose_project, args.duration_seconds, artifact_dir)
+                futures.append(standard_future)
+            futures.append(executor.submit(health_probe_loop, stop_at, base_url, args.health_interval_seconds, args.max_health_latency_seconds, endpoint_metrics, errors_path))
+            futures.append(executor.submit(cluster_info_probe_loop, stop_at, base_url, args.token, namespaces[0], args.control_plane_interval_seconds, args.max_control_plane_latency_seconds, endpoint_metrics, errors_path))
             for index in range(max(1, args.concurrency)):
                 futures.append(
                     executor.submit(
@@ -1336,30 +1484,6 @@ def main() -> int:
                         errors_path,
                     )
                 )
-                futures.append(
-                    executor.submit(
-                        health_probe_loop,
-                        stop_at,
-                        base_url,
-                        args.health_interval_seconds,
-                        args.max_health_latency_seconds,
-                        endpoint_metrics,
-                        errors_path,
-                    )
-                )
-                futures.append(
-                    executor.submit(
-                        cluster_info_probe_loop,
-                        stop_at,
-                        base_url,
-                        args.token,
-                        namespaces[0],
-                        args.control_plane_interval_seconds,
-                        args.max_control_plane_latency_seconds,
-                        endpoint_metrics,
-                        errors_path,
-                    )
-                )
 
             next_sample = time.monotonic()
             while time.monotonic() < stop_at:
@@ -1386,9 +1510,13 @@ def main() -> int:
                     metrics.record_error()
                     write_jsonl(errors_path, {"worker_exception": repr(exception)})
 
+            if standard_future is not None:
+                standard_result = standard_future.result()
+                run_command(compose_command(args.compose_project, "stop", "-t", "10", "soak-sdk")).check_returncode()
+
         emit_progress(f"draining for {max(0, args.drain_seconds)}s before final sample")
         time.sleep(max(0, args.drain_seconds))
-        final_sample = sample(args.compose_project)
+        final_sample = sample(args.compose_project, include_sdk=False)
         samples.append(final_sample)
         metrics.update_sample(final_sample)
         write_jsonl(samples_path, final_sample | {"phase": "final"})
@@ -1488,6 +1616,9 @@ def main() -> int:
             "final_workflow_runs": final_workflow_runs,
             "final_ready_tasks": final_ready_tasks,
             "workflow_growth": workflow_growth,
+            "standard_workflows": standard_result,
+            "resources": resource_summary(samples),
+            "host": {"logical_cpus": os.cpu_count(), "kernel": os.uname().release},
             "polling_observation_status": polling_observation_status,
             "server_memory_slope_mb_hour": None if slope is None else round(slope, 2),
             "sampling_health": sampling_health,
@@ -1515,36 +1646,13 @@ def main() -> int:
             },
         }
 
-        failures = list(workflow_growth_failures)
+        failures = list(workflow_growth_failures) + list(standard_result.get("failures", []))
         if metrics.errors > 0:
             failures.append(f"{metrics.errors} load-generator errors")
+        required_endpoints = ["health", "ready", "cluster_info", "worker_poll"]
         if args.workflow_runs > 0:
-            for endpoint in ("health", "ready", "cluster_info", "workflow_list", "worker_poll"):
-                result = request_availability.get(endpoint, {})
-                if int(result.get("requests") or 0) == 0:
-                    failures.append(f"{endpoint} availability was not sampled during workflow growth")
-                elif float(result.get("availability") or 0.0) < 1.0:
-                    failures.append(
-                        f"{endpoint} availability fell below 1.0 "
-                        f"(observed {result.get('availability')})"
-                    )
-
-            for endpoint in ("health", "ready"):
-                result = request_availability.get(endpoint, {})
-                observed_latency = float((result.get("latency_seconds") or {}).get("max") or 0.0)
-                if observed_latency > args.max_health_latency_seconds:
-                    failures.append(
-                        f"{endpoint} latency exceeded {args.max_health_latency_seconds}s "
-                        f"(observed {observed_latency}s)"
-                    )
-
-            cluster_info = request_availability.get("cluster_info", {})
-            cluster_info_latency = float((cluster_info.get("latency_seconds") or {}).get("max") or 0.0)
-            if cluster_info_latency > args.max_control_plane_latency_seconds:
-                failures.append(
-                    f"cluster_info latency exceeded {args.max_control_plane_latency_seconds}s "
-                    f"(observed {cluster_info_latency}s)"
-                )
+            required_endpoints.append("workflow_list")
+        failures.extend(evaluate_availability(request_availability, required_endpoints, args.max_health_latency_seconds, args.max_control_plane_latency_seconds))
         if periodic_sample_count < min_samples:
             failures.append(
                 f"sample coverage below trusted minimum {min_samples} "
@@ -1636,6 +1744,7 @@ def main() -> int:
 
         metrics_path.write_text(metrics.prometheus(), encoding="utf-8")
         summary_path.write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+        (artifact_dir / "summary.md").write_text(render_summary(summary), encoding="utf-8")
         print(json.dumps(summary, indent=2, sort_keys=True))
 
         return 1 if failures else 0
diff --git a/scripts/perf/standard-workflow.compose.yml b/scripts/perf/standard-workflow.compose.yml
new file mode 100644
index 00000000..9c0719f3
--- /dev/null
+++ b/scripts/perf/standard-workflow.compose.yml
@@ -0,0 +1,15 @@
+services:
+  soak-sdk:
+    profiles: [standard-soak]
+    build:
+      context: .
+      dockerfile: scripts/perf/Dockerfile.sdk
+    environment:
+      DURABLE_WORKFLOW_RUNTIME_URL: http://server:8080
+      DURABLE_WORKFLOW_NAMESPACE: perf-standard
+      DURABLE_WORKFLOW_TASK_QUEUE: perf-standard
+      DURABLE_WORKFLOW_TOKEN: ${DW_AUTH_TOKEN:-perf-token}
+    cpus: 0.5
+    mem_limit: 256m
+    init: true
+    restart: "no"
diff --git a/scripts/perf/standard_workflow_soak.php b/scripts/perf/standard_workflow_soak.php
new file mode 100644
index 00000000..72c5000a
--- /dev/null
+++ b/scripts/perf/standard_workflow_soak.php
@@ -0,0 +1,81 @@
+<?php
+
+declare(strict_types=1);
+
+require dirname(__DIR__, 2).'/benchmarks/capacity/v1/bindings/php/capacity_adapter.php';
+capacityAutoload();
+
+$duration = filter_var($argv[1] ?? null, FILTER_VALIDATE_INT, ['options' => ['min_range' => 1, 'max_range' => 14400]]);
+if ($duration === false) {
+    throw new InvalidArgumentException('Supply a duration between 1 and 14400 seconds.');
+}
+$client = capacityClient(false);
+$queue = capacityEnvironment('DURABLE_WORKFLOW_TASK_QUEUE');
+$readyBy = microtime(true) + 30;
+do {
+    if ($client->listWorkers($queue, 'active') !== []) {
+        break;
+    }
+    if (microtime(true) >= $readyBy) {
+        throw new RuntimeException('Standard workflow worker did not register within 30 seconds.');
+    }
+    usleep(200000);
+} while (true);
+
+$payload = str_repeat('s', 1024);
+$input = [
+    'blob' => $payload,
+    'payload_contract' => [
+        'workflow_input_bytes' => 1024,
+        'workflow_result_bytes' => 1024,
+        'activity_input_bytes' => 1024,
+        'activity_result_bytes' => 1024,
+        'signal_bytes' => 0,
+    ],
+];
+$expectedEvents = ['WorkflowStarted', 'ActivityScheduled', 'ActivityStarted', 'ActivityCompleted', 'WorkflowCompleted'];
+$started = hrtime(true);
+$stopAt = $started + $duration * 1000000000;
+$nextStart = $started;
+$failed = false;
+echo json_encode([
+    'phase' => 'started',
+    'sdk' => Composer\InstalledVersions::getPrettyVersion('durable-workflow/sdk'),
+    'php' => PHP_VERSION,
+    'interval_seconds' => 5,
+    'duration_seconds' => $duration,
+], JSON_THROW_ON_ERROR).PHP_EOL;
+
+// Closed-loop canary: never accumulate unbounded work when the runtime slows.
+while (hrtime(true) < $stopAt) {
+    if (hrtime(true) < $nextStart) {
+        usleep((int) min(200000, ($nextStart - hrtime(true)) / 1000));
+
+        continue;
+    }
+    $workflowId = 'soak-standard-'.bin2hex(random_bytes(12));
+    $attemptAt = hrtime(true);
+    $row = ['workflow_id' => $workflowId, 'phase' => 'workflow', 'completed' => false];
+    try {
+        $handle = $client->startWorkflow('capacity.v1.one_activity', $workflowId, $queue, [$input]);
+        $row['run_id'] = $handle->selectedRunId;
+        $result = $handle->result(30, 1);
+        $row['latency_seconds'] = (hrtime(true) - $attemptAt) / 1000000000;
+        $execution = $handle->describe();
+        $history = $client->workflowHistory($workflowId, $handle->selectedRunId);
+        $events = array_column($history['events'] ?? [], 'event_type');
+        $semanticEvents = array_values(array_intersect($events, $expectedEvents));
+        if ($result !== $payload || $execution->status !== 'completed' || $semanticEvents !== $expectedEvents) {
+            throw new RuntimeException('Standard workflow result, status, or ordered activity history did not match.');
+        }
+        $row['completed'] = true;
+    } catch (Throwable $error) {
+        $row['error'] = get_class($error).': '.$error->getMessage();
+        $failed = true;
+    }
+    echo json_encode($row, JSON_THROW_ON_ERROR | JSON_INVALID_UTF8_SUBSTITUTE).PHP_EOL;
+    // Skip missed start slots instead of generating a catch-up burst.
+    $nextStart = $started + ((int) ((hrtime(true) - $started) / 5000000000) + 1) * 5000000000;
+}
+echo json_encode(['phase' => 'finished', 'elapsed_seconds' => (hrtime(true) - $started) / 1000000000], JSON_THROW_ON_ERROR).PHP_EOL;
+exit($failed ? 1 : 0);
diff --git a/tests/Unit/Support/server_soak_test.py b/tests/Unit/Support/server_soak_test.py
index a5188053..e6db1951 100644
--- a/tests/Unit/Support/server_soak_test.py
+++ b/tests/Unit/Support/server_soak_test.py
@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 
 import importlib.util
+import json
 import os
 from pathlib import Path
 import unittest
@@ -98,5 +99,60 @@ def test_remote_execution_environment_overrides_github_runner_metadata(self) ->
             self.assertEqual("self-hosted", server_soak.runner_environment())
 
 
+class EnduranceCoverageTest(unittest.TestCase):
+    def standard_output(self, completed=True, elapsed=60):
+        return "\n".join(json.dumps(row) for row in [
+            {"phase": "started", "sdk": "test", "php": "test"},
+            {"phase": "workflow", "completed": completed, "latency_seconds": 0.7},
+            {"phase": "finished", "elapsed_seconds": elapsed},
+        ])
+
+    def test_validated_completions_are_distinct_from_poll_requests(self):
+        result = server_soak.evaluate_standard_workflows(self.standard_output(), 0, 60)
+        self.assertEqual([], result["failures"])
+        self.assertEqual(1, result["completed"])
+        self.assertEqual(0.7, result["latency_seconds"]["p99"])
+
+    def test_partial_failed_empty_and_malformed_runs_do_not_pass(self):
+        for output, code in [
+            (self.standard_output(False), 0),
+            (self.standard_output(), 1),
+            (self.standard_output(elapsed=20), 0),
+            ("", 0), ("not json", 0), ("[]", 0),
+        ]:
+            with self.subTest(output=output, code=code):
+                self.assertTrue(server_soak.evaluate_standard_workflows(output, code, 60)["failures"])
+
+    def test_health_failures_are_detected_without_workflow_growth(self):
+        metrics = server_soak.EndpointMetrics()
+        metrics.record("health", 200, 0.1, valid=False)
+        metrics.record("ready", 503, 0.1)
+        failures = server_soak.evaluate_availability(metrics.snapshot(), ["health", "ready", "cluster_info"], 3, 5)
+        self.assertIn("health returned request or payload errors", failures)
+        self.assertIn("ready availability fell below 1.0", failures)
+        self.assertIn("cluster_info availability was not sampled", failures)
+
+    def test_all_backpressure_is_not_a_healthy_poll_experiment(self):
+        for status in (200, 429):
+            metrics = server_soak.EndpointMetrics()
+            metrics.record("worker_poll", status, 0.1, backpressured=status == 200)
+            self.assertTrue(server_soak.evaluate_availability(metrics.snapshot(), ["worker_poll"], 3, 5))
+            metrics.record("worker_poll", 200, 0.1)
+            self.assertEqual([], server_soak.evaluate_availability(metrics.snapshot(), ["worker_poll"], 3, 5))
+
+    def test_resource_summary_does_not_substitute_zero_for_missing_cpu(self):
+        result = server_soak.resource_summary([{"server_memory_bytes": 1048576}])
+        self.assertEqual(1, result["server"]["peak_memory_mib"])
+        self.assertIsNone(result["server"]["cpu_mean_percent"])
+        result = server_soak.resource_summary([{"server_memory_bytes": 1048576, "server_cpu_percent": 0}])
+        self.assertEqual(0, result["server"]["cpu_mean_percent"])
+
+    def test_summary_exposes_missing_standard_workflow_coverage(self):
+        report = server_soak.render_summary({"failures": ["partial"]})
+        self.assertIn("FAIL: partial", report)
+        self.assertIn("Not exercised", report)
+        self.assertIn("not a maximum-capacity benchmark", report)
+
+
 if __name__ == "__main__":
     unittest.main()

From 0997b9c97170954e40b2f85a952b0e8a37940cad Mon Sep 17 00:00:00 2001
From: Durable Workflow <support@durable-workflow.com>
Date: Wed, 9 Sep 2026 14:02:54 +0000
Subject: [PATCH 2/3] Keep soak registrations live and require sustained
 polling coverage

---
 docker-compose.yml                           |   1 +
 docs/perf-runner.md                          |  14 +++
 scripts/perf/run-server-soak.sh              |   3 +-
 scripts/perf/server_soak.py                  | 118 ++++++++++++-------
 scripts/perf/standard_workflow_soak.php      |   3 +-
 tests/Unit/ServerPerfHarnessContractTest.php |   9 +-
 tests/Unit/Support/server_soak_test.py       |  18 +++
 7 files changed, 123 insertions(+), 43 deletions(-)

diff --git a/docker-compose.yml b/docker-compose.yml
index 07773cc0..b156dcba 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -111,6 +111,7 @@ services:
         WORKFLOW_PACKAGE_REF: ${WORKFLOW_PACKAGE_REF:-}
         WORKFLOW_PACKAGE_COMMIT: ${WORKFLOW_PACKAGE_COMMIT:-}
     command: php artisan queue:work --sleep=1 --tries=3 --max-time=3600
+    restart: unless-stopped
     healthcheck:
       test: ["CMD", "server-process-healthcheck", "worker"]
       interval: 5s
diff --git a/docs/perf-runner.md b/docs/perf-runner.md
index 6e9e7b9b..f901de6d 100644
--- a/docs/perf-runner.md
+++ b/docs/perf-runner.md
@@ -18,6 +18,20 @@ polls, sampled CPU/memory for Server, its queue worker/scheduler, MySQL, Redis,
 and the PHP SDK process container. The SDK client and worker share a 0.5-CPU,
 256-MiB container on the existing host. A 200 with an invalid health payload,
 missing probe results, or an entirely backpressured poll run fails validation.
+Compose restarts the Server queue worker after its configured hourly recycle;
+an exited queue worker must not disappear unnoticed for the second soak hour.
+Each synthetic polling thread owns and heartbeats one registration, distributed
+across the configured namespaces and queues (24 workers, not 128 Cartesian
+registrations, in the daily default). HTTP success carrying a stale/rejected
+worker status fails the run. Polling-cache activity must be present in at least
+the configured minimum sample fraction, not merely once at startup.
+
+The default post-load drain is 310 seconds. Real activity poll-result caches
+retain replay evidence through their five-minute lease plus five seconds;
+heartbeat retention throttles and wake signals also outlive the old 12-second
+idle-only drain. No cache flush, shorter production TTL, or relaxed final-key
+limit is used to obtain a pass. The load remains two hours; build/bootstrap,
+bounded drain, and teardown are additional host time.
 
 `DW_PERF_STANDARD_WORKFLOWS=false` disables only the SDK canary for a focused
 polling reproduction. The Compose wrapper enables it by default. Raw
diff --git a/scripts/perf/run-server-soak.sh b/scripts/perf/run-server-soak.sh
index b4902602..f85e937b 100755
--- a/scripts/perf/run-server-soak.sh
+++ b/scripts/perf/run-server-soak.sh
@@ -35,10 +35,11 @@ PROMETHEUS_CONTAINER="${PROJECT}-prometheus"
 PROMETHEUS_CONFIG_DIR=""
 
 mkdir -p "$ARTIFACT_DIR"
+export DW_PERF_DRAIN_SECONDS="${DW_PERF_DRAIN_SECONDS:-310}"
 
 if [ -z "$LOAD_TIMEOUT_SECONDS" ]; then
   DURATION_SECONDS="${DW_PERF_DURATION_SECONDS:-120}"
-  DRAIN_SECONDS="${DW_PERF_DRAIN_SECONDS:-12}"
+  DRAIN_SECONDS="$DW_PERF_DRAIN_SECONDS"
   LOAD_TIMEOUT_SECONDS=$((DURATION_SECONDS + DRAIN_SECONDS + 300))
   if [ "$LOAD_TIMEOUT_SECONDS" -lt 300 ]; then
     LOAD_TIMEOUT_SECONDS=300
diff --git a/scripts/perf/server_soak.py b/scripts/perf/server_soak.py
index 57ee3552..e92eabf2 100755
--- a/scripts/perf/server_soak.py
+++ b/scripts/perf/server_soak.py
@@ -9,7 +9,6 @@
 import json
 import math
 import os
-import random
 import re
 import subprocess
 import sys
@@ -347,7 +346,7 @@ def parse_args() -> argparse.Namespace:
         type=float,
         default=float(os.environ.get("DW_PERF_MAX_CONTROL_PLANE_LATENCY_SECONDS", "5")),
     )
-    parser.add_argument("--drain-seconds", type=int, default=int(os.environ.get("DW_PERF_DRAIN_SECONDS", "12")))
+    parser.add_argument("--drain-seconds", type=int, default=int(os.environ.get("DW_PERF_DRAIN_SECONDS", "310")))
     parser.add_argument("--artifact-dir", default=os.environ.get("DW_PERF_ARTIFACT_DIR", "build/perf"))
     parser.add_argument("--compose-project", default=os.environ.get("DW_PERF_COMPOSE_PROJECT", ""))
     parser.add_argument("--metrics-port", type=int, default=int(os.environ.get("DW_PERF_METRICS_PORT", "19090")))
@@ -536,35 +535,23 @@ def create_namespaces(base_url: str, token: str, namespaces: list[str]) -> None:
 PERF_WORKFLOW_TYPE = "perf.harness.workflow"
 
 
-def register_workers(base_url: str, token: str, namespaces: list[str], queues: list[str]) -> list[tuple[str, str, str]]:
+def register_workers(base_url: str, token: str, namespaces: list[str], queues: list[str], count: int) -> list[tuple[str, str, str]]:
     workers: list[tuple[str, str, str]] = []
-    for namespace in namespaces:
-        for queue in queues:
-            worker_id = f"perf-worker-{namespace}-{queue}"
-            status, body = http_json(
-                "POST",
-                f"{base_url}/api/worker/register",
-                auth_headers(token, namespace, worker=True),
-                {
-                    "worker_id": worker_id,
-                    "task_queue": queue,
-                    # Model a published remote SDK that requires successful
-                    # empty poll responses under capacity backpressure.
-                    "runtime": "python",
-                    "sdk_version": "perf-harness",
-                    "max_concurrent_workflow_tasks": 100,
-                    # Workers must advertise at least one workflow type so
-                    # the server treats them as workflow-task-eligible. A
-                    # registration with no types short-circuits every poll
-                    # at no_workflow_capability and the polling cache surface
-                    # never runs — leaving the bounded-growth smoke without
-                    # any observation of the path it asserts on.
-                    "supported_workflow_types": [PERF_WORKFLOW_TYPE],
-                },
-            )
-            if status not in (200, 201):
-                raise RuntimeError(f"failed to register {worker_id}: HTTP {status}: {body}")
-            workers.append((namespace, queue, worker_id))
+    # Each polling thread owns one registration and maintains its heartbeat.
+    for index in range(max(1, count)):
+        namespace, queue = namespaces[index % len(namespaces)], queues[index % len(queues)]
+        worker_id = f"perf-worker-{namespace}-{queue}-{index}"
+        status, body = http_json(
+            "POST", f"{base_url}/api/worker/register", auth_headers(token, namespace, worker=True),
+            {
+                "worker_id": worker_id, "task_queue": queue, "runtime": "python",
+                "sdk_version": "perf-harness", "max_concurrent_workflow_tasks": 1,
+                "supported_workflow_types": [PERF_WORKFLOW_TYPE],
+            },
+        )
+        if status not in (200, 201):
+            raise RuntimeError(f"failed to register {worker_id}: HTTP {status}: {body}")
+        workers.append((namespace, queue, worker_id))
 
     return workers
 
@@ -639,7 +626,8 @@ def docker_stats(project: str, include_sdk: bool = True) -> dict[str, Any]:
         if container_id:
             ids_by_service[service] = container_id
 
-    if set(ids_by_service) != set(services):
+    required_services = {"server", "mysql", "redis"}
+    if not required_services.issubset(ids_by_service):
         return {"docker_stats_ok": 0}
 
     result = run_command(["docker", "stats", "--no-stream", "--format", "{{json .}}", *ids_by_service.values()])
@@ -663,7 +651,14 @@ def docker_stats(project: str, include_sdk: bool = True) -> dict[str, Any]:
             pass
 
     stats = {f"{service}_memory_bytes": memory_by_id.get(container_id[:12], 0) for service, container_id in ids_by_service.items()}
-    stats["docker_stats_ok"] = 1 if result.returncode == 0 and all(stats.values()) and len(cpu_by_id) == len(services) else 0
+    # Queue-worker recycling can overlap a sample. Record that absence without
+    # treating an intentional process restart as corrupt core-resource evidence.
+    stats["docker_stats_ok"] = int(result.returncode == 0 and all(
+        stats[f"{service}_memory_bytes"] > 0 and ids_by_service[service][:12] in cpu_by_id
+        for service in required_services
+    ))
+    for service in services:
+        stats[f"{service}_running"] = int(stats.get(f"{service}_memory_bytes", 0) > 0)
     stats.update({f"{service}_cpu_percent": cpu_by_id.get(container_id[:12]) for service, container_id in ids_by_service.items()})
     health = command_output(
         ["docker", "inspect", "--format", "{{.State.Health.Status}}", ids_by_service["server"]],
@@ -840,16 +835,27 @@ def worker_loop(
     errors_path: Path,
     worker_index: int,
 ) -> None:
-    rng = random.Random(worker_index)
     sequence = 0
+    namespace, queue, worker_id = workers[worker_index % len(workers)]
+    heartbeat_at = 0.0
 
     while time.monotonic() < stop_at:
-        namespace, queue, worker_id = rng.choice(workers)
         sequence += 1
         poll_request_id = f"perf-{worker_index}-{sequence}-{time.time_ns()}"
         started = time.monotonic()
 
         try:
+            if time.monotonic() >= heartbeat_at:
+                heartbeat_started = time.monotonic()
+                heartbeat_status, heartbeat_body = http_json(
+                    "POST", f"{base_url}/api/worker/heartbeat", auth_headers(token, namespace, worker=True),
+                    {"worker_id": worker_id}, timeout_seconds=5,
+                )
+                endpoint_metrics.record("worker_heartbeat", heartbeat_status, time.monotonic() - heartbeat_started)
+                if heartbeat_status != 200:
+                    raise RuntimeError(f"worker heartbeat failed: HTTP {heartbeat_status}: {heartbeat_body}")
+                heartbeat_at = time.monotonic() + 10
+            started = time.monotonic()
             status, body = http_json(
                 "POST",
                 f"{base_url}/api/worker/workflow-tasks/poll",
@@ -872,11 +878,12 @@ def worker_loop(
                 status,
                 latency,
                 backpressured=compatible_backpressure,
+                valid=poll_response_valid(body),
             )
             if status == 429:
                 retry_after = body.get("retry_after_seconds", 1) if isinstance(body, dict) else 1
                 time.sleep(max(0.05, min(5.0, float(retry_after))))
-            elif status != 200:
+            elif status != 200 or not poll_response_valid(body):
                 metrics.record_error()
                 write_jsonl(errors_path, {"status": status, "body": body, "namespace": namespace, "queue": queue})
         except Exception as exc:  # noqa: BLE001
@@ -885,6 +892,20 @@ def worker_loop(
             write_jsonl(errors_path, {"exception": repr(exc), "namespace": namespace, "queue": queue})
 
 
+def poll_response_valid(body: Any) -> bool:
+    rejected = {"stale_worker_registration", "worker_not_registered", "worker_registration_superseded", "no_workflow_capability", "unsupported", "rejected", "conflict"}
+    return isinstance(body, dict) and body.get("reason") not in rejected and body.get("poll_status") not in rejected
+
+
+def polling_activity_summary(samples: list[dict[str, Any]], minimum_fraction: float) -> dict[str, Any]:
+    active = sum(int(row.get("redis_polling_keys") or 0) > 0 for row in samples)
+    return {
+        "active_samples": active, "samples": len(samples),
+        "fraction": active / len(samples) if samples else 0,
+        "sustained": bool(samples) and active >= math.ceil(len(samples) * minimum_fraction),
+    }
+
+
 def workflow_start_loop(
     stop_at: float,
     base_url: str,
@@ -1338,6 +1359,7 @@ def resource_summary(samples: list[dict[str, Any]]) -> dict[str, Any]:
                 "final_memory_mib": round(memory[-1], 2),
                 "cpu_mean_percent": round(sum(cpu) / len(cpu), 2) if cpu else None,
                 "cpu_peak_percent": round(max(cpu), 2) if cpu else None,
+                "missing_samples": sum(row.get(f"{service}_running") == 0 for row in samples),
             }
     return result
 
@@ -1355,6 +1377,7 @@ def render_summary(summary: dict[str, Any]) -> str:
         f"Measured duration: {summary.get('duration_seconds')}s. Source: `{summary.get('evidence', {}).get('provenance', {}).get('sha', 'unknown')}`.",
         f"Server memory slope: {summary.get('server_memory_slope_mb_hour')} MiB/h. Final Server cache keys: {summary.get('final_server_cache_keys')}.",
         f"Sampling: {summary.get('periodic_sample_count')}/{summary.get('expected_periodic_samples')}; unhealthy samples: {summary.get('sampling_health', {}).get('unhealthy_samples')}.",
+        f"Sustained polling activity: {summary.get('polling_activity', 'not measured by this version')}.",
         "",
         "| HTTP endpoint | Requests | Successful | Backpressure | Errors | p50 s | p95 s | p99 s |",
         "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: |",
@@ -1372,9 +1395,9 @@ def render_summary(summary: dict[str, Any]) -> str:
         ])
     else:
         lines.append("Not exercised in this run.")
-    lines.extend(["", "| Component | Peak MiB | Final MiB | Mean CPU % | Peak CPU % |", "| --- | ---: | ---: | ---: | ---: |"])
+    lines.extend(["", "| Component | Peak MiB | Final MiB | Mean CPU % | Peak CPU % | Missing samples |", "| --- | ---: | ---: | ---: | ---: | ---: |"])
     for name, row in summary.get("resources", {}).items():
-        lines.append(f"| {name} | {row['peak_memory_mib']} | {row['final_memory_mib']} | {row['cpu_mean_percent']} | {row['cpu_peak_percent']} |")
+        lines.append(f"| {name} | {row['peak_memory_mib']} | {row['final_memory_mib']} | {row['cpu_mean_percent']} | {row['cpu_peak_percent']} | {row.get('missing_samples', 0)} |")
     lines.extend(["", "CPU uses Docker's scale: 100% is one logical CPU. Samples are not whole-host peak RSS.",
                   "Not covered: timer/signal/query execution, backend failure injection, cross-SDK parity, or HA. Zero keys for an unexercised policy do not qualify that feature.",
                   "Inspect the artifact's errors and provisioning log as well: workload PASS does not override setup, artifact-upload, or host-cleanup failure.", ""])
@@ -1412,13 +1435,18 @@ def main() -> int:
         namespaces = [f"perf-ns-{index:03d}" for index in range(max(1, args.namespaces))]
         queues = [f"perf-queue-{index:03d}" for index in range(max(1, args.task_queues))]
         create_namespaces(base_url, args.token, namespaces)
-        workers = register_workers(base_url, args.token, namespaces, queues)
+        workers = register_workers(base_url, args.token, namespaces, queues, args.concurrency)
         if standard_enabled:
             if not args.compose_project:
                 raise ValueError("Standard workflows require the isolated Compose fixture.")
             create_namespaces(base_url, args.token, ["perf-standard"])
             run_command(compose_command(args.compose_project, "up", "-d", "--no-deps", "soak-sdk"), timeout=60).check_returncode()
         resolved_artifact_versions = artifact_versions(base_url, args.token, namespaces[0])
+        if args.compose_project:
+            installed = command_output(["docker", "exec", f"{args.compose_project}-server-1", "php", "-r", "require 'vendor/autoload.php'; echo Composer\\InstalledVersions::getPrettyVersion('durable-workflow/workflow');"])
+            if not installed:
+                raise RuntimeError("Could not identify the workflow package actually installed in the Server image.")
+            resolved_artifact_versions["workflow"] = installed
         emit_progress(
             f"registered {len(workers)} workers across {len(namespaces)} namespaces and {len(queues)} task queues"
         )
@@ -1588,6 +1616,7 @@ def main() -> int:
             "duration_seconds": args.duration_seconds,
             "elapsed_seconds": round(elapsed_seconds, 2),
             "concurrency": args.concurrency,
+            "synthetic_worker_registrations": len(workers),
             "workflow_runs_target": args.workflow_runs,
             "start_concurrency": args.start_concurrency,
             "namespaces": len(namespaces),
@@ -1618,8 +1647,12 @@ def main() -> int:
             "workflow_growth": workflow_growth,
             "standard_workflows": standard_result,
             "resources": resource_summary(samples),
-            "host": {"logical_cpus": os.cpu_count(), "kernel": os.uname().release},
+            "host": {
+                "logical_cpus": os.cpu_count(), "kernel": os.uname().release, "architecture": os.uname().machine,
+                "physical_memory_mib": round(os.sysconf("SC_PHYS_PAGES") * os.sysconf("SC_PAGE_SIZE") / 1048576, 2),
+            },
             "polling_observation_status": polling_observation_status,
+            "polling_activity": polling_activity_summary(samples[:-1], sample_coverage),
             "server_memory_slope_mb_hour": None if slope is None else round(slope, 2),
             "sampling_health": sampling_health,
             "request_availability": request_availability,
@@ -1649,10 +1682,12 @@ def main() -> int:
         failures = list(workflow_growth_failures) + list(standard_result.get("failures", []))
         if metrics.errors > 0:
             failures.append(f"{metrics.errors} load-generator errors")
-        required_endpoints = ["health", "ready", "cluster_info", "worker_poll"]
+        required_endpoints = ["health", "ready", "cluster_info", "worker_poll", "worker_heartbeat"]
         if args.workflow_runs > 0:
             required_endpoints.append("workflow_list")
         failures.extend(evaluate_availability(request_availability, required_endpoints, args.max_health_latency_seconds, args.max_control_plane_latency_seconds))
+        if args.compose_project and not summary["polling_activity"]["sustained"]:
+            failures.append("polling cache activity was not sustained through the measurement window")
         if periodic_sample_count < min_samples:
             failures.append(
                 f"sample coverage below trusted minimum {min_samples} "
@@ -1664,6 +1699,9 @@ def main() -> int:
                 f"{sampling_health['unhealthy_samples']} compose-backed samples "
                 f"(field failures: {sampling_health.get('unhealthy_field_counts')})"
             )
+        for service in ("worker", "scheduler"):
+            if args.compose_project and final_sample.get(f"{service}_running") != 1:
+                failures.append(f"{service} was not running after the soak")
         if max_server_memory_bytes > args.max_server_memory_mb * 1024 * 1024:
             failures.append(
                 f"server memory exceeded {args.max_server_memory_mb} MB "
diff --git a/scripts/perf/standard_workflow_soak.php b/scripts/perf/standard_workflow_soak.php
index 72c5000a..274a3bc2 100644
--- a/scripts/perf/standard_workflow_soak.php
+++ b/scripts/perf/standard_workflow_soak.php
@@ -1,6 +1,7 @@
 <?php
 
 declare(strict_types=1);
+use Composer\InstalledVersions;
 
 require dirname(__DIR__, 2).'/benchmarks/capacity/v1/bindings/php/capacity_adapter.php';
 capacityAutoload();
@@ -40,7 +41,7 @@
 $failed = false;
 echo json_encode([
     'phase' => 'started',
-    'sdk' => Composer\InstalledVersions::getPrettyVersion('durable-workflow/sdk'),
+    'sdk' => InstalledVersions::getPrettyVersion('durable-workflow/sdk'),
     'php' => PHP_VERSION,
     'interval_seconds' => 5,
     'duration_seconds' => $duration,
diff --git a/tests/Unit/ServerPerfHarnessContractTest.php b/tests/Unit/ServerPerfHarnessContractTest.php
index 522b13ed..f70cd014 100644
--- a/tests/Unit/ServerPerfHarnessContractTest.php
+++ b/tests/Unit/ServerPerfHarnessContractTest.php
@@ -9,6 +9,13 @@
 
 class ServerPerfHarnessContractTest extends TestCase
 {
+    public function test_time_bounded_queue_worker_is_supervised_between_recycles(): void
+    {
+        $compose = Yaml::parseFile(dirname(__DIR__, 2).'/docker-compose.yml');
+        $this->assertStringContainsString('--max-time=3600', $compose['services']['worker']['command']);
+        $this->assertSame('unless-stopped', $compose['services']['worker']['restart']);
+    }
+
     public function test_contract_qualification_installs_the_verified_workflow_source(): void
     {
         $workflow = file_get_contents(dirname(__DIR__, 2).'/.github/workflows/server-perf.yml');
@@ -389,7 +396,7 @@ public function test_perf_smoke_records_environment_setup_failures_before_load_s
             );
         }
 
-        $composeOffset = strpos($source, 'docker compose -p "$PROJECT" -f "$ROOT_DIR/docker-compose.yml" -f "$OVERRIDE_FILE" up -d --build --wait');
+        $composeOffset = strpos($source, '"${compose[@]}" up -d --build --wait');
         $loadOffset = strpos($source, 'Running perf load against ${BASE_URL}');
         $this->assertIsInt($composeOffset);
         $this->assertIsInt($loadOffset);
diff --git a/tests/Unit/Support/server_soak_test.py b/tests/Unit/Support/server_soak_test.py
index e6db1951..4d96bc85 100644
--- a/tests/Unit/Support/server_soak_test.py
+++ b/tests/Unit/Support/server_soak_test.py
@@ -100,6 +100,24 @@ def test_remote_execution_environment_overrides_github_runner_metadata(self) ->
 
 
 class EnduranceCoverageTest(unittest.TestCase):
+    def test_startup_only_cache_activity_does_not_qualify_a_long_soak(self):
+        rows = [{"redis_polling_keys": 10}] * 7 + [{"redis_polling_keys": 0}] * 1433
+        self.assertFalse(server_soak.polling_activity_summary(rows, 0.8)["sustained"])
+        self.assertTrue(server_soak.polling_activity_summary(rows[:7], 0.8)["sustained"])
+
+    def test_http_success_with_stale_registration_is_not_a_valid_poll(self):
+        for body in ({"poll_status": "stale_worker_registration"}, {"reason": "worker_not_registered"}, {"poll_status": "no_workflow_capability"}, None):
+            self.assertFalse(server_soak.poll_response_valid(body))
+        self.assertTrue(server_soak.poll_response_valid({"task": None, "poll_status": "empty"}))
+
+    def test_each_polling_thread_gets_a_registration_across_the_configured_dimensions(self):
+        with patch.object(server_soak, "http_json", return_value=(201, {})) as register:
+            workers = server_soak.register_workers("http://fixture", "fixture", [f"ns-{i}" for i in range(8)], [f"queue-{i}" for i in range(16)], 24)
+        self.assertEqual(24, register.call_count)
+        self.assertEqual(24, len({worker[2] for worker in workers}))
+        self.assertEqual(8, len({worker[0] for worker in workers}))
+        self.assertEqual(16, len({worker[1] for worker in workers}))
+
     def standard_output(self, completed=True, elapsed=60):
         return "\n".join(json.dumps(row) for row in [
             {"phase": "started", "sdk": "test", "php": "test"},

From 882628fad8edb947f934fa0924ad726a86b7747a Mon Sep 17 00:00:00 2001
From: Durable Workflow <support@durable-workflow.com>
Date: Wed, 9 Sep 2026 14:13:32 +0000
Subject: [PATCH 3/3] Keep idle drain out of soak load measurements

---
 scripts/perf/server_soak.py             |  6 +++---
 scripts/perf/standard_workflow_soak.php |  2 +-
 tests/Unit/Support/server_soak_test.py  | 10 ++++++++++
 3 files changed, 14 insertions(+), 4 deletions(-)

diff --git a/scripts/perf/server_soak.py b/scripts/perf/server_soak.py
index e92eabf2..e62b088f 100755
--- a/scripts/perf/server_soak.py
+++ b/scripts/perf/server_soak.py
@@ -1148,7 +1148,7 @@ def memory_slope_mb_hour(samples: list[dict[str, Any]]) -> float | None:
     points = [
         (float(row["timestamp"]), float(row.get("server_memory_bytes") or 0) / (1024 * 1024))
         for row in samples
-        if row.get("server_memory_bytes")
+        if row.get("phase") != "final" and row.get("server_memory_bytes")
     ]
     if len(points) < 4:
         return None
@@ -1352,7 +1352,7 @@ def resource_summary(samples: list[dict[str, Any]]) -> dict[str, Any]:
     result = {}
     for service in ("server", "worker", "scheduler", "mysql", "redis", "soak-sdk"):
         memory = [int(row[f"{service}_memory_bytes"]) / 1048576 for row in samples if f"{service}_memory_bytes" in row]
-        cpu = [float(row[f"{service}_cpu_percent"]) for row in samples if row.get(f"{service}_cpu_percent") is not None]
+        cpu = [float(row[f"{service}_cpu_percent"]) for row in samples if row.get("phase") != "final" and row.get(f"{service}_cpu_percent") is not None]
         if memory:
             result[service] = {
                 "peak_memory_mib": round(max(memory), 2),
@@ -1544,7 +1544,7 @@ def main() -> int:
 
         emit_progress(f"draining for {max(0, args.drain_seconds)}s before final sample")
         time.sleep(max(0, args.drain_seconds))
-        final_sample = sample(args.compose_project, include_sdk=False)
+        final_sample = sample(args.compose_project, include_sdk=False) | {"phase": "final"}
         samples.append(final_sample)
         metrics.update_sample(final_sample)
         write_jsonl(samples_path, final_sample | {"phase": "final"})
diff --git a/scripts/perf/standard_workflow_soak.php b/scripts/perf/standard_workflow_soak.php
index 274a3bc2..50d7656b 100644
--- a/scripts/perf/standard_workflow_soak.php
+++ b/scripts/perf/standard_workflow_soak.php
@@ -50,7 +50,7 @@
 // Closed-loop canary: never accumulate unbounded work when the runtime slows.
 while (hrtime(true) < $stopAt) {
     if (hrtime(true) < $nextStart) {
-        usleep((int) min(200000, ($nextStart - hrtime(true)) / 1000));
+        usleep((int) max(0, min(200000, ($nextStart - hrtime(true)) / 1000)));
 
         continue;
     }
diff --git a/tests/Unit/Support/server_soak_test.py b/tests/Unit/Support/server_soak_test.py
index 4d96bc85..23f10889 100644
--- a/tests/Unit/Support/server_soak_test.py
+++ b/tests/Unit/Support/server_soak_test.py
@@ -165,6 +165,16 @@ def test_resource_summary_does_not_substitute_zero_for_missing_cpu(self):
         result = server_soak.resource_summary([{"server_memory_bytes": 1048576, "server_cpu_percent": 0}])
         self.assertEqual(0, result["server"]["cpu_mean_percent"])
 
+    def test_idle_drain_does_not_dilute_load_memory_slope_or_cpu(self):
+        rows = [{"timestamp": i * 60, "server_memory_bytes": (100 + i) * 1048576,
+                 "server_cpu_percent": 50} for i in range(10)]
+        rows.append({"phase": "final", "timestamp": 900, "server_memory_bytes": 1048576,
+                     "server_cpu_percent": 0})
+        self.assertAlmostEqual(60, server_soak.memory_slope_mb_hour(rows))
+        result = server_soak.resource_summary(rows)["server"]
+        self.assertEqual(50, result["cpu_mean_percent"])
+        self.assertEqual(1, result["final_memory_mib"])
+
     def test_summary_exposes_missing_standard_workflow_coverage(self):
         report = server_soak.render_summary({"failures": ["partial"]})
         self.assertIn("FAIL: partial", report)