Add adversarial regression and performance suites

.github/workflows/tests.yml

@@ -62,6 +62,7 @@ jobs:
         run: |
           python -m mypy --strict examples/typed_consumer.py
           python scripts/validate_docs.py
+          python scripts/benchmark_runtime.py --check
           python -m compileall -q examples
       - name: Run tests with coverage threshold
         run: python -m pytest --cov=base_cli --cov-report=term-missing --cov-fail-under=80

docs/performance.md

@@ -0,0 +1,33 @@
+# Performance and adversarial-regression contract
+
+`base-cli` treats startup and filesystem behavior as part of its public
+quality contract. The checked benchmark is intentionally small and runs from
+the source checkout:
+
+```bash
+python scripts/benchmark_runtime.py --check
+```
+
+It records fresh-process import time and the cost of an isolated production
+invocation through `base_cli.testing.invoke`. The CI quality job checks the
+sample p95 against these budgets:
+
+| Measurement | Budget |
+| --- | ---: |
+| Fresh `import base_cli` | 750 ms |
+| Isolated invocation and runtime filesystem setup | 1,500 ms |
+
+The benchmark reports the median, p95, and maximum for seven samples. These
+budgets are intentionally broad enough for hosted runners while still
+detecting accidental quadratic startup work, unbounded metadata scans, or
+unexpected dependency imports. A performance improvement should preserve the
+same lifecycle and persistence assertions covered by the adversarial tests.
+
+The regression suite uses deterministic Hypothesis examples (`derandomize`
+enabled), fixed multiprocessing workloads, and explicit seed values in every
+worker payload. Property cases cover redaction and command-protocol framing;
+spawned processes cover history append, private metadata replacement, logging,
+extension discovery caches, and run-bundle retention. Ctrl+C is tested through
+both the lifecycle boundary and a real POSIX subprocess signal. Windows keeps
+the portable lifecycle and persistence checks while skipping only assertions
+that require POSIX signal or descriptor semantics.

docs/releasing.md

@@ -23,7 +23,10 @@ source tree on `sys.path`. Tests run across Python 3.10 through 3.14 on Linux,
 macOS, and Windows, with Debian, Fedora, and WSL validation retained. Blocking
 quality gates cover Ruff formatting/lint, strict public-sample typing, an 80%
 branch-coverage threshold, documentation/example checks, and dependency/static
-security scans.
+security scans. The [performance contract](performance.md) also checks fresh
+import and isolated invocation budgets, while the adversarial suite exercises
+redaction, protocol framing, persistence, concurrency, retention, and signal
+cleanup.
 
 The publish job downloads that same reviewed artifact; it does not rebuild
 during publication.

lib/python/base_cli/_private_files.py

@@ -5,6 +5,7 @@
 import json
 import os
 import secrets
+import time
 from collections.abc import Mapping
 from pathlib import Path
 from typing import Any
@@ -98,7 +99,7 @@ def write_private_json(path: Path, value: Mapping[str, Any]) -> None:
         else:
             if path.is_symlink():
                 raise OSError(f"refusing to replace symlink '{path}'")
-            os.replace(path.parent / temporary_name, path)
+            _replace_with_retry(path.parent / temporary_name, path)
         temporary_name = None
         if parent_fd is not None:
             _sync_directory(parent_fd)
@@ -177,3 +178,17 @@ def _sync_directory(parent_fd: int) -> None:
         # Directory fsync is not available on all supported filesystems and
         # platforms.  The file itself was still flushed before replacement.
         pass
+
+
+def _replace_with_retry(source: Path, destination: Path) -> None:
+    """Replace a private file, tolerating transient Windows sharing races."""
+
+    attempts = 1 if os.name != "nt" else 10
+    for attempt in range(attempts):
+        try:
+            os.replace(source, destination)
+            return
+        except PermissionError:
+            if attempt == attempts - 1:
+                raise
+            time.sleep(0.001 * (attempt + 1))

lib/python/base_cli/logging.py

@@ -6,7 +6,17 @@
 import sys
 import time
 from pathlib import Path
-from typing import TextIO
+from typing import BinaryIO, TextIO
+
+try:
+    import fcntl as _fcntl
+except ImportError:  # pragma: no cover - fcntl is unavailable on Windows.
+    _fcntl = None  # type: ignore[assignment]
+
+try:
+    import msvcrt as _msvcrt
+except ImportError:  # pragma: no cover - msvcrt is unavailable outside Windows.
+    _msvcrt = None  # type: ignore[assignment]
 
 from ._private_files import restrict_file
 from .context import get_current_context
@@ -107,6 +117,10 @@ def secure_log_file_permissions(log_file: Path) -> None:
 
 
 class SecureLogFileHandler(logging.FileHandler):
+    def __init__(self, filename: str | os.PathLike[str], *args: object, **kwargs: object) -> None:
+        self._lock_path = Path(filename).with_name(f".{Path(filename).name}.lock")
+        super().__init__(filename, *args, **kwargs)  # type: ignore[arg-type]
+
     def _open(self) -> TextIO:
         fd = os.open(self.baseFilename, _secure_log_file_open_flags(self.mode), 0o600)
         try:
@@ -118,6 +132,47 @@ def _open(self) -> TextIO:
             os.close(fd)
             raise
 
