akadmin/headroom
1
0
Fork 0
mirror of https://github.com/headroomlabs-ai/headroom.git synced 2026-08-27 14:17:10 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix: harden persistent install startup (#1851)

Browse source 
## Description

Hardens persistent install startup and proxy compression behavior for
issue #1843. Repeated `headroom install start` / scheduled ensure calls
no longer spawn duplicate runtimes by default, and `/v1/compress` now
fails open on compression timeout instead of returning a 503. The PR
also adds a machine-readable platform feature matrix and app-level
stabilization tests for health, compression functionality, timeout
behavior, and matrix evidence.

Refs #1843

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [x] Documentation update
- [x] Performance improvement
- [ ] Code refactoring (no functional changes)

## Changes Made

- Wrapped direct persistent deployment starts with the existing
profile-local runtime start lock.
- Made `headroom install start` idempotent when the deployment is
already healthy.
- Added wedged-runtime handling: if a PID is running but `/readyz` does
not recover inside the grace window, stop it before starting again.
- Kept `install agent ensure` inside the already-held lock while
delegating to the shared start helper.
- Changed `/v1/compress` timeout behavior from `503 compression_timeout`
to fail-open `200` with original messages, `compression_skipped: true`,
and `skip_reason: compression_timeout`.
- Added `tests/test_platform_stabilization_functional.py` covering real
FastAPI health/compression routes, successful compression metrics,
timeout fail-open speed, and a real JSON tool payload that reduces
tokens.
- Added `docs/platform-feature-matrix.json` and
`docs/platform-stabilization.md` for Linux/macOS/Windows hardening
coverage and known gaps.
- Strengthened matrix tests so cited local test/workflow paths must
exist.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .`)
- [x] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality
- [x] Manual testing performed

### Test Output

```text
# Clean source tree without copied Rust extension: functional module is skipped locally, as CI copies _core from the built wheel.
> python -m pytest tests/test_install tests/test_cli/test_install_cli.py tests/test_platform_feature_matrix.py tests/test_platform_stabilization_functional.py -q
collected 120 items / 1 skipped
119 passed, 2 skipped in 17.08s

> python -m ruff check headroom/proxy/handlers/openai.py tests/test_platform_stabilization_functional.py tests/test_platform_feature_matrix.py
All checks passed!

# Local Windows compiled-core proof:
> python -m maturin build --profile ci --out dist-local
Built wheel for abi3 Python >= 3.10 to dist-local\headroom_ai-0.29.0-cp310-abi3-win_amd64.whl

# Copied _core.pyd from the wheel into headroom/ for local route execution, then:
> python -m pytest tests/test_platform_stabilization_functional.py -q
collected 4 items
4 passed in 6.71s

> python -m pytest tests/test_install tests/test_cli/test_install_cli.py tests/test_platform_feature_matrix.py -q
collected 120 items
119 passed, 1 skipped in 17.16s

