ci: gate concurrent retention and logging throughput

docs/performance.md

@@ -86,7 +86,7 @@ and [macOS validation-only run](https://github.com/basefoundry/base-cli/actions/
 A sustained slowdown over 50 ms still fails; p95 over 125 ms also fails.
 Windows, WSL, parser, import, and non-persistence limits are unchanged.
 
-Each report is versioned as `base-cli.benchmark` schema version 1 and contains
+Each report is versioned as `base-cli.benchmark` schema version 2 and contains
 the package version, source revision, UTC timestamp, platform profile, Python
 version/ABI, OS release, architecture, CPU count, sample count, medians, p95,
 maximum, median absolute deviation, and parser/lifecycle comparison values.
@@ -151,3 +151,48 @@ default policy uses the byte-policy recursive-walk bounds described above. A
 consumer that needs only count/age retention can explicitly omit the byte cap.
 The concurrent benchmark in #391 measures twelve processes against one warmed
 cache and gates the p95-to-serial ratio.
+
+### Benchmark report v2: contention and logging
+
+`base-cli.benchmark` schema version 2 adds `results.base-cli.stress` and
+`stress_budgets` with explicit units. Twelve synchronized subprocesses share a
+cache warmed past the default 20-bundle cap. Three batches report 36 invocation
+samples, serial/concurrent p95 milliseconds, and their p95 ratio; process import
+and the start barrier are outside the invocation timer. Logging measures 3,000
+INFO records per sample through a real lifecycle, both with and without persistent
+files, reporting microseconds/record and records/second.
+
+| Profile | Concurrent / serial p95 cap | Log p95 microseconds/record cap |
+| --- | ---: | ---: |
+| unix | 6 | 40 |
+| macos | 6 | 60 |
+| windows | 10 | 150 |
+| wsl | 10 | 100 |
+
+These initial hosted caps allow scheduling/filesystem variation while detecting
+material regressions. `--check` rejects missing, nonfinite, and over-budget stress
+metrics. Reports retain the same CI artifact name and 90-day retention with the
+v2 schema marker; consumers must branch on that marker. Local and first hosted
+measurements are retained with this PR before further tightening of the caps.
+
+Development-host calibration (macOS, Python 3.14.6, 31 serial/log samples and
+36 concurrent samples): concurrent/serial p95 ratio 2.42; ephemeral logging p95
+6.20 microseconds/record; persistent logging p95 14.15 microseconds/record.
+These are local measurements; hosted per-profile results are retained separately.
+
+The [first hosted stress run](https://github.com/basefoundry/base-cli/actions/runs/37054920383)
+recorded the following 31-sample logging and 36-sample concurrency results:
+
+| Profile | Concurrent / serial p95 | Ephemeral log p95 (us/record) | Persistent log p95 (us/record) |
+| --- | ---: | ---: | ---: |
+| macos (3-core arm64, Python 3.13) | 4.49 | 16.01 | 46.42 |
+| windows | 2.99 | 20.14 | 49.03 |
+| wsl | 4.25 | 11.44 | 22.90 |
+
+The initial macOS 40 us/record p95 estimate rejected a run whose persistent median
+was 25.82 us/record. Its hosted cap is therefore calibrated to 60 us/record; the
+other logging and concurrency caps are unchanged. This remains below the original
+107 us/record development-host regression, while retaining room for hosted tails.
+The corresponding `base-cli-benchmark-{profile}-37054920383` artifacts contain
+machine metadata and all measured summaries. Unix calibration remains subject to
+its hosted check before merge.

scripts/benchmark_runtime.py

@@ -7,6 +7,7 @@
 import datetime as dt
 import importlib.util
 import json
+import math
 import os
 import platform
 import statistics
@@ -56,7 +57,9 @@
 DEFAULT_ITERATIONS = 31
 FRAMEWORKS = ("base-cli", "click", "typer", "cyclopts")
 RESULT_SCHEMA = "base-cli.benchmark"
-RESULT_SCHEMA_VERSION = 1
+RESULT_SCHEMA_VERSION = 2
+CONCURRENCY_RATIO_BUDGETS = {"unix": 6.0, "macos": 6.0, "windows": 10.0, "wsl": 10.0}
+LOG_P95_BUDGETS_US = {"unix": 40.0, "macos": 60.0, "windows": 150.0, "wsl": 100.0}
 
 
 class Summary(TypedDict):
@@ -128,6 +131,7 @@ def main() -> int:
             _measure_production_invocations(args.iterations)
         )
         results["base-cli"]["features"] = cast(Any, _measure_base_cli_features(args.iterations))
+        results["base-cli"]["stress"] = _measure_stress(args.iterations)
 
     report = {
         "schema": RESULT_SCHEMA,
@@ -142,10 +146,21 @@ def main() -> int:
         "results": results,
         "comparisons": _comparisons(results),
         "budgets_ms": _budgets_for_platform(BENCHMARK_PLATFORM),
+        "stress_budgets": {
+            "concurrent_to_serial_p95_ratio": CONCURRENCY_RATIO_BUDGETS[BENCHMARK_PLATFORM],
+            "log_p95_us_per_record": LOG_P95_BUDGETS_US[BENCHMARK_PLATFORM],
+        },
     }
 
     failures = _check_results(results) if args.check else []
     _write_github_summary(report)
+    if os.environ.get("GITHUB_STEP_SUMMARY"):
+        with open(os.environ["GITHUB_STEP_SUMMARY"], "a", encoding="utf-8") as summary:
+            summary.write(
+                "\n### Concurrent retention and logging\n\n```json\n"
+                + json.dumps(results["base-cli"].get("stress", {}), indent=2)
+                + "\n```\n"
+            )
     if args.json:
         print(json.dumps(report, indent=2, sort_keys=True))
     else:
@@ -331,6 +346,18 @@ def _check_results(results: dict[str, FrameworkMetrics]) -> list[str]:
             feature_budget = _feature_budget_for_platform(name, BENCHMARK_PLATFORM)
             if p95 is not None and p95 > feature_budget:
                 failures.append(f"base-cli {name} p95 exceeded {feature_budget:.0f} ms")
+    stress = base.get("stress", {})
+    ratio = stress.get("concurrent_to_serial_p95_ratio")
+    if (
+        not isinstance(ratio, (int, float))
+        or not math.isfinite(ratio)
+        or ratio > CONCURRENCY_RATIO_BUDGETS[BENCHMARK_PLATFORM]
+    ):
+        failures.append("concurrent-to-serial p95 ratio is missing, invalid, or exceeds budget")
+    for name in ("logging_persistent_us_per_record", "logging_ephemeral_us_per_record"):
+        p95 = _metric_p95(stress, name)
+        if p95 is None or not math.isfinite(p95) or p95 > LOG_P95_BUDGETS_US[BENCHMARK_PLATFORM]:
+            failures.append(f"{name} p95 is missing, invalid, or exceeds budget")
     if BENCHMARK_PLATFORM in {"unix", "macos"} and isinstance(features, dict):
         persistence = features.get("persistence_enabled_ms", {})
         median = persistence.get("median") if isinstance(persistence, dict) else None
@@ -702,6 +729,108 @@ def log_with_file(ctx: Any) -> None:
     return results
 
 
+def _measure_stress(iterations: int) -> dict[str, Any]:
+    from contextlib import redirect_stderr
+
+    import base_cli
+    from base_cli.testing import invoke
+
+    worker_code = r"""
+import os, sys, time
+from pathlib import Path
+import base_cli
+app = base_cli.App(name="benchmark-shared-retention")
+@app.command()
+def main(ctx):
+    pass
+_ = app.click_command
+if sys.argv[1] == "warm":
+    for _ in range(22):
+        assert base_cli.run_app(app, []) == 0
+    raise SystemExit(0)
+if sys.argv[1] == "parallel":
+    print("ready", flush=True)
+    deadline = time.monotonic() + 30
+    while not Path(sys.argv[2]).exists():
+        if time.monotonic() > deadline:
+            raise RuntimeError("benchmark start barrier timed out")
+        time.sleep(0.001)
+started = time.perf_counter_ns()
+status = base_cli.run_app(app, [])
+assert status == 0
+print((time.perf_counter_ns() - started) / 1_000_000, flush=True)
+"""
+    workers = 12
+    with tempfile.TemporaryDirectory(prefix="base-cli-stress-") as temporary:
+        root = Path(temporary)
+        env = {**os.environ, "BASE_CLI_CACHE_DIR": str(root / "cache")}
+        command = [sys.executable, "-c", worker_code]
+        subprocess.run([*command, "warm"], env=env, check=True, capture_output=True, timeout=60)
+        serial = []
+        for _ in range(iterations):
+            result = subprocess.run(
+                [*command, "serial"], env=env, check=True, capture_output=True, text=True, timeout=30
+            )
+            serial.append(float(result.stdout.strip()))
+        concurrent = []
+        for batch in range(3):
+            barrier = root / f"start-{batch}"
+            processes = [
+                subprocess.Popen(
+                    [*command, "parallel", str(barrier)],
+                    env=env,
+                    stdout=subprocess.PIPE,
+                    stderr=subprocess.PIPE,
+                    text=True,
+                )
+                for _ in range(workers)
+            ]
+            try:
+                for process in processes:
+                    assert process.stdout is not None
+                    if process.stdout.readline().strip() != "ready":
+                        raise RuntimeError("concurrent benchmark worker failed before barrier")
+                barrier.touch()
+                for process in processes:
+                    stdout, stderr = process.communicate(timeout=30)
+                    if process.returncode:
+                        raise RuntimeError(f"concurrent benchmark failed: {stderr}")
+                    concurrent.append(float(stdout.strip()))
+            finally:
+                for process in processes:
+                    if process.poll() is None:
+                        process.kill()
+                    process.communicate()
+        result_metrics: dict[str, Any] = {
+            "workers": workers,
+            "concurrent_samples": len(concurrent),
+            "serial_ms": _summary(serial),
+            "concurrent_ms": _summary(concurrent),
+            "concurrent_to_serial_p95_ratio": _summary(concurrent)["p95"] / _summary(serial)["p95"],
+            "logging_records_per_sample": 3000,
+        }
+        for persistent, label in ((False, "ephemeral"), (True, "persistent")):
+            samples = []
+            app = base_cli.App(name=f"benchmark-log-{label}", log_to_file=persistent)
+
+            @app.command()
+            def log_records(ctx: Any, _samples: list[float] = samples) -> None:
+                ctx.log.info("warm source cache")
+                started = time.perf_counter_ns()
+                for index in range(3000):
+                    ctx.log.info("record %s", index)
+                _samples.append((time.perf_counter_ns() - started) / 3000 / 1000)
+
+            with open(os.devnull, "w", encoding="utf-8") as quiet, redirect_stderr(quiet):
+                for _ in range(iterations):
+                    result = invoke(app, [], home=root / label)
+                    if result.exit_code:
+                        raise RuntimeError(f"logging benchmark failed: {result.output}")
+            result_metrics[f"logging_{label}_us_per_record"] = _summary(samples)
+            result_metrics[f"logging_{label}_records_per_second"] = _summary([1_000_000 / sample for sample in samples])
+        return result_metrics
+
+
 def _measure_runner(iterations: int, callback: Callable[[], Any]) -> list[float]:
     samples: list[float] = []
     for _ in range(iterations):

tests/test_benchmark_runtime.py

@@ -159,6 +159,26 @@ def test_github_summary_separates_lifecycle_overhead_from_parser(self) -> None:
         self.assertIn("base-cli lifecycle increment over Click warm p95", summary)
         self.assertIn("Base CLI feature scenarios", summary)
 
+    def test_stress_regressions_fail_every_platform_profile(self) -> None:
+        for profile in ("unix", "macos", "windows", "wsl"):
+            for metric in (
+                "concurrent_to_serial_p95_ratio",
+                "logging_persistent_us_per_record",
+                "logging_ephemeral_us_per_record",
+            ):
+                with self.subTest(profile=profile, metric=metric):
+                    metrics = self._complete_results()
+                    stress = metrics["base-cli"]["stress"]
+                    stress[metric] = 1000.0 if metric.endswith("ratio") else self._summary(1000.0)
+                    with mock.patch.object(benchmark_runtime, "BENCHMARK_PLATFORM", profile):
+                        failures = benchmark_runtime._check_results(metrics)
+                    self.assertTrue(any("exceeds budget" in failure for failure in failures))
+
+    def test_stress_missing_and_nonfinite_samples_fail(self) -> None:
+        metrics = self._complete_results()
+        metrics["base-cli"]["stress"] = {"concurrent_to_serial_p95_ratio": float("nan")}
+        self.assertEqual(sum("missing, invalid" in failure for failure in benchmark_runtime._check_results(metrics)), 3)
+
     @staticmethod
     def _summary(p95: float) -> dict[str, float]:
         return {
@@ -187,6 +207,11 @@ def _complete_results(
             {
                 "lifecycle_warm_invocation_ms": cls._summary(lifecycle_p95),
                 "production_warm_invocation_ms": cls._summary(4.0),
+                "stress": {
+                    "concurrent_to_serial_p95_ratio": 2.0,
+                    "logging_persistent_us_per_record": cls._summary(10.0),
+                    "logging_ephemeral_us_per_record": cls._summary(5.0),
+                },
                 "features": {
                     "lifecycle_noop_ms": cls._summary(1.0),
                     "json_success_ms": cls._summary(1.0),