From a13112b135274ff310df3f72cfeee51a5a7919c2 Mon Sep 17 00:00:00 2001 From: Daniele21 Date: Sat, 15 Aug 2026 12:58:20 +0200 Subject: [PATCH 1/4] runtime: add pressure-policy dry-run admin endpoint --- src/local_llm_server/pressure_api.py | 95 ++++++++++++++++++++++++++++ 1 file changed, 95 insertions(+) create mode 100644 src/local_llm_server/pressure_api.py diff --git a/src/local_llm_server/pressure_api.py b/src/local_llm_server/pressure_api.py new file mode 100644 index 0000000..a9835fb --- /dev/null +++ b/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, + } From f940329c8db4e593b2d06eaf325e7c498044ac2d Mon Sep 17 00:00:00 2001 From: Daniele21 Date: Sat, 15 Aug 2026 12:58:36 +0200 Subject: [PATCH 2/4] runtime: compose pressure dry-run API into product stack --- src/local_llm_server/product_composition.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/local_llm_server/product_composition.py b/src/local_llm_server/product_composition.py index cfac231..da0ec4c 100644 --- a/src/local_llm_server/product_composition.py +++ b/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 From 392975218c20c4d5e611edaff5e3705655b39b89 Mon Sep 17 00:00:00 2001 From: Daniele21 Date: Sat, 15 Aug 2026 12:59:00 +0200 Subject: [PATCH 3/4] test: cover non-destructive pressure dry-run API --- tests/test_pressure_api.py | 174 +++++++++++++++++++++++++++++++++++++ 1 file changed, 174 insertions(+) create mode 100644 tests/test_pressure_api.py diff --git a/tests/test_pressure_api.py b/tests/test_pressure_api.py new file mode 100644 index 0000000..26fc283 --- /dev/null +++ b/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 From b8fdd692bf7555a9e07a8a53a80b25e3cf89d62f Mon Sep 17 00:00:00 2001 From: Daniele21 Date: Sat, 15 Aug 2026 12:59:22 +0200 Subject: [PATCH 4/4] test: assert pressure dry-run participates in product composition --- tests/test_product_composition.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/tests/test_product_composition.py b/tests/test_product_composition.py index ef24b34..515a635 100644 --- a/tests/test_product_composition.py +++ b/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():