runtime: add explicit pressure-policy dry-run API

src/local_llm_server/pressure_api.py

@@ -0,0 +1,95 @@
+"""Admin-only dry-run evaluation for host memory pressure and residency policy.
+
+The endpoint samples the host on explicit POST requests and advances the
+hysteretic policy state, but it never unloads a runtime. This makes pressure
+policy behavior observable on representative devices before any automatic action
+is enabled.
+"""
+from __future__ import annotations
+
+import platform
+from typing import Any
+
+from fastapi import FastAPI, HTTPException, Request
+
+from .residency_pressure import PressureEvictionPolicy
+from .resources import (
+    ResourceObserver,
+    ResourceValue,
+    StandardLibraryResourceObserver,
+    classify_memory_pressure,
+)
+from .resources_macos import MacOSResourceObserver
+
+
+def install_pressure_dry_run_api(
+    application: FastAPI,
+    *,
+    observer: ResourceObserver | None = None,
+    policy: PressureEvictionPolicy | None = None,
+) -> FastAPI:
+    """Install one explicit, stateful, non-destructive pressure evaluator."""
+    if getattr(application.state, "pressure_dry_run_api_installed", False):
+        return application
+    application.state.pressure_dry_run_api_installed = True
+
+    settings = getattr(application.state, "settings", None)
+    if not bool(getattr(settings, "enable_admin_api", False)):
+        return application
+
+    application.state.pressure_dry_run_observer = observer or _default_observer()
+    application.state.pressure_eviction_policy = policy or PressureEvictionPolicy()
+
+    def evaluate_pressure(request: Request) -> dict[str, Any]:
+        manager = request.app.state.runtime_manager
+        residency_snapshot = getattr(manager, "residency_policy_snapshot", None)
+        if not callable(residency_snapshot):
+            raise HTTPException(
+                status_code=501,
+                detail="Runtime manager does not expose residency policy state.",
+            )
+
+        resource_snapshot = request.app.state.pressure_dry_run_observer.snapshot()
+        pressure = classify_memory_pressure(resource_snapshot)
+        evaluation = request.app.state.pressure_eviction_policy.observe(
+            pressure,
+            residency_snapshot(),
+        )
+        return {
+            "mode": "dry_run",
+            "action_executed": False,
+            "resource": {
+                "platform": resource_snapshot.platform,
+                "total_memory_bytes": _resource_value(resource_snapshot.total_memory_bytes),
+                "available_memory_bytes": _resource_value(resource_snapshot.available_memory_bytes),
+                "thermal_pressure": _resource_value(resource_snapshot.thermal_pressure),
+            },
+            "evaluation": evaluation.to_public_dict(),
+            "claim_boundary": (
+                "Dry-run pressure policy only. No runtime was unloaded and no "
+                "memory-reclamation or production-safety claim is made."
+            ),
+        }
+
+    application.add_api_route(
+        "/api/v1/residency/pressure/evaluate",
+        evaluate_pressure,
+        methods=["POST"],
+        tags=["Resources"],
+        name="evaluate_residency_pressure_dry_run",
+    )
+    return application
+
+
+def _default_observer() -> ResourceObserver:
+    if platform.system().lower() == "darwin":
+        return MacOSResourceObserver()
+    return StandardLibraryResourceObserver()
+
+
+def _resource_value(value: ResourceValue) -> dict[str, object]:
+    return {
+        "value": value.value,
+        "source": value.source.value,
+        "unit": value.unit,
+    }

src/local_llm_server/product_composition.py

@@ -8,6 +8,7 @@
 from .completion_metrics import install_completion_metrics
 from .control_plane_api import install_product_api
 from .policy_evidence import install_policy_evidence_api
+from .pressure_api import install_pressure_dry_run_api
 from .request_middleware import install_request_policy
 from .request_scheduler import install_request_scheduler
 from .residency_api import install_residency_api