Commit hooks:
Sync plugin versions.....................................................Passed
check for merge conflicts................................................Passed
ruff.....................................................................Passed
ruff-format..............................................................Passed
mypy.....................................................................Passed
```

## Real Behavior Proof

- Environment: Windows 11, PowerShell, Python 3.13.13, worktree
`C:\git\headroom-stabilization` on branch
`jd/cross-platform-stabilization`.
- Exact command / steps: built the Windows wheel with `maturin`,
extracted `_core.pyd`, ran the new FastAPI route tests and
install/matrix tests listed above, then removed generated artifacts
before committing.
- Observed result: direct start paths now no-op when healthy, skip
spawning when the start lock is contended, and stop a wedged runtime
before restart. `/v1/compress` now returns original messages quickly on
timeout instead of a 503. The real JSON tool-payload smoke test returns
`tokens_before > tokens_after`, `tokens_saved > 0`, `compression_ratio <
1.0`, and non-empty transforms through the public route.
- Not tested: full native Windows persistent process e2e remains blocked
by the upstream CRT/wheel issue already documented in workflows and in
the matrix. No real OS service was installed locally; service manager
behavior is covered by argument-level unit tests.

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

## Checklist

- [x] My code follows the project's style guidelines
- [x] I have performed a self-review of my code
- [x] I have commented my code, particularly in hard-to-understand areas
- [x] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [ ] I have updated the CHANGELOG.md if applicable

## Screenshots (if applicable)

N/A

## Additional Notes

CHANGELOG is not updated because this is an unreleased
hardening/test/documentation pass. The platform matrix intentionally
records partial/blocked Windows/macOS e2e gaps instead of claiming full
coverage where the repo cannot currently run it.
This commit is contained in:
JD Davis 2026-07-10 04:40:34 +00:00 • committed by GitHub
parent 28ca61fc9d
commit 1d2b76e72e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 747 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

329
docs/platform-feature-matrix.json Normal file
View file

@ -0,0 +1,329 @@
{
"schema_version": 1,
"updated": "2026-07-06",
"source_issues": [
"https://github.com/headroomlabs-ai/headroom/issues/1843"
],
"status_values": [
"covered",
"partial",
"gap",
"blocked"
],
"platforms": [
"linux",
"macos",
"windows"
],
"features": [
{
"id": "install_apply_python",
"name": "Persistent Python install applies and starts",
"risk": "install",
"platforms": {
"linux": {
"status": "covered",
"tests": [
"tests/test_cli/test_install_cli.py",
"tests/test_install/test_supervisors.py",
".github/workflows/install-native-e2e.yml"
]
},
"macos": {
"status": "covered",
"tests": [
"tests/test_install/test_supervisors.py",
".github/workflows/install-native-e2e.yml"
]
},
"windows": {
"status": "partial",
"tests": [
"tests/test_install/test_supervisors.py",
"tests/test_install/test_runtime.py",
".github/workflows/ci.yml#windows-native-wrapper"
],
"gap": "Native install smoke workflow is source-level until Windows wheel CRT conflicts are resolved."
}
}
},
{
"id": "install_windows_service",
"name": "Windows service install uses sc.exe safely",
"risk": "install",
"platforms": {
"linux": {
"status": "covered",
"tests": [
"tests/test_install/test_supervisors.py"
]
},
"macos": {
"status": "covered",
"tests": [
"tests/test_install/test_supervisors.py"
]
},
"windows": {
"status": "covered",
"tests": [
"tests/test_install/test_supervisors.py"
]
}
}
},
{
"id": "single_instance_start",
"name": "Persistent runtime start is single-instance by default",
"risk": "runtime",
"platforms": {
"linux": {
"status": "covered",
"tests": [
"tests/test_cli/test_install_cli.py",
"tests/test_install/test_runtime.py"
]
},
"macos": {
"status": "covered",
"tests": [
"tests/test_cli/test_install_cli.py",
"tests/test_install/test_runtime.py"
]
},
"windows": {
"status": "covered",
"tests": [
"tests/test_cli/test_install_cli.py",
"tests/test_install/test_runtime.py"
]
}
}
},
{
"id": "compression_fail_open",
"name": "Slow or saturated compression fails open",
"risk": "performance",
"platforms": {
"linux": {
"status": "covered",
"tests": [
"tests/test_platform_stabilization_functional.py",
"tests/test_kompress_request_nonblocking.py",
"tests/test_proxy_compression_executor.py",
"tests/test_openai_codex_ws_lifecycle.py"
]
},
"macos": {
"status": "partial",
"tests": [
"tests/test_platform_stabilization_functional.py",
"tests/test_kompress_request_nonblocking.py",
"tests/test_proxy_compression_executor.py"
],
"gap": "Native workflow coverage should add focused performance-gate tests without full model downloads."
},
"windows": {
"status": "partial",
"tests": [
"tests/test_platform_stabilization_functional.py",
"tests/test_kompress_request_nonblocking.py",
"tests/test_proxy_compression_executor.py"
],
"gap": "Source-level tests validate fail-open semantics; full native proxy e2e waits on Windows wheel availability."
}
}
},
{
"id": "proxy_functional_smoke",
"name": "Proxy health and /v1/compress work through the app surface",
"risk": "proxy",
"platforms": {
"linux": {
"status": "covered",
"tests": [
"tests/test_platform_stabilization_functional.py",
"tests/test_ccr_row_drop_store_bridge.py"
]
},
"macos": {
"status": "partial",
"tests": [
"tests/test_platform_stabilization_functional.py",
"tests/test_ccr_row_drop_store_bridge.py"
],
"gap": "FastAPI route coverage exists; native persistent proxy process smoke should be added."
},
"windows": {
"status": "partial",
"tests": [
"tests/test_platform_stabilization_functional.py",
"tests/test_ccr_row_drop_store_bridge.py"
],
"gap": "FastAPI route coverage exists; native persistent proxy process smoke waits on Windows wheel availability."
}
}
},
{
"id": "ccr_persistence",
"name": "CCR survives restart when persistent storage is enabled",
"risk": "cache",
"platforms": {
"linux": {
"status": "covered",
"tests": [
"crates/headroom-core/tests/ccr_backends.rs",
"tests/test_storage_backends.py"
]
},
"macos": {
"status": "partial",
"tests": [
"crates/headroom-core/tests/ccr_backends.rs",
"tests/test_storage_backends.py"
],
"gap": "Rust backend tests run in the Rust workflow; native macOS restart e2e is not yet present."
},
"windows": {
"status": "partial",
"tests": [
"crates/headroom-core/tests/ccr_backends.rs",
"tests/test_storage_backends.py"
],
"gap": "Windows restart e2e is blocked by the native wheel CRT conflict."
}
}
},
{
"id": "init_cli",
"name": "headroom init configures supported agents",
"risk": "install",
"platforms": {
"linux": {
"status": "covered",
"tests": [
".github/workflows/init-native-e2e.yml"
]
},
"macos": {
"status": "covered",
"tests": [
".github/workflows/init-native-e2e.yml"
]
},
"windows": {
"status": "blocked",
"tests": [
".github/workflows/init-native-e2e.yml"
],
"gap": "Matrix entry is intentionally excluded until Windows wheel CRT conflicts are resolved."
}
}
},
{
"id": "wrap_prepare_only",
"name": "headroom wrap prepare-only mutates config safely",
"risk": "install",
"platforms": {
"linux": {
"status": "covered",
"tests": [
".github/workflows/wrap-native-e2e.yml"
]
},
"macos": {
"status": "covered",
"tests": [
".github/workflows/wrap-native-e2e.yml"
]
},
"windows": {
"status": "blocked",
"tests": [
".github/workflows/wrap-native-e2e.yml"
],
"gap": "Matrix entry is intentionally excluded until Windows wheel CRT conflicts are resolved."
}
}
},
{
"id": "toin_skip_recommendations",
"name": "TOIN skip-compression recommendations are exposed",
"risk": "cache",
"platforms": {
"linux": {
"status": "covered",
"tests": [
"tests/test_toin.py",
"tests/test_proxy_ccr.py",
"tests/test_compression_policy_toin_gate.py"
]
},
"macos": {
"status": "partial",
"tests": [
"tests/test_toin.py",
"tests/test_compression_policy_toin_gate.py"
],
"gap": "Native end-to-end verification of served recommendations is still needed."
},
"windows": {
"status": "partial",
"tests": [
"tests/test_toin.py",
"tests/test_compression_policy_toin_gate.py"
],
"gap": "Native end-to-end verification is blocked by Windows wheel availability."
}
}
}
],
"sanity_tests": [
{
"id": "cli_help",
"description": "Every public CLI command renders help without importing optional heavy runtimes.",
"tests": [
"tests/test_cli"
]
},
{
"id": "install_paths",
"description": "Install paths resolve under the active user profile and never require administrator paths for user scope.",
"tests": [
"tests/test_install/test_paths.py"
]
},
{
"id": "runtime_selection",
"description": "Python, Docker, service, and task runtime commands are generated deterministically for each platform.",
"tests": [
"tests/test_install/test_runtime.py",
"tests/test_install/test_supervisors.py"
]
},
{
"id": "health_startup",
"description": "Persistent starts wait for /readyz and surface startup failure instead of silently succeeding.",
"tests": [
"tests/test_cli/test_install_cli.py",
"tests/test_install/test_health.py"
]
},
{
"id": "compression_backpressure",
"description": "Compression queue saturation and model cold-start fail open without hanging request paths.",
"tests": [
"tests/test_platform_stabilization_functional.py",
"tests/test_kompress_request_nonblocking.py",
"tests/test_proxy_compression_executor.py"
]
},
{
"id": "proxy_route_smoke",
"description": "Health and /v1/compress routes return functional metrics and fail open on timeout.",
"tests": [
"tests/test_platform_stabilization_functional.py"
]
}
]
}

28
docs/platform-stabilization.md Normal file
View file

@ -0,0 +1,28 @@
# Platform Stabilization Matrix
This matrix is the hardening source of truth for install, startup, runtime, cache, and compression behavior across Linux, macOS, and Windows. The machine-readable matrix lives in `docs/platform-feature-matrix.json`; CI tests validate its shape so every feature has explicit platform status and test evidence.
## Status Values
- `covered`: unit, integration, or native e2e coverage exists for the platform.
- `partial`: coverage exists, but a named gap remains.
- `gap`: no meaningful coverage exists yet.
- `blocked`: coverage is intentionally excluded by a known external blocker.
## Current Priorities
1. Keep install/setup/run idempotent. Persistent starts must not create duplicate proxy instances unless a future explicit opt-in exists.
2. Keep compression fail-open. Cold model downloads, saturated executors, or timeout paths must pass through unchanged traffic instead of hanging agents.
3. Keep cache behavior visible. CCR persistence and TOIN skip recommendations must have restart and served-recommendation coverage.
4. Keep Windows honest. Windows-specific tests should run locally and in CI whenever they do not require the currently blocked native wheel build.
## Issue 1843 Coverage Map
- Windows service quoting: `install_windows_service`
- Duplicate startup processes: `single_instance_start`
- Slow compression and hangs: `compression_fail_open`
- CCR restart persistence: `ccr_persistence`
- TOIN recommendation wiring: `toin_skip_recommendations`
- Install/setup/run sanity: `install_apply_python`, `init_cli`, `wrap_prepare_only`
When adding or closing a hardening item, update `docs/platform-feature-matrix.json` in the same PR as the implementation or test change.

23
headroom/cli/install.py
View file

@ -59,11 +59,26 @@ def _require_manifest(profile: str) -> DeploymentManifest:
return manifest
def _start_deployment(manifest: DeploymentManifest) -> None:
def _start_deployment(manifest: DeploymentManifest, *, assume_start_lock: bool = False) -> None:
if not assume_start_lock:
with acquire_runtime_start_lock(manifest.profile) as acquired:
if not acquired:
click.echo(f"Deployment '{manifest.profile}' start is already in progress.")
return
_start_deployment(manifest, assume_start_lock=True)
return
if probe_ready(manifest.health_url):
return
if manifest.preset == InstallPreset.PERSISTENT_DOCKER.value and shutil.which("docker") is None:
raise click.ClickException(
"Docker is required for this deployment but 'docker' was not found on PATH."
)
if runtime_status(manifest) == "running":
if wait_ready(manifest, timeout_seconds=_STARTUP_READY_TIMEOUT_SECONDS):
return
stop_runtime(manifest)
try:
if manifest.preset == InstallPreset.PERSISTENT_DOCKER.value:
start_persistent_docker(manifest)
@ -77,7 +92,7 @@ def _start_deployment(manifest: DeploymentManifest) -> None:
except subprocess.CalledProcessError as e:
raise click.ClickException(
f"Cannot start deployment '{manifest.profile}': command failed "
f"({' '.join(map(str, e.cmd)) if isinstance(e.cmd, (list, tuple)) else e.cmd})"
f"({' '.join(map(str, e.cmd)) if isinstance(e.cmd, list | tuple) else e.cmd})"
) from None
if not wait_ready(manifest, timeout_seconds=45):
@ -276,7 +291,7 @@ def install_apply(
_restore_deployment(existing)
# Surface non-Click errors (OSError, CalledProcessError, …) as a clean
# message rather than a raw traceback; Click errors pass through as-is.
if isinstance(exc, (click.ClickException, click.Abort)):
if isinstance(exc, click.ClickException | click.Abort):
raise
raise click.ClickException(f"Failed to install deployment '{profile}': {exc}") from exc
@ -409,5 +424,5 @@ def install_agent_ensure(profile: str) -> None:
click.echo(f"Deployment '{profile}' is healthy.")
return
stop_runtime(manifest)
_start_deployment(manifest)
_start_deployment(manifest, assume_start_lock=True)
click.echo(f"Deployment '{profile}' is healthy.")

20
headroom/proxy/handlers/openai.py
View file

@ -7218,19 +7218,21 @@ class OpenAIHandlerMixin:
)
except TimeoutError:
logger.warning(
"Compression timed out after %.0fs (payload too large)",
"Compression timed out after %.0fs; failing open with original messages",
COMPRESSION_TIMEOUT_SECONDS,
)
return JSONResponse(
status_code=503,
content={
"error": {
"type": "compression_timeout",
"message": (
"Compression exceeded "
f"{COMPRESSION_TIMEOUT_SECONDS:.0f}s; payload too large."
),
}
"messages": messages,
"tokens_before": 0,
"tokens_after": 0,
"tokens_saved": 0,
"compression_ratio": 1.0,
"transforms_applied": [],
"transforms_summary": {},
"ccr_hashes": [],
"compression_skipped": True,
"skip_reason": "compression_timeout",
},
)
except Exception as e:

120
tests/test_cli/test_install_cli.py
View file

@ -43,6 +43,8 @@ def test_install_apply_starts_service_supervisor(monkeypatch) -> None:
monkeypatch.setattr(
"headroom.cli.install.wait_ready", lambda deployment, timeout_seconds=45: True
)
monkeypatch.setattr("headroom.cli.install.probe_ready", lambda url: False)
monkeypatch.setattr("headroom.cli.install.runtime_status", lambda manifest: "stopped")
result = runner.invoke(main, ["install", "apply"])
@ -153,6 +155,8 @@ def test_install_restart_uses_internal_helpers(monkeypatch) -> None:
monkeypatch.setattr(
"headroom.cli.install.wait_ready", lambda manifest, timeout_seconds=45: True
)
monkeypatch.setattr("headroom.cli.install.probe_ready", lambda url: False)
monkeypatch.setattr("headroom.cli.install.runtime_status", lambda manifest: "stopped")
result = runner.invoke(main, ["install", "restart"])
@ -161,6 +165,114 @@ def test_install_restart_uses_internal_helpers(monkeypatch) -> None:
assert calls == ["stop_supervisor", "stop_runtime", "start_supervisor"]
def test_install_start_noops_when_already_healthy(monkeypatch) -> None:
runner = CliRunner()
calls: list[str] = []
class Manifest:
profile = "default"
preset = "persistent-service"
runtime_kind = "python"
supervisor_kind = "service"
scope = "user"
health_url = "http://127.0.0.1:8787/readyz"
monkeypatch.setattr("headroom.cli.install.load_manifest", lambda profile: Manifest())
monkeypatch.setattr("headroom.cli.install.probe_ready", lambda url: True)
monkeypatch.setattr(
"headroom.cli.install.start_supervisor", lambda manifest: calls.append("start_supervisor")
)
result = runner.invoke(main, ["install", "start"])
assert result.exit_code == 0, result.output
assert "Started deployment 'default'." in result.output
assert calls == []
def test_install_start_noops_for_healthy_docker_without_docker_on_path(monkeypatch) -> None:
runner = CliRunner()
class Manifest:
profile = "default"
preset = "persistent-docker"
runtime_kind = "docker"
supervisor_kind = "none"
scope = "user"
health_url = "http://127.0.0.1:8787/readyz"
monkeypatch.setattr("headroom.cli.install.load_manifest", lambda profile: Manifest())
monkeypatch.setattr("headroom.cli.install.probe_ready", lambda url: True)
monkeypatch.setattr("headroom.cli.install.shutil.which", lambda name, *args, **kwargs: None)
result = runner.invoke(main, ["install", "start"])
assert result.exit_code == 0, result.output
assert "Started deployment 'default'." in result.output
def test_install_start_does_not_spawn_when_start_lock_is_contended(monkeypatch) -> None:
runner = CliRunner()
calls: list[str] = []
class Manifest:
profile = "default"
preset = "persistent-service"
runtime_kind = "python"
supervisor_kind = "service"
scope = "user"
health_url = "http://127.0.0.1:8787/readyz"
monkeypatch.setattr("headroom.cli.install.load_manifest", lambda profile: Manifest())
import contextlib
@contextlib.contextmanager
def fake_lock(profile):
yield False
monkeypatch.setattr("headroom.cli.install.acquire_runtime_start_lock", fake_lock)
monkeypatch.setattr(
"headroom.cli.install.start_supervisor", lambda manifest: calls.append("start_supervisor")
)
result = runner.invoke(main, ["install", "start"])
assert result.exit_code == 0, result.output
assert "start is already in progress" in result.output
assert calls == []
def test_install_start_restarts_wedged_runtime_under_single_lock(monkeypatch) -> None:
runner = CliRunner()
calls: list[str] = []
class Manifest:
profile = "default"
preset = "persistent-service"
runtime_kind = "python"
supervisor_kind = "service"
scope = "user"
health_url = "http://127.0.0.1:8787/readyz"
monkeypatch.setattr("headroom.cli.install.load_manifest", lambda profile: Manifest())
monkeypatch.setattr("headroom.cli.install.probe_ready", lambda url: False)
monkeypatch.setattr("headroom.cli.install.runtime_status", lambda manifest: "running")
wait_results = iter([False, True])
monkeypatch.setattr(
"headroom.cli.install.wait_ready", lambda manifest, timeout_seconds: next(wait_results)
)
monkeypatch.setattr("headroom.cli.install.stop_runtime", lambda manifest: calls.append("stop"))
monkeypatch.setattr(
"headroom.cli.install.start_supervisor", lambda manifest: calls.append("start_supervisor")
)
result = runner.invoke(main, ["install", "start"])
assert result.exit_code == 0, result.output
assert calls == ["stop", "start_supervisor"]
def test_install_apply_rejects_invalid_profile() -> None:
runner = CliRunner()
@ -350,6 +462,7 @@ def test_install_apply_uses_docker_runtime_for_persistent_docker(monkeypatch) ->
supervisor_kind = "none"
scope = "user"
health_url = "http://127.0.0.1:8787/readyz"
container_name = "headroom-default"
targets: list[str] = []
mutations = []
artifacts = []
@ -366,6 +479,8 @@ def test_install_apply_uses_docker_runtime_for_persistent_docker(monkeypatch) ->
monkeypatch.setattr(
"headroom.cli.install.wait_ready", lambda deployment, timeout_seconds=45: True
)
monkeypatch.setattr("headroom.cli.install.probe_ready", lambda url: False)
monkeypatch.setattr("headroom.cli.install.runtime_status", lambda deployment: "stopped")
# _start_deployment guards the persistent-docker preset with
# `shutil.which("docker")`. Fake docker as present so the test exercises the
# runtime-selection path itself rather than the host's docker install —
@ -517,7 +632,8 @@ def test_install_agent_ensure_stops_wedged_runtime_before_restart(monkeypatch) -
monkeypatch.setattr("headroom.cli.install.acquire_runtime_start_lock", fake_lock)
monkeypatch.setattr(
"headroom.cli.install._start_deployment", lambda manifest: calls.append("start_deployment")
"headroom.cli.install._start_deployment",
lambda manifest, **kwargs: calls.append("start_deployment"),
)
result = runner.invoke(main, ["install", "agent", "ensure"])
@ -626,7 +742,7 @@ def test_install_agent_ensure_propagates_start_deployment_failure(monkeypatch) -
monkeypatch.setattr("headroom.cli.install.acquire_runtime_start_lock", fake_lock)
def boom(manifest):
def boom(manifest, **kwargs):
raise click.ClickException("simulated start failure")
monkeypatch.setattr("headroom.cli.install._start_deployment", boom)

7
tests/test_install/test_runtime.py
View file

@ -272,7 +272,12 @@ def test_runtime_start_lock_blocks_another_process(monkeypatch, tmp_path: Path)
"with acquire_runtime_start_lock('default') as acquired:\n"
" print(acquired)\n"
)
env = {**os.environ, "HOME": str(tmp_path), "PYTHONPATH": str(Path.cwd())}
env = {
**os.environ,
"HOME": str(tmp_path),
"USERPROFILE": str(tmp_path),
"PYTHONPATH": str(Path.cwd()),
}
with acquire_runtime_start_lock("default") as acquired:
assert acquired is True

70
tests/test_platform_feature_matrix.py Normal file
View file

@ -0,0 +1,70 @@
from __future__ import annotations
import json
from pathlib import Path
MATRIX_PATH = Path("docs/platform-feature-matrix.json")
VALID_STATUSES = {"covered", "partial", "gap", "blocked"}
PLATFORMS = {"linux", "macos", "windows"}
def test_platform_feature_matrix_is_complete() -> None:
matrix = json.loads(MATRIX_PATH.read_text(encoding="utf-8"))
assert matrix["schema_version"] == 1
assert set(matrix["platforms"]) == PLATFORMS
assert set(matrix["status_values"]) == VALID_STATUSES
assert matrix["features"], "matrix must list hardening features"
feature_ids: set[str] = set()
for feature in matrix["features"]:
feature_id = feature["id"]
assert feature_id not in feature_ids, f"duplicate feature id: {feature_id}"
feature_ids.add(feature_id)
assert feature["name"]
assert feature["risk"] in {"install", "runtime", "performance", "cache", "proxy"}
assert set(feature["platforms"]) == PLATFORMS
for platform, coverage in feature["platforms"].items():
status = coverage["status"]
assert status in VALID_STATUSES, f"{feature_id}/{platform} has invalid status"
assert coverage["tests"], f"{feature_id}/{platform} must cite tests or workflows"
for test_ref in coverage["tests"]:
path = Path(test_ref.split("#", 1)[0])
assert path.exists(), f"{feature_id}/{platform} cites missing path {test_ref}"
if status in {"partial", "gap", "blocked"}:
assert coverage.get("gap"), f"{feature_id}/{platform} must explain {status}"
def test_platform_feature_matrix_covers_issue_1843_regression_areas() -> None:
matrix = json.loads(MATRIX_PATH.read_text(encoding="utf-8"))
feature_ids = {feature["id"] for feature in matrix["features"]}
assert {
"install_windows_service",
"single_instance_start",
"compression_fail_open",
"proxy_functional_smoke",
"ccr_persistence",
"toin_skip_recommendations",
} <= feature_ids
def test_platform_feature_matrix_sanity_tests_are_enumerated() -> None:
matrix = json.loads(MATRIX_PATH.read_text(encoding="utf-8"))
sanity_ids = {item["id"] for item in matrix["sanity_tests"]}
assert {
"cli_help",
"install_paths",
"runtime_selection",
"health_startup",
"compression_backpressure",
"proxy_route_smoke",
} <= sanity_ids
for item in matrix["sanity_tests"]:
assert item["description"]
assert item["tests"]
for test_ref in item["tests"]:
path = Path(test_ref.split("#", 1)[0])
assert path.exists(), f"{item['id']} cites missing path {test_ref}"

166
tests/test_platform_stabilization_functional.py Normal file
View file

@ -0,0 +1,166 @@
from __future__ import annotations
import json
import time
from typing import Any
import pytest
pytest.importorskip("fastapi")
pytest.importorskip("headroom._core")
from fastapi.testclient import TestClient
from headroom.config import TransformResult
from headroom.proxy.server import ProxyConfig, create_app
def _proxy_config(**overrides: Any) -> ProxyConfig:
defaults: dict[str, Any] = {
"optimize": True,
"cache_enabled": False,
"rate_limit_enabled": False,
"cost_tracking_enabled": False,
"log_requests": False,
"ccr_inject_tool": False,
"ccr_handle_responses": False,
"ccr_context_tracking": False,
"image_optimize": False,
"disable_kompress": True,
"compression_max_workers": 1,
}
defaults.update(overrides)
return ProxyConfig(**defaults)
def test_proxy_health_surfaces_compression_runtime_metrics(monkeypatch) -> None:
monkeypatch.setenv("HEADROOM_SKIP_UPSTREAM_CHECK", "1")
app = create_app(_proxy_config(optimize=False))
with TestClient(app, base_url="http://127.0.0.1", client=("127.0.0.1", 12345)) as client:
live = client.get("/livez")
health = client.get("/health")
assert live.status_code == 200
assert live.json()["alive"] is True
assert health.status_code == 200
runtime = health.json()["runtime"]
assert runtime["compression_executor"]["max_workers"] == 1
assert runtime["compression_executor"]["queued"] == 0
assert runtime["compression_executor"]["queue_timeouts_total"] == 0
def test_v1_compress_success_reports_actual_metrics(monkeypatch) -> None:
monkeypatch.setenv("HEADROOM_SKIP_UPSTREAM_CHECK", "1")
app = create_app(_proxy_config())
proxy = app.state.proxy
request_messages = [{"role": "user", "content": "summarize this repeated payload"}]
compressed_messages = [{"role": "user", "content": "summary payload"}]
def fake_apply(**kwargs):
assert kwargs["messages"] == request_messages
assert kwargs["model"] == "gpt-4o"
return TransformResult(
messages=compressed_messages,
tokens_before=100,
tokens_after=40,
transforms_applied=["test:compress"],
markers_inserted=["marker-1"],
)
monkeypatch.setattr(proxy.openai_pipeline, "apply", fake_apply)
with TestClient(app, base_url="http://127.0.0.1", client=("127.0.0.1", 12345)) as client:
response = client.post(
"/v1/compress",
json={"model": "gpt-4o", "messages": request_messages},
)
body = response.json()
assert response.status_code == 200
assert body["messages"] == compressed_messages
assert body["tokens_before"] == 100
assert body["tokens_after"] == 40
assert body["tokens_saved"] == 60
assert body["compression_ratio"] == 0.4
assert body["transforms_applied"] == ["test:compress"]
assert body["transforms_summary"] == {"test:compress": 1}
assert body["ccr_hashes"] == ["marker-1"]
def test_v1_compress_timeout_fails_open_quickly(monkeypatch) -> None:
monkeypatch.setenv("HEADROOM_SKIP_UPSTREAM_CHECK", "1")
app = create_app(_proxy_config())
proxy = app.state.proxy
request_messages = [{"role": "user", "content": "do not mutate me"}]
async def timeout_executor(fn, *, timeout): # noqa: ANN001
raise TimeoutError("compression deadline exceeded")
monkeypatch.setattr(proxy, "_run_compression_in_executor", timeout_executor)
with TestClient(app, base_url="http://127.0.0.1", client=("127.0.0.1", 12345)) as client:
started = time.perf_counter()
response = client.post(
"/v1/compress",
json={"model": "gpt-4o", "messages": request_messages},
)
elapsed = time.perf_counter() - started
body = response.json()
assert response.status_code == 200
assert elapsed < 0.5
assert body["messages"] == request_messages
assert body["tokens_saved"] == 0
assert body["compression_ratio"] == 1.0
assert body["transforms_applied"] == []
assert body["compression_skipped"] is True
assert body["skip_reason"] == "compression_timeout"
def test_v1_compress_real_json_tool_payload_reduces_tokens(monkeypatch) -> None:
monkeypatch.setenv("HEADROOM_SKIP_UPSTREAM_CHECK", "1")
app = create_app(
_proxy_config(
ccr_inject_marker=False,
min_tokens_to_crush=20,
max_items_after_crush=10,
)
)
items = [
{
"id": i,
"status": "ok",
"score": i % 5,
"message": "same repeated value " * 20,
}
for i in range(80)
]
request = {
"model": "gpt-4o",
"messages": [
{"role": "user", "content": "summarize rows"},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call-1",
"type": "function",
"function": {"name": "list_rows", "arguments": "{}"},
}
],
},
{"role": "tool", "tool_call_id": "call-1", "content": json.dumps(items)},
],
}
with TestClient(app, base_url="http://127.0.0.1", client=("127.0.0.1", 12345)) as client:
response = client.post("/v1/compress", json=request)
body = response.json()
assert response.status_code == 200, response.text
assert body["tokens_before"] > body["tokens_after"], body
assert body["tokens_saved"] > 0
assert body["compression_ratio"] < 1.0
assert body["transforms_applied"], body