diff --git a/AGENTS.md b/AGENTS.md index c06f1bb71..0d0379f5f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -199,6 +199,7 @@ Each domain has exactly **one** write-owning module (or one tightly-scoped famil | Config (`config/journal.json`) | `solstone/think/journal_config.py` | | Schedules (`config/schedules.json`) | `solstone/think/schedule_config.py` | | Push devices (`config/push_devices.json`) | `solstone/think/push/devices.py` | +| Local inference operational telemetry (`health/local-inference/YYYYMMDD.jsonl`) | `solstone/think/providers/local_admission.py` | | Hosted backup binding (`backup/hosted/binding.json`) | `solstone/think/backup/hosted.py` | | Convey config (`config/convey.json`) | `solstone/convey/config.py` + `solstone/think/facets.py` | | Chat config (`config/chat.json`) | `solstone/apps/chat/config.py` | @@ -211,7 +212,7 @@ Each domain has exactly **one** write-owning module (or one tightly-scoped famil | Index (SQLite, `indexer/*`) | `solstone/think/indexer/*` | | Observer registry and sync history (`apps/observer/observers/*.json`, `apps/observer/observers/*/hist/*.jsonl`) | `solstone/apps/observer/utils.py` | | Import ingest/resolve staging (`imports/**`) | `solstone/apps/import/ingest.py` + `solstone/apps/import/resolve.py` + `solstone/apps/import/facet_ingest.py` + `solstone/apps/import/journal_sources.py` — HTTP-ingest + resolve staging state, plus the remote-ingest bundle under `imports//`. `journal_sources.py` owns only its `create_state_directory` `imports/` initializers; its source registry is app-storage. Import-bundle and sync-cursor content under `imports//` and `imports/.json` is written by `solstone/think/importers/{utils,cli,shared,sync}.py` (local/CLI import flows + sync cursor), and `solstone/think/importers/plaud.py` installs streamed imported audio onto `imports//` via journal_io's `install_file` primitive, as importer declared outputs (L7). | -| Operational-log/cache pruning and root task-log compaction — deletion/compaction only, across the dated allowlist (`chronicle//health/*.{log,jsonl}`, top-level `talents//.jsonl` run logs + `talents/.jsonl` day indexes, `tokens/`, `awareness/`, `config/actions/`, `facets/*/logs/`, `apps/observer/observers/*/hist/`, `.cache/cogitate-history/`) plus root `task_log.txt` epoch lines older than the same retention window | `solstone/think/log_retention.py` — prunes derived/operational artifacts by retention age and rewrites only root `task_log.txt` during compaction, preserving recent and unparseable lines. Routes its own deletes/compaction; does not write through each domain's owner. | +| Operational-log/cache pruning and root task-log compaction — deletion/compaction only, across the dated allowlist (`chronicle//health/*.{log,jsonl}`, top-level `talents//.jsonl` run logs + `talents/.jsonl` day indexes, `tokens/`, `health/local-inference/`, `awareness/`, `config/actions/`, `facets/*/logs/`, `apps/observer/observers/*/hist/`, `.cache/cogitate-history/`) plus root `task_log.txt` epoch lines older than the same retention window | `solstone/think/log_retention.py` — prunes derived/operational artifacts by retention age and rewrites only root `task_log.txt` during compaction, preserving recent and unparseable lines. Routes its own deletes/compaction; does not write through each domain's owner. | If you're about to write to a domain from a module not in this table, stop and route through the owner. diff --git a/docs/PROVIDERS.md b/docs/PROVIDERS.md index 6087bb7de..76cdf8456 100644 --- a/docs/PROVIDERS.md +++ b/docs/PROVIDERS.md @@ -224,6 +224,89 @@ usage_dict = { - Return usage in `GenerateResult["usage"]` - wrapper handles logging - For `run_cogitate()`, include usage in the `finish` event +## Bundled-local admission and inference telemetry + +The `local` provider has one shared admission boundary for the supervisor-owned +Qwen server. It applies only when `providers.local` resolves to the bundled +loopback runtime. A configured OpenAI-compatible endpoint and every cloud +provider bypass this boundary. + +Capacity remains explicit and intentionally small: + +| Runtime profile | Serving capacity | Evidence | +|---|---:|---| +| Linux floor | 1 | supervisor `ServerTier`; live `/props.total_slots` wins | +| Linux capable (at least 16 GiB tiering VRAM) | 2 | supervisor `ServerTier`; live `/props.total_slots` wins | +| Apple MLX | 1 | conservative explicit fallback because mlx-vlm 0.6.2 does not advertise a slot limit | + +The provider memoizes the capacity once per process. It first reads live +`/props.total_slots`, then the persisted `health/local.ctx` launch tier, then +falls back to one. The supervisor remains the configuration owner: changing a +Linux tier's `parallel_slots` changes both `llama-server --parallel` and provider +admission after the journal processes restart. Apple stays at one until that +runtime exposes a stable capacity contract and a separate measurement justifies +raising it. + +Admission uses one `flock` file per slot under +`health/local-inference-admission/`. This coordinates independent journal +processes without a scheduler service or in-memory queue. Waiting async calls +are cancellation-safe; exceptions and cancellation release acquired locks; +process exit releases kernel locks. Queue time consumes the caller's existing +provider deadline, so waiting cannot silently extend a request beyond its +configured timeout. Cogitate holds one permit for its run because the OpenHands +SDK owns its internal multi-turn HTTP calls; this is conservative and avoids an +uncontrolled second path to the same server. + +Every bundled-local attempt appends a content-free JSON record to +`health/local-inference/YYYYMMDD.jsonl`. These files follow the configured +`retention.journal_logs.days` policy. Records contain: + +- request id, timestamp, generate/cogitate kind, provider, logical model, and + runtime profile; +- serving capacity, evidence source, admission slot, and client queue wait; +- client wall time plus server prompt-evaluation, generation, and total timing + when the response exposes them; +- prompt/generated token counts, reused prompt tokens, prompt-cache cold/warm + state, and selected server slot when exposed; +- retry index, finish reason, outcome, timeout/cancellation flags, and a safe + reason code on failure. + +Records never contain prompt text, generated text, messages, schemas, images, +endpoint URLs, or credentials. A successful `GenerateResult` also carries the +same record as `inference`; callers that already retain the full result can use +it without rereading the log. Fields unavailable from a runtime remain null or +`unknown` rather than being inferred. + +Run the synthetic journal-shaped benchmark against an isolated server: + +```bash +python scripts/benchmark_local_inference_admission.py \ + --endpoint http://127.0.0.1:8080 --slots 2 --concurrency 10 \ + --requests 30 --mode baseline +python scripts/benchmark_local_inference_admission.py \ + --endpoint http://127.0.0.1:8080 --slots 2 --concurrency 10 \ + --requests 30 --mode admitted +``` + +It reports throughput, latency P50/P95/P99, explicit queue wait, residual opaque +wait, failures, and NVIDIA peak memory/utilization when `nvidia-smi` is present. +The payloads are fixed synthetic text/JSON requests and never read a journal. + +The implementation gate on `fedora.local` used a fresh b9957 server with two +slots, ten producers, and twelve mixed requests for each side. Baseline versus +admitted results were 0.1291 versus 0.1296 requests/s, P95 latency 78.71 versus +77.22 seconds, P99 80.28 versus 77.30 seconds, and identical 4,652 MiB peak GPU +memory. Most importantly, P95 wait hidden inside the server fell from 65.56 +seconds to 1.22 seconds; the admitted run reported 62.38 seconds as explicit +client queue wait. The boundary preserved throughput, slightly improved the +tail, and made the backlog observable without changing model work. + +Rollback is one code revert plus a journal-process restart. The lock files hold +no state and may remain on disk; removing the provider calls to +`acquire_local_slot*()` immediately restores server-side queueing. Telemetry +JSONL files are ordinary operational logs and can remain until retention prunes +them. + ## Context & Routing Context strings determine provider and model selection. Providers receive already-resolved models, but understanding the system helps: diff --git a/scripts/benchmark_local_inference_admission.py b/scripts/benchmark_local_inference_admission.py new file mode 100644 index 000000000..ae9292e04 --- /dev/null +++ b/scripts/benchmark_local_inference_admission.py @@ -0,0 +1,275 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Reproducible synthetic mixed-load benchmark for bundled-local admission.""" + +from __future__ import annotations + +import argparse +import concurrent.futures +import json +import os +import statistics +import subprocess +import tempfile +import threading +import time +from pathlib import Path +from typing import Any + +import httpx + +from solstone.think.providers.local_admission import acquire_local_slot + +WORKLOADS = ( + { + "name": "short_json", + "prompt": "Return JSON with keys summary and confidence about a synthetic note.", + "max_tokens": 48, + "response_format": { + "type": "json_schema", + "json_schema": { + "name": "short", + "strict": True, + "schema": { + "type": "object", + "properties": { + "summary": {"type": "string"}, + "confidence": {"type": "number"}, + }, + "required": ["summary", "confidence"], + "additionalProperties": False, + }, + }, + }, + }, + { + "name": "entity_json", + "prompt": ( + "From this synthetic project update, return JSON containing people and " + "projects arrays. Alex discussed Orion with Sam; no real data is present." + ), + "max_tokens": 96, + "response_format": { + "type": "json_schema", + "json_schema": { + "name": "entities", + "strict": True, + "schema": { + "type": "object", + "properties": { + "people": {"type": "array", "items": {"type": "string"}}, + "projects": {"type": "array", "items": {"type": "string"}}, + }, + "required": ["people", "projects"], + "additionalProperties": False, + }, + }, + }, + }, + { + "name": "brief_text", + "prompt": "Summarize a synthetic meeting in three concise bullet points.", + "max_tokens": 72, + }, +) + + +def _percentile(values: list[float], percentile: float) -> float: + if not values: + return 0.0 + ordered = sorted(values) + rank = (len(ordered) - 1) * percentile + lower = int(rank) + upper = min(lower + 1, len(ordered) - 1) + fraction = rank - lower + return ordered[lower] + (ordered[upper] - ordered[lower]) * fraction + + +def _distribution(values: list[float]) -> dict[str, float]: + return { + "p50": round(_percentile(values, 0.50), 3), + "p95": round(_percentile(values, 0.95), 3), + "p99": round(_percentile(values, 0.99), 3), + "mean": round(statistics.fmean(values), 3) if values else 0.0, + "max": round(max(values), 3) if values else 0.0, + } + + +def _server_ms(data: dict[str, Any]) -> float: + timings = data.get("timings") + if not isinstance(timings, dict): + return 0.0 + return float(timings.get("prompt_ms") or 0) + float( + timings.get("predicted_ms") or 0 + ) + + +def _run_one( + *, + endpoint: str, + model: str, + workload: dict[str, Any], + slots: int, + admitted: bool, + timeout_s: float, +) -> dict[str, Any]: + started = time.monotonic() + permit = acquire_local_slot(slots, timeout_s) if admitted else None + queue_wait_ms = permit.queue_wait_ms if permit is not None else 0.0 + try: + body = { + "model": model, + "messages": [{"role": "user", "content": workload["prompt"]}], + "max_tokens": workload["max_tokens"], + "temperature": 0.2, + "stream": False, + "chat_template_kwargs": {"enable_thinking": False}, + } + if "response_format" in workload: + body["response_format"] = workload["response_format"] + response = httpx.post( + f"{endpoint.rstrip('/')}/v1/chat/completions", + json=body, + timeout=timeout_s, + ) + response.raise_for_status() + data = response.json() + latency_ms = (time.monotonic() - started) * 1000.0 + server_ms = _server_ms(data) + return { + "workload": workload["name"], + "ok": True, + "latency_ms": latency_ms, + "queue_wait_ms": queue_wait_ms, + "server_ms": server_ms, + "opaque_wait_ms": max(0.0, latency_ms - queue_wait_ms - server_ms), + } + except Exception as exc: + return { + "workload": workload["name"], + "ok": False, + "latency_ms": (time.monotonic() - started) * 1000.0, + "queue_wait_ms": queue_wait_ms, + "server_ms": 0.0, + "opaque_wait_ms": 0.0, + "error_type": type(exc).__name__, + } + finally: + if permit is not None: + permit.release() + + +def _sample_gpu(stop: threading.Event, samples: list[dict[str, float]]) -> None: + while not stop.wait(0.1): + try: + output = subprocess.run( + [ + "nvidia-smi", + "--query-gpu=memory.used,utilization.gpu", + "--format=csv,noheader,nounits", + ], + check=True, + capture_output=True, + text=True, + timeout=2, + ).stdout.splitlines()[0] + memory_mib, utilization = ( + float(item.strip()) for item in output.split(",") + ) + samples.append( + {"gpu_memory_mib": memory_mib, "gpu_utilization_pct": utilization} + ) + except (FileNotFoundError, IndexError, ValueError, subprocess.SubprocessError): + return + + +def _summarize(rows: list[dict[str, Any]]) -> dict[str, Any]: + successful = [row for row in rows if row["ok"]] + return { + "requests": len(rows), + "succeeded": len(successful), + "failed": len(rows) - len(successful), + "latency_ms": _distribution([row["latency_ms"] for row in successful]), + "queue_wait_ms": _distribution([row["queue_wait_ms"] for row in successful]), + "opaque_wait_ms": _distribution([row["opaque_wait_ms"] for row in successful]), + } + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--endpoint", required=True) + parser.add_argument("--model", default="local/qwen3.5-4b") + parser.add_argument("--slots", type=int, required=True) + parser.add_argument("--concurrency", type=int, default=10) + parser.add_argument("--requests", type=int, default=30) + parser.add_argument("--timeout", type=float, default=300.0) + parser.add_argument("--mode", choices=("baseline", "admitted"), required=True) + parser.add_argument("--output", type=Path) + args = parser.parse_args() + if args.slots < 1 or args.concurrency < 1 or args.requests < 1: + parser.error("slots, concurrency, and requests must be positive") + + with tempfile.TemporaryDirectory(prefix="solstone-admission-bench-") as state_dir: + os.environ["SOLSTONE_JOURNAL"] = state_dir + started = time.monotonic() + gpu_samples: list[dict[str, float]] = [] + stop = threading.Event() + sampler = threading.Thread( + target=_sample_gpu, args=(stop, gpu_samples), daemon=True + ) + sampler.start() + with concurrent.futures.ThreadPoolExecutor( + max_workers=args.concurrency + ) as executor: + futures = [ + executor.submit( + _run_one, + endpoint=args.endpoint, + model=args.model, + workload=WORKLOADS[index % len(WORKLOADS)], + slots=args.slots, + admitted=args.mode == "admitted", + timeout_s=args.timeout, + ) + for index in range(args.requests) + ] + rows = [future.result() for future in futures] + elapsed_s = time.monotonic() - started + stop.set() + sampler.join(timeout=1) + + result = { + "mode": args.mode, + "endpoint": args.endpoint, + "model": args.model, + "configured_slots": args.slots, + "producer_concurrency": args.concurrency, + "elapsed_s": round(elapsed_s, 3), + "throughput_requests_per_s": round(args.requests / elapsed_s, 4), + "all": _summarize(rows), + "by_workload": { + workload["name"]: _summarize( + [row for row in rows if row["workload"] == workload["name"]] + ) + for workload in WORKLOADS + }, + "resources": { + "gpu_samples": len(gpu_samples), + "peak_gpu_memory_mib": max( + (sample["gpu_memory_mib"] for sample in gpu_samples), default=None + ), + "peak_gpu_utilization_pct": max( + (sample["gpu_utilization_pct"] for sample in gpu_samples), + default=None, + ), + }, + } + rendered = json.dumps(result, indent=2, sort_keys=True) + if args.output: + args.output.write_text(rendered + "\n", encoding="utf-8") + print(rendered) + + +if __name__ == "__main__": + main() diff --git a/scripts/check_journal_io_access.py b/scripts/check_journal_io_access.py index 9167f5205..476d17314 100644 --- a/scripts/check_journal_io_access.py +++ b/scripts/check_journal_io_access.py @@ -116,6 +116,8 @@ OWNER_FILES: frozenset[str] = frozenset( "solstone/think/identity.py", "solstone/think/journal_config.py", "solstone/think/log_retention.py", + # Sole writer of content-free bundled-local inference telemetry. + "solstone/think/providers/local_admission.py", "solstone/think/schedule_config.py", "solstone/think/push/devices.py", # Backup hosted-tier binding (0600 broker-token cache). diff --git a/solstone/think/log_retention.py b/solstone/think/log_retention.py index ceee18cfe..2fedb0512 100644 --- a/solstone/think/log_retention.py +++ b/solstone/think/log_retention.py @@ -27,6 +27,7 @@ CLASS_NAMES = ( "talent_day_index", "cogitate_history_cache", "tokens", + "local_inference", "awareness_logs", "config_actions", "facet_logs", @@ -170,6 +171,15 @@ def prune( base=journal_path / "tokens", pattern="*.jsonl", ) + _scan_dated_files( + journal_path, + cutoff, + result, + dry_run=dry_run, + class_name="local_inference", + base=journal_path / "health" / "local-inference", + pattern="*.jsonl", + ) _scan_dated_files( journal_path, cutoff, diff --git a/solstone/think/providers/local.py b/solstone/think/providers/local.py index eb46fcac9..124ed179d 100644 --- a/solstone/think/providers/local.py +++ b/solstone/think/providers/local.py @@ -12,7 +12,9 @@ from __future__ import annotations import asyncio import copy import logging +import time import traceback +import uuid from collections.abc import Callable from dataclasses import dataclass from typing import Any @@ -265,11 +267,17 @@ def _extract_usage(data: dict[str, Any]) -> dict[str, int] | None: input_tokens = int(usage.get("prompt_tokens") or 0) output_tokens = int(usage.get("completion_tokens") or 0) total_tokens = int(usage.get("total_tokens") or input_tokens + output_tokens) - return { + normalized = { "input_tokens": input_tokens, "output_tokens": output_tokens, "total_tokens": total_tokens, } + prompt_details = usage.get("prompt_tokens_details") + if isinstance(prompt_details, dict): + cached_tokens = int(prompt_details.get("cached_tokens") or 0) + if cached_tokens: + normalized["cached_tokens"] = cached_tokens + return normalized def _parse_response(data: dict[str, Any]) -> GenerateResult: @@ -295,6 +303,97 @@ def _parse_response(data: dict[str, Any]) -> GenerateResult: ) +def _number(value: Any) -> int | float | None: + if isinstance(value, bool) or not isinstance(value, int | float): + return None + return value + + +def _server_response_fields(data: dict[str, Any]) -> dict[str, Any]: + """Normalize content-free timing/cache/slot fields exposed by local servers.""" + timings = data.get("timings") + timings = timings if isinstance(timings, dict) else {} + usage = data.get("usage") + usage = usage if isinstance(usage, dict) else {} + prompt_details = usage.get("prompt_tokens_details") + prompt_details = prompt_details if isinstance(prompt_details, dict) else {} + + cached_tokens = _number(timings.get("cache_n")) + if cached_tokens is None: + cached_tokens = _number(prompt_details.get("cached_tokens")) + slot_id = _number(data.get("id_slot")) + if slot_id is None: + slot_id = _number(data.get("slot_id")) + if slot_id is None: + slot_id = _number(timings.get("slot_id")) + + prompt_ms = _number(timings.get("prompt_ms")) + generation_ms = _number(timings.get("predicted_ms")) + server_total_ms = None + if prompt_ms is not None or generation_ms is not None: + server_total_ms = float(prompt_ms or 0) + float(generation_ms or 0) + + return { + "prompt_eval_ms": prompt_ms, + "generation_ms": generation_ms, + "server_total_ms": server_total_ms, + "prompt_tokens": _number(usage.get("prompt_tokens")) + or _number(timings.get("prompt_n")), + "generated_tokens": _number(usage.get("completion_tokens")) + or _number(timings.get("predicted_n")), + "prompt_cached_tokens": cached_tokens, + "selected_slot": slot_id, + "prompt_cache_state": ( + "warm" + if cached_tokens is not None and cached_tokens > 0 + else "cold" + if cached_tokens is not None + else "unknown" + ), + } + + +def _telemetry_record( + *, + request_id: str, + kind: str, + model: str, + profile: str, + capacity: int, + capacity_source: str, + started: float, + queue_wait_ms: float, + admission_slot: int | None, + retry_index: int | None, + outcome: str, + finish_reason: str | None = None, + response_data: dict[str, Any] | None = None, + reason_code: str | None = None, +) -> dict[str, Any]: + record: dict[str, Any] = { + "timestamp": time.time(), + "request_id": request_id, + "kind": kind, + "provider": "local", + "model": model, + "profile": profile, + "serving_capacity": capacity, + "capacity_source": capacity_source, + "admission_slot": admission_slot, + "queue_wait_ms": round(queue_wait_ms, 3), + "client_total_ms": round((time.monotonic() - started) * 1000.0, 3), + "retry_index": retry_index, + "outcome": outcome, + "finish_reason": finish_reason, + "reason_code": reason_code, + "timed_out": outcome == "timeout", + "cancelled": outcome == "cancelled", + } + if response_data is not None: + record.update(_server_response_fields(response_data)) + return record + + def _classify_byo_generate_error(exc: BaseException) -> LocalProviderError: import httpx @@ -320,6 +419,66 @@ def _classify_byo_generate_error(exc: BaseException) -> LocalProviderError: ) +def _remaining_timeout(started: float, timeout_s: float) -> float: + remaining = timeout_s - (time.monotonic() - started) + if remaining <= 0: + from solstone.think.providers.local_admission import LocalAdmissionTimeout + + raise LocalAdmissionTimeout( + f"Local inference request exceeded its {timeout_s:.3f}s deadline." + ) + return remaining + + +def _prepare_bundled_request( + *, + server: Any, + contents: str | list[Any], + system_instruction: str | None, + temperature: float, + max_output_tokens: int, + json_output: bool, + json_schema: dict | None, +) -> tuple[dict[str, Any], dict[str, Any] | None]: + from solstone.think.providers import local_budget + + def counter(text: str) -> int: + return local_budget.count_tokens(text, server.base_url) + + fitted_contents, input_budget = local_budget.fit_contents( + contents, + system_instruction, + max_output_tokens, + count=counter, + ) + messages = _build_messages(fitted_contents, system_instruction) + return ( + _build_request_body( + server.served_model_id, + messages, + temperature, + max_output_tokens, + json_output, + json_schema, + True, + ), + input_budget, + ) + + +def _raise_bundled_status(response: Any) -> None: + import httpx + + try: + response.raise_for_status() + except httpx.HTTPStatusError as exc: + if _contains_any(response.text.lower(), _CONTEXT_WINDOW_PATTERNS): + raise ContextBudgetExceeded( + "Local request exceeded the model context window after fitting." + ) from exc + raise + + def run_generate( contents: str | list[Any], model: str, @@ -332,56 +491,106 @@ def run_generate( timeout_s: float | None = None, **kwargs: Any, ) -> GenerateResult: - del thinking_budget, kwargs + del thinking_budget + retry_index = int(kwargs.pop("inference_retry_index", 0) or 0) endpoint = resolve_local_endpoint() # Validate the requested logical id; served id comes from the server. normalize_model_id(model) messages = _build_messages(contents, system_instruction) if endpoint.is_bundled: from solstone.think.providers import local_server + from solstone.think.providers.local_admission import ( + LocalAdmissionTimeout, + acquire_local_slot, + record_local_inference, + ) + started = time.monotonic() + request_id = uuid.uuid4().hex + timeout = timeout_s or _DEFAULT_TIMEOUT server = local_server.connect() - from solstone.think.providers import local_budget - - def counter(text: str) -> int: - return local_budget.count_tokens(text, server.base_url) - - contents, input_budget = local_budget.fit_contents( - contents, - system_instruction, - max_output_tokens, - count=counter, - ) - messages = _build_messages(contents, system_instruction) - body = _build_request_body( - server.served_model_id, - messages, - temperature, - max_output_tokens, - json_output, - json_schema, - endpoint.is_bundled, + capacity = local_server.read_server_capacity() + body, input_budget = _prepare_bundled_request( + server=server, + contents=contents, + system_instruction=system_instruction, + temperature=temperature, + max_output_tokens=max_output_tokens, + json_output=json_output, + json_schema=json_schema, ) import httpx - response = httpx.post( - f"{server.base_url}/v1/chat/completions", - json=body, - timeout=timeout_s or _DEFAULT_TIMEOUT, - ) + permit = None try: - response.raise_for_status() - except httpx.HTTPStatusError as exc: - if _contains_any(response.text.lower(), _CONTEXT_WINDOW_PATTERNS): - raise ContextBudgetExceeded( - "Local request exceeded the model context window after fitting." - ) from exc + permit = acquire_local_slot( + capacity.parallel_slots, + _remaining_timeout(started, timeout), + ) + with permit: + response = httpx.post( + f"{server.base_url}/v1/chat/completions", + json=body, + timeout=_remaining_timeout(started, timeout), + ) + _raise_bundled_status(response) + response_data = response.json() + result = _parse_response(response_data) + telemetry = _telemetry_record( + request_id=request_id, + kind="generate", + model=LOCAL_MODEL, + profile=capacity.profile, + capacity=capacity.parallel_slots, + capacity_source=capacity.source, + started=started, + queue_wait_ms=permit.queue_wait_ms, + admission_slot=permit.slot_index, + retry_index=retry_index, + outcome="success", + finish_reason=result.get("finish_reason"), + response_data=response_data, + ) + record_local_inference(telemetry) + result["inference"] = telemetry + if input_budget is not None: + result["input_budget"] = input_budget + return result + except BaseException as exc: + if isinstance(exc, KeyboardInterrupt | SystemExit): + raise + if permit is not None: + permit.release() + outcome = ( + "timeout" + if isinstance(exc, (LocalAdmissionTimeout, httpx.TimeoutException)) + else "cancelled" + if isinstance(exc, asyncio.CancelledError) + else "error" + ) + record_local_inference( + _telemetry_record( + request_id=request_id, + kind="generate", + model=LOCAL_MODEL, + profile=capacity.profile, + capacity=capacity.parallel_slots, + capacity_source=capacity.source, + started=started, + queue_wait_ms=( + permit.queue_wait_ms + if permit is not None + else (time.monotonic() - started) * 1000.0 + ), + admission_slot=permit.slot_index if permit is not None else None, + retry_index=retry_index, + outcome=outcome, + reason_code=getattr(exc, "reason_code", None) + or classify_provider_error(exc, "local"), + ) + ) raise - result = _parse_response(response.json()) - if input_budget is not None: - result["input_budget"] = input_budget - return result body = _build_request_body( endpoint.served_model_id, @@ -424,50 +633,192 @@ async def run_agenerate( timeout_s: float | None = None, **kwargs: Any, ) -> GenerateResult: - return await asyncio.to_thread( - run_generate, - contents, - model, - temperature, - max_output_tokens, - system_instruction, - json_output, - thinking_budget, - json_schema, - timeout_s, - **kwargs, + del thinking_budget + retry_index = int(kwargs.pop("inference_retry_index", 0) or 0) + endpoint = resolve_local_endpoint() + normalize_model_id(model) + messages = _build_messages(contents, system_instruction) + + import httpx + + if not endpoint.is_bundled: + body = _build_request_body( + endpoint.served_model_id, + messages, + temperature, + max_output_tokens, + json_output, + json_schema, + False, + ) + post_kwargs: dict[str, Any] = { + "json": body, + "timeout": timeout_s or _DEFAULT_TIMEOUT, + } + if endpoint.credential: + post_kwargs["headers"] = {"Authorization": f"Bearer {endpoint.credential}"} + try: + async with httpx.AsyncClient() as client: + response = await client.post( + f"{endpoint.base_url}/v1/chat/completions", **post_kwargs + ) + response.raise_for_status() + return _parse_response(response.json()) + except asyncio.CancelledError: + raise + except Exception as exc: + raise _classify_byo_generate_error(exc) from exc + + from solstone.think.providers import local_server + from solstone.think.providers.local_admission import ( + LocalAdmissionTimeout, + acquire_local_slot_async, + record_local_inference, ) + started = time.monotonic() + request_id = uuid.uuid4().hex + timeout = timeout_s or _DEFAULT_TIMEOUT + server = local_server.connect() + capacity = local_server.read_server_capacity() + body, input_budget = await asyncio.to_thread( + _prepare_bundled_request, + server=server, + contents=contents, + system_instruction=system_instruction, + temperature=temperature, + max_output_tokens=max_output_tokens, + json_output=json_output, + json_schema=json_schema, + ) + permit = None + try: + permit = await acquire_local_slot_async( + capacity.parallel_slots, + _remaining_timeout(started, timeout), + ) + async with permit: + async with httpx.AsyncClient() as client: + response = await client.post( + f"{server.base_url}/v1/chat/completions", + json=body, + timeout=_remaining_timeout(started, timeout), + ) + _raise_bundled_status(response) + response_data = response.json() + result = _parse_response(response_data) + telemetry = _telemetry_record( + request_id=request_id, + kind="generate", + model=LOCAL_MODEL, + profile=capacity.profile, + capacity=capacity.parallel_slots, + capacity_source=capacity.source, + started=started, + queue_wait_ms=permit.queue_wait_ms, + admission_slot=permit.slot_index, + retry_index=retry_index, + outcome="success", + finish_reason=result.get("finish_reason"), + response_data=response_data, + ) + record_local_inference(telemetry) + result["inference"] = telemetry + if input_budget is not None: + result["input_budget"] = input_budget + return result + except BaseException as exc: + if isinstance(exc, KeyboardInterrupt | SystemExit): + raise + if permit is not None: + permit.release() + outcome = ( + "cancelled" + if isinstance(exc, asyncio.CancelledError) + else "timeout" + if isinstance(exc, (LocalAdmissionTimeout, httpx.TimeoutException)) + else "error" + ) + record_local_inference( + _telemetry_record( + request_id=request_id, + kind="generate", + model=LOCAL_MODEL, + profile=capacity.profile, + capacity=capacity.parallel_slots, + capacity_source=capacity.source, + started=started, + queue_wait_ms=( + permit.queue_wait_ms + if permit is not None + else (time.monotonic() - started) * 1000.0 + ), + admission_slot=permit.slot_index if permit is not None else None, + retry_index=retry_index, + outcome=outcome, + reason_code=getattr(exc, "reason_code", None) + or classify_provider_error(exc, "local"), + ) + ) + raise + async def run_cogitate( config: dict[str, Any], on_event: Callable[[dict], None] | None = None, ) -> str: from solstone.think.providers import local_server, openhands + from solstone.think.providers.local_admission import ( + LocalAdmissionTimeout, + acquire_local_slot_async, + record_local_inference, + ) config = {**config, "model": normalize_model_id(config.get("model", LOCAL_MODEL))} endpoint = resolve_local_endpoint() + started = time.monotonic() + request_id = uuid.uuid4().hex + server = None + capacity = None + permit = None + outcome = "success" + reason_code: str | None = None try: if endpoint.is_bundled: - local_server.connect() + server = local_server.connect() + capacity = local_server.read_server_capacity() + timeout = float(config.get("timeout_seconds", 600) or 600) + permit = await acquire_local_slot_async( + capacity.parallel_slots, + _remaining_timeout(started, timeout), + ) return await openhands.run_cogitate(config, on_event=on_event) + except asyncio.CancelledError: + outcome = "cancelled" + reason_code = "cancelled" + raise except Exception as exc: + outcome = ( + "timeout" + if isinstance(exc, LocalAdmissionTimeout) + or getattr(exc, "reason_code", None) == "wall_clock_exceeded" + else "error" + ) from solstone.think.talents import TalentHookError if isinstance(exc, TalentHookError): raise - reason_code = None if not endpoint.is_bundled: reason_code = classify_byo_cogitate_error(exc) or getattr( exc, "reason_code", None ) + reason_code = ( + reason_code + or getattr(exc, "reason_code", None) + or classify_provider_error(exc, "local") + ) if on_event and not getattr(exc, "_evented", False): - reason_code = ( - reason_code - or getattr(exc, "reason_code", None) - or classify_provider_error(exc, "local") - ) error_text = str(exc) trace_text = traceback.format_exc() fixed_copy = local_endpoint_reason_copy(reason_code) @@ -493,6 +844,31 @@ async def run_cogitate( setattr(wrapped, "_evented", getattr(exc, "_evented", False)) raise wrapped from exc raise + finally: + if permit is not None: + permit.release() + if server is not None and capacity is not None: + record_local_inference( + _telemetry_record( + request_id=request_id, + kind="cogitate", + model=LOCAL_MODEL, + profile=capacity.profile, + capacity=capacity.parallel_slots, + capacity_source=capacity.source, + started=started, + queue_wait_ms=( + permit.queue_wait_ms + if permit is not None + else (time.monotonic() - started) * 1000.0 + ), + admission_slot=permit.slot_index if permit is not None else None, + retry_index=None, + outcome=outcome, + finish_reason="stop" if outcome == "success" else None, + reason_code=reason_code, + ) + ) def list_models(provider: str = "local") -> list[dict[str, Any]]: diff --git a/solstone/think/providers/local_admission.py b/solstone/think/providers/local_admission.py new file mode 100644 index 000000000..df14d735c --- /dev/null +++ b/solstone/think/providers/local_admission.py @@ -0,0 +1,220 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Cross-process admission and content-free telemetry for local inference.""" + +from __future__ import annotations + +import asyncio +import errno +import fcntl +import logging +import os +import time +import uuid +from dataclasses import dataclass +from pathlib import Path +from typing import IO, Any + +from solstone.think.journal_io import append_jsonl +from solstone.think.utils import get_journal + +LOG = logging.getLogger(__name__) + +_POLL_INTERVAL_S = 0.025 + + +class LocalAdmissionTimeout(TimeoutError): + """No bundled-local inference slot became available before the deadline.""" + + reason_code = "local_queue_timeout" + + +@dataclass +class LocalPermit: + """One flock-backed serving-capacity permit.""" + + slot_index: int + capacity: int + queue_wait_ms: float + _file: IO[str] + + def release(self) -> None: + if self._file.closed: + return + try: + fcntl.flock(self._file, fcntl.LOCK_UN) + finally: + self._file.close() + + def __enter__(self) -> LocalPermit: + return self + + def __exit__(self, exc_type: Any, exc: Any, tb: Any) -> None: + self.release() + + async def __aenter__(self) -> LocalPermit: + return self + + async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None: + self.release() + + +def _admission_dir() -> Path: + return Path(get_journal()) / "health" / "local-inference-admission" + + +@dataclass +class _WaitTicket: + path: Path + file: IO[str] + + +def _create_ticket(root: Path) -> _WaitTicket: + root.mkdir(parents=True, exist_ok=True) + identity = f"{time.monotonic_ns():020d}-{os.getpid()}-{uuid.uuid4().hex}" + path = root / f"wait-{identity}.ticket" + creating_path = root / f".creating-{identity}.ticket" + ticket_file = open(creating_path, "x+", encoding="utf-8") + fcntl.flock(ticket_file, fcntl.LOCK_EX | fcntl.LOCK_NB) + creating_path.rename(path) + return _WaitTicket(path=path, file=ticket_file) + + +def _drop_ticket(ticket: _WaitTicket) -> None: + try: + ticket.path.unlink(missing_ok=True) + finally: + try: + fcntl.flock(ticket.file, fcntl.LOCK_UN) + finally: + ticket.file.close() + + +def _ticket_has_turn(root: Path, ticket: _WaitTicket) -> bool: + """Return whether ticket is oldest, pruning tickets whose owners exited.""" + while True: + waiting = sorted(root.glob("wait-*.ticket")) + if not waiting: + return False + first = waiting[0] + if first == ticket.path: + return True + try: + stale_file = open(first, "a+", encoding="utf-8") + except FileNotFoundError: + continue + try: + try: + fcntl.flock(stale_file, fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError as exc: + if exc.errno in (errno.EACCES, errno.EAGAIN, errno.EWOULDBLOCK): + return False + raise + first.unlink(missing_ok=True) + finally: + stale_file.close() + + +def _try_acquire(capacity: int, started: float, root: Path) -> LocalPermit | None: + for slot_index in range(capacity): + lock_file = open(root / f"slot-{slot_index}.lock", "a+", encoding="utf-8") + try: + fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError as exc: + lock_file.close() + if exc.errno in (errno.EACCES, errno.EAGAIN, errno.EWOULDBLOCK): + continue + raise + return LocalPermit( + slot_index=slot_index, + capacity=capacity, + queue_wait_ms=(time.monotonic() - started) * 1000.0, + _file=lock_file, + ) + return None + + +def _deadline(started: float, timeout_s: float | None) -> float | None: + if timeout_s is None: + return None + return started + max(0.0, timeout_s) + + +def acquire_local_slot(capacity: int, timeout_s: float | None) -> LocalPermit: + """Wait synchronously for one bundled-local serving slot.""" + if capacity < 1: + raise ValueError("local inference capacity must be at least one") + started = time.monotonic() + deadline = _deadline(started, timeout_s) + root = _admission_dir() + ticket = _create_ticket(root) + try: + while True: + if _ticket_has_turn(root, ticket): + permit = _try_acquire(capacity, started, root) + if permit is not None: + return permit + now = time.monotonic() + if deadline is not None and now >= deadline: + raise LocalAdmissionTimeout( + f"Local inference queue exceeded its {timeout_s:.3f}s deadline." + ) + sleep_s = _POLL_INTERVAL_S + if deadline is not None: + sleep_s = min(sleep_s, max(0.0, deadline - now)) + time.sleep(sleep_s) + finally: + _drop_ticket(ticket) + + +async def acquire_local_slot_async( + capacity: int, timeout_s: float | None +) -> LocalPermit: + """Wait cancellation-safely for one bundled-local serving slot.""" + if capacity < 1: + raise ValueError("local inference capacity must be at least one") + started = time.monotonic() + deadline = _deadline(started, timeout_s) + root = _admission_dir() + ticket = _create_ticket(root) + try: + while True: + if _ticket_has_turn(root, ticket): + permit = _try_acquire(capacity, started, root) + if permit is not None: + return permit + now = time.monotonic() + if deadline is not None and now >= deadline: + raise LocalAdmissionTimeout( + f"Local inference queue exceeded its {timeout_s:.3f}s deadline." + ) + sleep_s = _POLL_INTERVAL_S + if deadline is not None: + sleep_s = min(sleep_s, max(0.0, deadline - now)) + await asyncio.sleep(sleep_s) + finally: + _drop_ticket(ticket) + + +def record_local_inference(record: dict[str, Any]) -> None: + """Durably append one prompt/output-free local inference record.""" + try: + path = ( + Path(get_journal()) + / "health" + / "local-inference" + / f"{time.strftime('%Y%m%d')}.jsonl" + ) + append_jsonl(path, record) + except Exception: + LOG.warning("failed to record local inference telemetry", exc_info=True) + + +__all__ = [ + "LocalAdmissionTimeout", + "LocalPermit", + "acquire_local_slot", + "acquire_local_slot_async", + "record_local_inference", +] diff --git a/solstone/think/providers/local_server.py b/solstone/think/providers/local_server.py index 4b2809435..653c7cda2 100644 --- a/solstone/think/providers/local_server.py +++ b/solstone/think/providers/local_server.py @@ -6,6 +6,7 @@ from __future__ import annotations import logging +import sys from dataclasses import dataclass from pathlib import Path from typing import Any @@ -65,6 +66,16 @@ class LocalServerInfo: binary_path: str | None = None model_path: str | None = None served_model_id: str = LOCAL_MODEL + parallel_slots: int = 1 + capacity_source: str = "default" + profile: str = "floor" + + +@dataclass(frozen=True) +class ServerCapacity: + parallel_slots: int + source: str + profile: str def _base_url(port: int) -> str: @@ -166,7 +177,7 @@ def read_local_context_window() -> int | None: # server that never launched has no slots at all. _UNKNOWN_SLOTS = 1 -_PARALLEL_SLOTS_CACHE: int | None = None +_SERVER_CAPACITY_CACHE: ServerCapacity | None = None def _extract_total_slots(props: dict[str, Any]) -> int | None: @@ -186,13 +197,23 @@ def _slots_from_launched_tier(context_tokens: int | None) -> int | None: return None -def _discover_parallel_slots() -> int: +def _profile_for_slots(slots: int) -> str: + if sys.platform == "darwin": + return "apple" + if slots == _CAPABLE_TIER.parallel_slots: + return _CAPABLE_TIER.name + if slots == _FLOOR_TIER.parallel_slots: + return _FLOOR_TIER.name + return "advertised" + + +def _discover_server_capacity() -> ServerCapacity: port = read_service_port(_SERVICE_NAME) props = fetch_props(port) if port is not None else None if props is not None: slots = _extract_total_slots(props) if slots is not None: - return slots + return ServerCapacity(slots, "props", _profile_for_slots(slots)) context_tokens = read_local_context_window() slots = _slots_from_launched_tier(context_tokens) @@ -206,7 +227,15 @@ def _discover_parallel_slots() -> int: context_tokens, source, ) - return slots + return ServerCapacity(slots, source, _profile_for_slots(slots)) + + +def read_server_capacity() -> ServerCapacity: + """Return the memoized bundled-server capacity and its evidence source.""" + global _SERVER_CAPACITY_CACHE + if _SERVER_CAPACITY_CACHE is None: + _SERVER_CAPACITY_CACHE = _discover_server_capacity() + return _SERVER_CAPACITY_CACHE def read_server_parallel_slots() -> int: @@ -216,16 +245,13 @@ def read_server_parallel_slots() -> int: the context window persisted at launch, then to ``_UNKNOWN_SLOTS``. Discovered once per process. Never raises; always returns >= 1. """ - global _PARALLEL_SLOTS_CACHE - if _PARALLEL_SLOTS_CACHE is None: - _PARALLEL_SLOTS_CACHE = _discover_parallel_slots() - return _PARALLEL_SLOTS_CACHE + return read_server_capacity().parallel_slots def reset_parallel_slots_cache() -> None: """Clear the per-process slot-discovery memo.""" - global _PARALLEL_SLOTS_CACHE - _PARALLEL_SLOTS_CACHE = None + global _SERVER_CAPACITY_CACHE + _SERVER_CAPACITY_CACHE = None def _resolve_served_model_id(health_body: dict[str, Any] | None) -> str | None: @@ -263,12 +289,16 @@ def connect() -> LocalServerInfo: served_model_id = _resolve_served_model_id(body) if served_model_id is None: raise LocalProviderError("local_model_not_ready", LOCAL_MODEL_NOT_READY_COPY) + capacity = read_server_capacity() return LocalServerInfo( model_id=LOCAL_MODEL, port=port, base_url=_base_url(port), state=STATE_READY, served_model_id=served_model_id, + parallel_slots=capacity.parallel_slots, + capacity_source=capacity.source, + profile=capacity.profile, ) @@ -276,6 +306,7 @@ __all__ = [ "LOCAL_MIN_CONTEXT_TOKENS", "LOCAL_MODEL_NOT_READY_COPY", "LocalServerInfo", + "ServerCapacity", "ServerTier", "STATE_IDLE", "STATE_STARTING", @@ -289,6 +320,7 @@ __all__ = [ "probe_state", "read_local_context_window", "read_server_context_window", + "read_server_capacity", "read_server_parallel_slots", "reset_parallel_slots_cache", "select_server_tier", diff --git a/solstone/think/providers/shared.py b/solstone/think/providers/shared.py index 4b35f7ade..e67a74516 100644 --- a/solstone/think/providers/shared.py +++ b/solstone/think/providers/shared.py @@ -430,6 +430,7 @@ class GenerateResult(TypedDict, total=False): input_budget: Optional[ dict ] # Out-of-band truncation metadata when the bundled-local input was clipped + inference: Optional[dict] # Content-free local inference timing/admission record # --------------------------------------------------------------------------- diff --git a/solstone/think/talents.py b/solstone/think/talents.py index 54e185ee5..5728219ae 100644 --- a/solstone/think/talents.py +++ b/solstone/think/talents.py @@ -1612,6 +1612,7 @@ async def _execute_generate( timeout_s=timeout_s, provider=config.get("provider"), model=config.get("model"), + inference_retry_index=1, ) else: if config.get("fallback_from") or not _should_fallback(exc): diff --git a/tests/test_local.py b/tests/test_local.py index cb013281c..b5cedc1a3 100644 --- a/tests/test_local.py +++ b/tests/test_local.py @@ -27,6 +27,18 @@ from solstone.think.models import ( from solstone.think.talents import TalentHookError +@pytest.fixture(autouse=True) +def _isolate_local_admission(monkeypatch, tmp_path): + from solstone.think.providers import local_admission + + monkeypatch.setattr( + local_admission, + "_admission_dir", + lambda: tmp_path / "local-inference-admission", + ) + monkeypatch.setattr(local_admission, "record_local_inference", lambda _record: None) + + def _provider(): providers_pkg = importlib.import_module("solstone.think.providers") if hasattr(providers_pkg, "local_budget"): @@ -1629,9 +1641,12 @@ def test_local_server_connect_linux_health_shape_uses_logical_model(monkeypatch) def _local_journal(monkeypatch, tmp_path: Path) -> Path: + from solstone.think.providers import local_server + journal = tmp_path / "journal" (journal / "health").mkdir(parents=True) monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal)) + local_server.reset_parallel_slots_cache() return journal @@ -1652,6 +1667,11 @@ def test_read_server_parallel_slots_prefers_live_props(monkeypatch, tmp_path): # /props is ground truth; it wins over the persisted tier. assert local_server.read_server_parallel_slots() == 2 + assert local_server.read_server_capacity() == local_server.ServerCapacity( + parallel_slots=2, + source="props", + profile="capable", + ) def test_read_server_parallel_slots_no_port_returns_floor( @@ -1674,6 +1694,19 @@ def test_read_server_parallel_slots_no_port_returns_floor( ) +def test_server_capacity_uses_explicit_apple_profile(monkeypatch, tmp_path): + from solstone.think.providers import local_server + + _local_journal(monkeypatch, tmp_path) + monkeypatch.setattr(local_server.sys, "platform", "darwin") + + assert local_server.read_server_capacity() == local_server.ServerCapacity( + parallel_slots=1, + source="default", + profile="apple", + ) + + @pytest.mark.parametrize("slots", [1, 2]) def test_read_server_parallel_slots_falls_back_to_launched_tier( monkeypatch, tmp_path, slots @@ -1735,3 +1768,50 @@ def test_read_server_parallel_slots_is_memoized_and_resettable(monkeypatch, tmp_ local_server.reset_parallel_slots_cache() assert local_server.read_server_parallel_slots() == 2 assert len(calls) == 2 + + +def test_local_response_telemetry_normalizes_llama_and_mlx_fields(): + provider = _provider() + + fields = provider._server_response_fields( + { + "timings": { + "cache_n": 236, + "prompt_n": 12, + "prompt_ms": 30.5, + "predicted_n": 35, + "predicted_ms": 661.0, + "slot_id": 1, + }, + "usage": { + "prompt_tokens": 248, + "completion_tokens": 35, + }, + } + ) + + assert fields == { + "prompt_eval_ms": 30.5, + "generation_ms": 661.0, + "server_total_ms": 691.5, + "prompt_tokens": 248, + "generated_tokens": 35, + "prompt_cached_tokens": 236, + "selected_slot": 1, + "prompt_cache_state": "warm", + } + assert provider._extract_usage( + { + "usage": { + "prompt_tokens": 248, + "completion_tokens": 35, + "total_tokens": 283, + "prompt_tokens_details": {"cached_tokens": 236}, + } + } + ) == { + "input_tokens": 248, + "output_tokens": 35, + "total_tokens": 283, + "cached_tokens": 236, + } diff --git a/tests/test_local_admission.py b/tests/test_local_admission.py new file mode 100644 index 000000000..2c67593e6 --- /dev/null +++ b/tests/test_local_admission.py @@ -0,0 +1,150 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +import asyncio +import json +import threading +import time + +import pytest + +from solstone.think.providers.local_admission import ( + LocalAdmissionTimeout, + acquire_local_slot, + acquire_local_slot_async, + record_local_inference, +) + + +def _isolated_journal(monkeypatch, tmp_path): + import solstone.think.utils as think_utils + + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) + think_utils._journal_path_cache = None + + +def test_cross_thread_admission_never_exceeds_capacity(monkeypatch, tmp_path): + _isolated_journal(monkeypatch, tmp_path) + active = 0 + peak = 0 + lock = threading.Lock() + + def work() -> None: + nonlocal active, peak + with acquire_local_slot(2, 2): + with lock: + active += 1 + peak = max(peak, active) + time.sleep(0.04) + with lock: + active -= 1 + + threads = [threading.Thread(target=work) for _ in range(8)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + assert peak == 2 + + +def test_failure_releases_permit(monkeypatch, tmp_path): + _isolated_journal(monkeypatch, tmp_path) + + with pytest.raises(RuntimeError, match="boom"): + with acquire_local_slot(1, 0.5): + raise RuntimeError("boom") + + with acquire_local_slot(1, 0.1) as permit: + assert permit.slot_index == 0 + + +def test_sync_queue_timeout(monkeypatch, tmp_path): + _isolated_journal(monkeypatch, tmp_path) + first = acquire_local_slot(1, 0.1) + try: + with pytest.raises(LocalAdmissionTimeout): + acquire_local_slot(1, 0.03) + finally: + first.release() + + +def test_async_queued_cancellation_does_not_leak(monkeypatch, tmp_path): + _isolated_journal(monkeypatch, tmp_path) + + async def exercise() -> None: + first = await acquire_local_slot_async(1, 0.5) + queued = asyncio.create_task(acquire_local_slot_async(1, 1.0)) + await asyncio.sleep(0.05) + queued.cancel() + with pytest.raises(asyncio.CancelledError): + await queued + assert not list( + (tmp_path / "health" / "local-inference-admission").glob("wait-*.ticket") + ) + first.release() + second = await acquire_local_slot_async(1, 0.1) + second.release() + + asyncio.run(exercise()) + + +def test_waiters_are_admitted_in_ticket_order(monkeypatch, tmp_path): + _isolated_journal(monkeypatch, tmp_path) + root = tmp_path / "health" / "local-inference-admission" + first = acquire_local_slot(1, 1) + order: list[int] = [] + + def wait(index: int) -> None: + with acquire_local_slot(1, 2): + order.append(index) + time.sleep(0.01) + + threads = [] + for index in range(5): + thread = threading.Thread(target=wait, args=(index,)) + thread.start() + threads.append(thread) + deadline = time.monotonic() + 1 + while len(list(root.glob("wait-*.ticket"))) < index + 1: + assert time.monotonic() < deadline + time.sleep(0.005) + + first.release() + for thread in threads: + thread.join() + + assert order == list(range(5)) + + +def test_stale_ticket_from_exited_owner_is_pruned(monkeypatch, tmp_path): + _isolated_journal(monkeypatch, tmp_path) + root = tmp_path / "health" / "local-inference-admission" + root.mkdir(parents=True) + stale = root / "wait-00000000000000000000-1-stale.ticket" + stale.write_text("", encoding="utf-8") + + with acquire_local_slot(1, 0.2): + assert not stale.exists() + + +def test_telemetry_is_durable_and_content_free(monkeypatch, tmp_path): + _isolated_journal(monkeypatch, tmp_path) + record_local_inference( + { + "timestamp": 1.0, + "request_id": "abc", + "provider": "local", + "model": "local/qwen3.5-4b", + "queue_wait_ms": 12.5, + "outcome": "success", + } + ) + + path = tmp_path / "health" / "local-inference" / time.strftime("%Y%m%d.jsonl") + row = json.loads(path.read_text(encoding="utf-8")) + assert row["request_id"] == "abc" + assert "prompt" not in row + assert "output" not in row diff --git a/tests/test_log_retention.py b/tests/test_log_retention.py index 6c15ad092..d874935c9 100644 --- a/tests/test_log_retention.py +++ b/tests/test_log_retention.py @@ -121,6 +121,19 @@ def test_ac3_disabled_config_deletes_nothing_and_writes_no_audit(journal): assert not (journal / "chronicle" / old_day / "task_log.txt").exists() +def test_local_inference_telemetry_follows_journal_log_retention(journal): + old_day = _day(31) + recent_day = _day(1) + old = _write(journal / "health" / "local-inference" / f"{old_day}.jsonl") + recent = _write(journal / "health" / "local-inference" / f"{recent_day}.jsonl") + + result = prune(config=LogRetentionConfig(days=30)) + + assert not old.exists() + assert recent.exists() + assert result.by_class["local_inference"]["files_deleted"] == 1 + + def test_ac4_dry_run_reports_candidates_without_deleting_or_audit(journal): old_day = _day(31) old_token = _write(journal / "tokens" / f"{old_day}.jsonl", "old")