@@ -28,8 +29,8 @@ def install_product_http_stack(
     metrics. At execution time stream timing is outermost, completion metrics
     handles only non-stream requests, policy prepares the canonical request and
     scheduler gates it before the route/runtime lease. Product API installs the
-    cold-state layer last. Residency and policy evidence routes remain admin-only
-    and expose bounded state rather than inference content or private paths.
+    cold-state layer last. Residency/policy/pressure routes remain admin-only;
+    pressure evaluation is explicit dry-run state and never unloads a runtime.
     """
     install_request_scheduler(application, settings=scheduler_settings)
     install_request_policy(application)
@@ -38,4 +39,5 @@ def install_product_http_stack(
     install_product_api(application, evaluation_root=evaluation_root)
     install_residency_api(application)
     install_policy_evidence_api(application)
+    install_pressure_dry_run_api(application)
     return application

tests/test_pressure_api.py

@@ -0,0 +1,174 @@
+from __future__ import annotations
+
+from collections import deque
+
+from fastapi.testclient import TestClient
+
+from local_llm_server.pressure_api import install_pressure_dry_run_api
+from local_llm_server.product_runtime_manager import ProductRuntimeManager
+from local_llm_server.residency_pressure import PressureEvictionPolicy
+from local_llm_server.resources import (
+    ResourceValue,
+    ResourceValueSource,
+    SystemResourceSnapshot,
+)
+from local_llm_server.server import ServerSettings, create_app
+
+
+def _measured(value: int) -> ResourceValue:
+    return ResourceValue(value, ResourceValueSource.MEASURED, "bytes")
+
+
+def _snapshot(*, available: int | None) -> SystemResourceSnapshot:
+    return SystemResourceSnapshot(
+        captured_at_monotonic=1.0,
+        platform="test",
+        total_memory_bytes=(
+            _measured(1_000)
+            if available is not None
+            else ResourceValue.unavailable("bytes")
+        ),
+        available_memory_bytes=(
+            _measured(available)
+            if available is not None
+            else ResourceValue.unavailable("bytes")
+        ),
+        process_rss_bytes=_measured(123),
+    )
+
+
+class _Observer:
+    def __init__(self, snapshots):
+        self.snapshots = deque(snapshots)
+
+    def snapshot(self):
+        return self.snapshots.popleft()
+
+
+class _Engine:
+    backend = "fake"
+
+    def close(self):
+        pass
+
+
+def _cfg(key: str):
+    return {
+        "model": key,
+        "model_id": f"org/{key}",
+        "backend": "fake",
+        "modalities": ["text"],
+        "max_concurrent_requests": 1,
+    }
+
+
+def _app(*, admin: bool, observer=None):
+    manager = ProductRuntimeManager(default_model="default")
+    manager.add(_cfg("default"), _Engine())
+    manager.add(_cfg("old"), _Engine())
+    application = create_app(
+        manager,
+        settings=ServerSettings(enable_admin_api=admin),
+    )
+    install_pressure_dry_run_api(
+        application,
+        observer=observer,
+        policy=PressureEvictionPolicy(),
+    )
+    return application, manager
+
+
+def test_two_critical_samples_trigger_candidate_without_unloading_runtime():
+    observer = _Observer([
+        _snapshot(available=50),
+        _snapshot(available=50),
+    ])
+    application, manager = _app(admin=True, observer=observer)
+    client = TestClient(application)
+
+    first = client.post("/api/v1/residency/pressure/evaluate")
+    second = client.post("/api/v1/residency/pressure/evaluate")
+
+    assert first.status_code == 200
+    assert first.json()["evaluation"]["pressure"] == "critical"
+    assert first.json()["evaluation"]["state"] == "watching"
+    assert first.json()["evaluation"]["should_attempt_eviction"] is False
+
+    payload = second.json()
+    assert payload["mode"] == "dry_run"
+    assert payload["action_executed"] is False
+    assert payload["evaluation"]["state"] == "triggered"
+    assert payload["evaluation"]["transition"] == "triggered"
+    assert payload["evaluation"]["should_attempt_eviction"] is True
+    assert [item["key"] for item in payload["evaluation"]["candidates"]] == ["old"]
+    assert payload["evaluation"]["automatic_eviction_enabled"] is False
+    assert payload["evaluation"]["reclamation_claim"] is False
+    assert "No runtime was unloaded" in payload["claim_boundary"]
+
+    assert sorted(runtime.key for runtime in manager.list()) == ["default", "old"]
+    assert manager.default_model == "default"
+
+
+def test_dry_run_exposes_host_memory_sources_but_not_process_rss():
+    application, _ = _app(
+        admin=True,
+        observer=_Observer([_snapshot(available=500)]),
+    )
+    payload = TestClient(application).post(
+        "/api/v1/residency/pressure/evaluate"
+    ).json()
+
+    assert payload["resource"] == {
+        "platform": "test",
+        "total_memory_bytes": {
+            "value": 1_000,
+            "source": "measured",
+            "unit": "bytes",
+        },
+        "available_memory_bytes": {
+            "value": 500,
+            "source": "measured",
+            "unit": "bytes",
+        },
+        "thermal_pressure": {
+            "value": None,
+            "source": "unavailable",
+            "unit": "level",
+        },
+    }
+    assert "process_rss" not in str(payload["resource"])
+
+
+def test_unknown_resource_observation_never_triggers_candidate():
+    application, manager = _app(
+        admin=True,
+        observer=_Observer([_snapshot(available=None)]),
+    )
+    payload = TestClient(application).post(
+        "/api/v1/residency/pressure/evaluate"
+    ).json()
+
+    assert payload["evaluation"]["pressure"] == "unknown"
+    assert payload["evaluation"]["should_attempt_eviction"] is False
+    assert payload["evaluation"]["candidates"] == []
+    assert len(manager.list()) == 2
+
+
+def test_pressure_dry_run_route_is_admin_only():
+    application, _ = _app(
+        admin=False,
+        observer=_Observer([_snapshot(available=50)]),
+    )
+    client = TestClient(application)
+
+    assert client.post("/api/v1/residency/pressure/evaluate").status_code == 404
+
+
+def test_pressure_evaluation_requires_post_because_it_advances_hysteresis_state():
+    application, _ = _app(
+        admin=True,
+        observer=_Observer([_snapshot(available=50)]),
+    )
+    client = TestClient(application)
+
+    assert client.get("/api/v1/residency/pressure/evaluate").status_code == 405

tests/test_product_composition.py

@@ -52,12 +52,18 @@ def test_product_http_stack_installs_scheduler_policy_completion_stream_metrics_
     assert app.state.completion_metrics_installed is True
     assert app.state.streaming_metrics_installed is True
     assert app.state.product_api_installed is True
+    assert app.state.residency_api_installed is True
+    assert app.state.policy_evidence_api_installed is True
+    assert app.state.pressure_dry_run_api_installed is True
     route_paths = [
         path
         for route in app.routes
         if (path := getattr(route, "path", None)) is not None
     ]
     assert route_paths.count("/v1/audio/transcriptions") == 1
+    # The admin-only pressure route is deliberately absent when the admin API
+    # is disabled even though composition remains idempotently installed.
+    assert "/api/v1/residency/pressure/evaluate" not in route_paths
 
 
 def test_product_http_stack_keeps_scheduler_disabled_when_queue_is_unconfigured(tmp_path):
@@ -71,6 +77,7 @@ def test_product_http_stack_keeps_scheduler_disabled_when_queue_is_unconfigured(
     assert app.state.request_scheduler_settings.enabled is False
     assert app.state.runtime_gate_registry is None
     assert app.state.completion_metrics_installed is True
+    assert app.state.pressure_dry_run_api_installed is True
 
 
 def test_supported_server_entrypoints_use_shared_product_composition():