+    def emit(self, record: logging.LogRecord) -> None:
+        lock_stream = _open_log_lock(self._lock_path)
+        try:
+            _lock_log_stream(lock_stream)
+            super().emit(record)
+        finally:
+            _unlock_log_stream(lock_stream)
+            lock_stream.close()
+
+
+def _open_log_lock(path: Path) -> BinaryIO:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    stream = path.open("a+b")
+    try:
+        if stream.seek(0, os.SEEK_END) == 0:
+            stream.write(b"0")
+            stream.flush()
+        restrict_file(path)
+        return stream
+    except BaseException:
+        stream.close()
+        raise
+
+
+def _lock_log_stream(stream: BinaryIO) -> None:
+    fd = stream.fileno()
+    if _fcntl is not None:
+        _fcntl.flock(fd, _fcntl.LOCK_EX)
+    elif _msvcrt is not None:
+        stream.seek(0)
+        _msvcrt.locking(fd, _msvcrt.LK_LOCK, 1)
+
+
+def _unlock_log_stream(stream: BinaryIO) -> None:
+    fd = stream.fileno()
+    if _fcntl is not None:
+        _fcntl.flock(fd, _fcntl.LOCK_UN)
+    elif _msvcrt is not None:
+        stream.seek(0)
+        _msvcrt.locking(fd, _msvcrt.LK_UNLCK, 1)
+
 
 def _secure_log_file_open_flags(mode: str) -> int:
     flags = os.O_CREAT

pyproject.toml

@@ -38,6 +38,7 @@ dependencies = [
 [project.optional-dependencies]
 dev = [
   "build>=1.2",
+  "hypothesis>=6.100,<7",
   "mypy>=1.17,<2",
   "pytest>=8.0",
 ]

scripts/benchmark_runtime.py

@@ -0,0 +1,114 @@
+#!/usr/bin/env python3
+"""Track import and isolated invocation costs with stable, checked budgets."""
+
+from __future__ import annotations
+
+import argparse
+import os
+import statistics
+import subprocess
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+
+IMPORT_P95_BUDGET_MS = 750.0
+INVOCATION_P95_BUDGET_MS = 1_500.0
+DEFAULT_ITERATIONS = 7
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--iterations",
+        type=int,
+        default=DEFAULT_ITERATIONS,
+        help=f"number of samples per benchmark (default: {DEFAULT_ITERATIONS})",
+    )
+    parser.add_argument(
+        "--check",
+        action="store_true",
+        help="fail when the documented p95 budgets are exceeded",
+    )
+    args = parser.parse_args()
+    if args.iterations < 3:
+        parser.error("--iterations must be at least 3")
+
+    import_samples = _measure_import(args.iterations)
+    invocation_samples = _measure_invocations(args.iterations)
+    metrics = {
+        "import_ms": _summary(import_samples),
+        "invocation_ms": _summary(invocation_samples),
+    }
+    print("import_ms: median={median:.2f} p95={p95:.2f} max={maximum:.2f}".format(**metrics["import_ms"]))
+    print("invocation_ms: median={median:.2f} p95={p95:.2f} max={maximum:.2f}".format(**metrics["invocation_ms"]))
+
+    if not args.check:
+        return 0
+    failures = []
+    if metrics["import_ms"]["p95"] > IMPORT_P95_BUDGET_MS:
+        failures.append(f"import p95 exceeded {IMPORT_P95_BUDGET_MS:.0f} ms")
+    if metrics["invocation_ms"]["p95"] > INVOCATION_P95_BUDGET_MS:
+        failures.append(f"invocation p95 exceeded {INVOCATION_P95_BUDGET_MS:.0f} ms")
+    if failures:
+        print("Performance budget failure: " + "; ".join(failures), file=sys.stderr)
+        return 1
+    return 0
+
+
+def _measure_import(iterations: int) -> list[float]:
+    package_root = Path(__file__).resolve().parents[1] / "lib" / "python"
+    environment = dict(os.environ)
+    existing_path = environment.get("PYTHONPATH")
+    environment["PYTHONPATH"] = f"{package_root}{os.pathsep}{existing_path}" if existing_path else str(package_root)
+    samples: list[float] = []
+    for _ in range(iterations):
+        started = time.perf_counter_ns()
+        subprocess.run(
+            [sys.executable, "-c", "import base_cli"],
+            check=True,
+            env=environment,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.PIPE,
+        )
+        samples.append(_elapsed_ms(started))
+    return samples
+
+
+def _measure_invocations(iterations: int) -> list[float]:
+    import base_cli
+    from base_cli.testing import invoke
+
+    app = base_cli.App(name="benchmark-runtime")
+
+    @app.command()
+    def main(ctx: base_cli.Context) -> None:
+        del ctx
+
+    samples: list[float] = []
+    with tempfile.TemporaryDirectory(prefix="base-cli-benchmark-") as tmpdir:
+        home = Path(tmpdir)
+        for _ in range(iterations):
+            started = time.perf_counter_ns()
+            result = invoke(app, [], home=home)
+            if result.exit_code != 0:
+                raise RuntimeError(f"benchmark invocation failed: {result.output}")
+            samples.append(_elapsed_ms(started))
+    return samples
+
+
+def _elapsed_ms(started_ns: int) -> float:
+    return (time.perf_counter_ns() - started_ns) / 1_000_000
+
+
+def _summary(samples: list[float]) -> dict[str, float]:
+    return {
+        "median": statistics.median(samples),
+        "p95": max(samples),
+        "maximum": max(samples),
+    }
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())