solpbc.org
personal memory agent
feat(local): add slot-aware inference admission telemetry
author
Jer Miller
date
(Jul 11, 2026, 6:07 AM -0600)
commit
dfd97994
dfd97994758ed80241bf4a9b86af6ddbc9f51815
parent
a61ab722
a61ab722437a20521316727ab7166a55f34b1f20
+1311
-67
+2
-1
AGENTS.md
+2
-1
AGENTS.md
···
199
199
| Config (`config/journal.json`) | `solstone/think/journal_config.py` |
200
200
| Schedules (`config/schedules.json`) | `solstone/think/schedule_config.py` |
201
201
| Push devices (`config/push_devices.json`) | `solstone/think/push/devices.py` |
202
+
| Local inference operational telemetry (`health/local-inference/YYYYMMDD.jsonl`) | `solstone/think/providers/local_admission.py` |
202
203
| Hosted backup binding (`backup/hosted/binding.json`) | `solstone/think/backup/hosted.py` |
203
204
| Convey config (`config/convey.json`) | `solstone/convey/config.py` + `solstone/think/facets.py` |
204
205
| Chat config (`config/chat.json`) | `solstone/apps/chat/config.py` |
···
211
212
| Index (SQLite, `indexer/*`) | `solstone/think/indexer/*` |
212
213
| Observer registry and sync history (`apps/observer/observers/*.json`, `apps/observer/observers/*/hist/*.jsonl`) | `solstone/apps/observer/utils.py` |
213
214
| 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/<id>/`. `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/<id>/` and `imports/<backend>.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/<id>/<name>` via journal_io's `install_file` primitive, as importer declared outputs (L7). |
214
-
| Operational-log/cache pruning and root task-log compaction — deletion/compaction only, across the dated allowlist (`chronicle/<day>/health/*.{log,jsonl}`, top-level `talents/<talent>/<epoch_ms>.jsonl` run logs + `talents/<YYYYMMDD>.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. |
215
+
| Operational-log/cache pruning and root task-log compaction — deletion/compaction only, across the dated allowlist (`chronicle/<day>/health/*.{log,jsonl}`, top-level `talents/<talent>/<epoch_ms>.jsonl` run logs + `talents/<YYYYMMDD>.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. |
215
216
216
217
If you're about to write to a domain from a module not in this table, stop and route through the owner.
217
218
+83
docs/PROVIDERS.md
+83
docs/PROVIDERS.md
···
224
224
- Return usage in `GenerateResult["usage"]` - wrapper handles logging
225
225
- For `run_cogitate()`, include usage in the `finish` event
226
226
227
+
## Bundled-local admission and inference telemetry
228
+
229
+
The `local` provider has one shared admission boundary for the supervisor-owned
230
+
Qwen server. It applies only when `providers.local` resolves to the bundled
231
+
loopback runtime. A configured OpenAI-compatible endpoint and every cloud
232
+
provider bypass this boundary.
233
+
234
+
Capacity remains explicit and intentionally small:
235
+
236
+
| Runtime profile | Serving capacity | Evidence |
237
+
|---|---:|---|
238
+
| Linux floor | 1 | supervisor `ServerTier`; live `/props.total_slots` wins |
239
+
| Linux capable (at least 16 GiB tiering VRAM) | 2 | supervisor `ServerTier`; live `/props.total_slots` wins |
240
+
| Apple MLX | 1 | conservative explicit fallback because mlx-vlm 0.6.2 does not advertise a slot limit |
241
+
242
+
The provider memoizes the capacity once per process. It first reads live
243
+
`/props.total_slots`, then the persisted `health/local.ctx` launch tier, then
244
+
falls back to one. The supervisor remains the configuration owner: changing a
245
+
Linux tier's `parallel_slots` changes both `llama-server --parallel` and provider
246
+
admission after the journal processes restart. Apple stays at one until that
247
+
runtime exposes a stable capacity contract and a separate measurement justifies
248
+
raising it.
249
+
250
+
Admission uses one `flock` file per slot under
251
+
`health/local-inference-admission/`. This coordinates independent journal
252
+
processes without a scheduler service or in-memory queue. Waiting async calls
253
+
are cancellation-safe; exceptions and cancellation release acquired locks;
254
+
process exit releases kernel locks. Queue time consumes the caller's existing
255
+
provider deadline, so waiting cannot silently extend a request beyond its
256
+
configured timeout. Cogitate holds one permit for its run because the OpenHands
257
+
SDK owns its internal multi-turn HTTP calls; this is conservative and avoids an
258
+
uncontrolled second path to the same server.
259
+
260
+
Every bundled-local attempt appends a content-free JSON record to
261
+
`health/local-inference/YYYYMMDD.jsonl`. These files follow the configured
262
+
`retention.journal_logs.days` policy. Records contain:
263
+
264
+
- request id, timestamp, generate/cogitate kind, provider, logical model, and
265
+
runtime profile;
266
+
- serving capacity, evidence source, admission slot, and client queue wait;
267
+
- client wall time plus server prompt-evaluation, generation, and total timing
268
+
when the response exposes them;
269
+
- prompt/generated token counts, reused prompt tokens, prompt-cache cold/warm
270
+
state, and selected server slot when exposed;
271
+
- retry index, finish reason, outcome, timeout/cancellation flags, and a safe
272
+
reason code on failure.
273
+
274
+
Records never contain prompt text, generated text, messages, schemas, images,
275
+
endpoint URLs, or credentials. A successful `GenerateResult` also carries the
276
+
same record as `inference`; callers that already retain the full result can use
277
+
it without rereading the log. Fields unavailable from a runtime remain null or
278
+
`unknown` rather than being inferred.
279
+
280
+
Run the synthetic journal-shaped benchmark against an isolated server:
281
+
282
+
```bash
283
+
python scripts/benchmark_local_inference_admission.py \
284
+
--endpoint http://127.0.0.1:8080 --slots 2 --concurrency 10 \
285
+
--requests 30 --mode baseline
286
+
python scripts/benchmark_local_inference_admission.py \
287
+
--endpoint http://127.0.0.1:8080 --slots 2 --concurrency 10 \
288
+
--requests 30 --mode admitted
289
+
```
290
+
291
+
It reports throughput, latency P50/P95/P99, explicit queue wait, residual opaque
292
+
wait, failures, and NVIDIA peak memory/utilization when `nvidia-smi` is present.
293
+
The payloads are fixed synthetic text/JSON requests and never read a journal.
294
+
295
+
The implementation gate on `fedora.local` used a fresh b9957 server with two
296
+
slots, ten producers, and twelve mixed requests for each side. Baseline versus
297
+
admitted results were 0.1291 versus 0.1296 requests/s, P95 latency 78.71 versus
298
+
77.22 seconds, P99 80.28 versus 77.30 seconds, and identical 4,652 MiB peak GPU
299
+
memory. Most importantly, P95 wait hidden inside the server fell from 65.56
300
+
seconds to 1.22 seconds; the admitted run reported 62.38 seconds as explicit
301
+
client queue wait. The boundary preserved throughput, slightly improved the
302
+
tail, and made the backlog observable without changing model work.
303
+
304
+
Rollback is one code revert plus a journal-process restart. The lock files hold
305
+
no state and may remain on disk; removing the provider calls to
306
+
`acquire_local_slot*()` immediately restores server-side queueing. Telemetry
307
+
JSONL files are ordinary operational logs and can remain until retention prunes
308
+
them.
309
+
227
310
## Context & Routing
228
311
229
312
Context strings determine provider and model selection. Providers receive already-resolved models, but understanding the system helps:
+275
scripts/benchmark_local_inference_admission.py
+275
scripts/benchmark_local_inference_admission.py
···
1
+
# SPDX-License-Identifier: AGPL-3.0-only
2
+
# Copyright (c) 2026 sol pbc
3
+
4
+
"""Reproducible synthetic mixed-load benchmark for bundled-local admission."""
5
+
6
+
from __future__ import annotations
7
+
8
+
import argparse
9
+
import concurrent.futures
10
+
import json
11
+
import os
12
+
import statistics
13
+
import subprocess
14
+
import tempfile
15
+
import threading
16
+
import time
17
+
from pathlib import Path
18
+
from typing import Any
19
+
20
+
import httpx
21
+
22
+
from solstone.think.providers.local_admission import acquire_local_slot
23
+
24
+
WORKLOADS = (
25
+
{
26
+
"name": "short_json",
27
+
"prompt": "Return JSON with keys summary and confidence about a synthetic note.",
28
+
"max_tokens": 48,
29
+
"response_format": {
30
+
"type": "json_schema",
31
+
"json_schema": {
32
+
"name": "short",
33
+
"strict": True,
34
+
"schema": {
35
+
"type": "object",
36
+
"properties": {
37
+
"summary": {"type": "string"},
38
+
"confidence": {"type": "number"},
39
+
},
40
+
"required": ["summary", "confidence"],
41
+
"additionalProperties": False,
42
+
},
43
+
},
44
+
},
45
+
},
46
+
{
47
+
"name": "entity_json",
48
+
"prompt": (
49
+
"From this synthetic project update, return JSON containing people and "
50
+
"projects arrays. Alex discussed Orion with Sam; no real data is present."
51
+
),
52
+
"max_tokens": 96,
53
+
"response_format": {
54
+
"type": "json_schema",
55
+
"json_schema": {
56
+
"name": "entities",
57
+
"strict": True,
58
+
"schema": {
59
+
"type": "object",
60
+
"properties": {
61
+
"people": {"type": "array", "items": {"type": "string"}},
62
+
"projects": {"type": "array", "items": {"type": "string"}},
63
+
},
64
+
"required": ["people", "projects"],
65
+
"additionalProperties": False,
66
+
},
67
+
},
68
+
},
69
+
},
70
+
{
71
+
"name": "brief_text",
72
+
"prompt": "Summarize a synthetic meeting in three concise bullet points.",
73
+
"max_tokens": 72,
74
+
},
75
+
)
76
+
77
+
78
+
def _percentile(values: list[float], percentile: float) -> float:
79
+
if not values:
80
+
return 0.0
81
+
ordered = sorted(values)
82
+
rank = (len(ordered) - 1) * percentile
83
+
lower = int(rank)
84
+
upper = min(lower + 1, len(ordered) - 1)
85
+
fraction = rank - lower
86
+
return ordered[lower] + (ordered[upper] - ordered[lower]) * fraction
87
+
88
+
89
+
def _distribution(values: list[float]) -> dict[str, float]:
90
+
return {
91
+
"p50": round(_percentile(values, 0.50), 3),
92
+
"p95": round(_percentile(values, 0.95), 3),
93
+
"p99": round(_percentile(values, 0.99), 3),
94
+
"mean": round(statistics.fmean(values), 3) if values else 0.0,
95
+
"max": round(max(values), 3) if values else 0.0,
96
+
}
97
+
98
+
99
+
def _server_ms(data: dict[str, Any]) -> float:
100
+
timings = data.get("timings")
101
+
if not isinstance(timings, dict):
102
+
return 0.0
103
+
return float(timings.get("prompt_ms") or 0) + float(
104
+
timings.get("predicted_ms") or 0
105
+
)
106
+
107
+
108
+
def _run_one(
109
+
*,
110
+
endpoint: str,
111
+
model: str,
112
+
workload: dict[str, Any],
113
+
slots: int,
114
+
admitted: bool,
115
+
timeout_s: float,
116
+
) -> dict[str, Any]:
117
+
started = time.monotonic()
118
+
permit = acquire_local_slot(slots, timeout_s) if admitted else None
119
+
queue_wait_ms = permit.queue_wait_ms if permit is not None else 0.0
120
+
try:
121
+
body = {
122
+
"model": model,
123
+
"messages": [{"role": "user", "content": workload["prompt"]}],
124
+
"max_tokens": workload["max_tokens"],
125
+
"temperature": 0.2,
126
+
"stream": False,
127
+
"chat_template_kwargs": {"enable_thinking": False},
128
+
}
129
+
if "response_format" in workload:
130
+
body["response_format"] = workload["response_format"]
131
+
response = httpx.post(
132
+
f"{endpoint.rstrip('/')}/v1/chat/completions",
133
+
json=body,
134
+
timeout=timeout_s,
135
+
)
136
+
response.raise_for_status()
137
+
data = response.json()
138
+
latency_ms = (time.monotonic() - started) * 1000.0
139
+
server_ms = _server_ms(data)
140
+
return {
141
+
"workload": workload["name"],
142
+
"ok": True,
143
+
"latency_ms": latency_ms,
144
+
"queue_wait_ms": queue_wait_ms,
145
+
"server_ms": server_ms,
146
+
"opaque_wait_ms": max(0.0, latency_ms - queue_wait_ms - server_ms),
147
+
}
148
+
except Exception as exc:
149
+
return {
150
+
"workload": workload["name"],
151
+
"ok": False,
152
+
"latency_ms": (time.monotonic() - started) * 1000.0,
153
+
"queue_wait_ms": queue_wait_ms,
154
+
"server_ms": 0.0,
155
+
"opaque_wait_ms": 0.0,
156
+
"error_type": type(exc).__name__,
157
+
}
158
+
finally:
159
+
if permit is not None:
160
+
permit.release()
161
+
162
+
163
+
def _sample_gpu(stop: threading.Event, samples: list[dict[str, float]]) -> None:
164
+
while not stop.wait(0.1):
165
+
try:
166
+
output = subprocess.run(
167
+
[
168
+
"nvidia-smi",
169
+
"--query-gpu=memory.used,utilization.gpu",
170
+
"--format=csv,noheader,nounits",
171
+
],
172
+
check=True,
173
+
capture_output=True,
174
+
text=True,
175
+
timeout=2,
176
+
).stdout.splitlines()[0]
177
+
memory_mib, utilization = (
178
+
float(item.strip()) for item in output.split(",")
179
+
)
180
+
samples.append(
181
+
{"gpu_memory_mib": memory_mib, "gpu_utilization_pct": utilization}
182
+
)
183
+
except (FileNotFoundError, IndexError, ValueError, subprocess.SubprocessError):
184
+
return
185
+
186
+
187
+
def _summarize(rows: list[dict[str, Any]]) -> dict[str, Any]:
188
+
successful = [row for row in rows if row["ok"]]
189
+
return {
190
+
"requests": len(rows),
191
+
"succeeded": len(successful),
192
+
"failed": len(rows) - len(successful),
193
+
"latency_ms": _distribution([row["latency_ms"] for row in successful]),
194
+
"queue_wait_ms": _distribution([row["queue_wait_ms"] for row in successful]),
195
+
"opaque_wait_ms": _distribution([row["opaque_wait_ms"] for row in successful]),
196
+
}
197
+
198
+
199
+
def main() -> None:
200
+
parser = argparse.ArgumentParser()
201
+
parser.add_argument("--endpoint", required=True)
202
+
parser.add_argument("--model", default="local/qwen3.5-4b")
203
+
parser.add_argument("--slots", type=int, required=True)
204
+
parser.add_argument("--concurrency", type=int, default=10)
205
+
parser.add_argument("--requests", type=int, default=30)
206
+
parser.add_argument("--timeout", type=float, default=300.0)
207
+
parser.add_argument("--mode", choices=("baseline", "admitted"), required=True)
208
+
parser.add_argument("--output", type=Path)
209
+
args = parser.parse_args()
210
+
if args.slots < 1 or args.concurrency < 1 or args.requests < 1:
211
+
parser.error("slots, concurrency, and requests must be positive")
212
+
213
+
with tempfile.TemporaryDirectory(prefix="solstone-admission-bench-") as state_dir:
214
+
os.environ["SOLSTONE_JOURNAL"] = state_dir
215
+
started = time.monotonic()
216
+
gpu_samples: list[dict[str, float]] = []
217
+
stop = threading.Event()
218
+
sampler = threading.Thread(
219
+
target=_sample_gpu, args=(stop, gpu_samples), daemon=True
220
+
)
221
+
sampler.start()
222
+
with concurrent.futures.ThreadPoolExecutor(
223
+
max_workers=args.concurrency
224
+
) as executor:
225
+
futures = [
226
+
executor.submit(
227
+
_run_one,
228
+
endpoint=args.endpoint,
229
+
model=args.model,
230
+
workload=WORKLOADS[index % len(WORKLOADS)],
231
+
slots=args.slots,
232
+
admitted=args.mode == "admitted",
233
+
timeout_s=args.timeout,
234
+
)
235
+
for index in range(args.requests)
236
+
]
237
+
rows = [future.result() for future in futures]
238
+
elapsed_s = time.monotonic() - started
239
+
stop.set()
240
+
sampler.join(timeout=1)
241
+
242
+
result = {
243
+
"mode": args.mode,
244
+
"endpoint": args.endpoint,
245
+
"model": args.model,
246
+
"configured_slots": args.slots,
247
+
"producer_concurrency": args.concurrency,
248
+
"elapsed_s": round(elapsed_s, 3),
249
+
"throughput_requests_per_s": round(args.requests / elapsed_s, 4),
250
+
"all": _summarize(rows),
251
+
"by_workload": {
252
+
workload["name"]: _summarize(
253
+
[row for row in rows if row["workload"] == workload["name"]]
254
+
)
255
+
for workload in WORKLOADS
256
+
},
257
+
"resources": {
258
+
"gpu_samples": len(gpu_samples),
259
+
"peak_gpu_memory_mib": max(
260
+
(sample["gpu_memory_mib"] for sample in gpu_samples), default=None
261
+
),
262
+
"peak_gpu_utilization_pct": max(
263
+
(sample["gpu_utilization_pct"] for sample in gpu_samples),
264
+
default=None,
265
+
),
266
+
},
267
+
}
268
+
rendered = json.dumps(result, indent=2, sort_keys=True)
269
+
if args.output:
270
+
args.output.write_text(rendered + "\n", encoding="utf-8")
271
+
print(rendered)
272
+
273
+
274
+
if __name__ == "__main__":
275
+
main()
+2
scripts/check_journal_io_access.py
+2
scripts/check_journal_io_access.py
···
116
116
"solstone/think/identity.py",
117
117
"solstone/think/journal_config.py",
118
118
"solstone/think/log_retention.py",
119
+
# Sole writer of content-free bundled-local inference telemetry.
120
+
"solstone/think/providers/local_admission.py",
119
121
"solstone/think/schedule_config.py",
120
122
"solstone/think/push/devices.py",
121
123
# Backup hosted-tier binding (0600 broker-token cache).
+10
solstone/think/log_retention.py
+10
solstone/think/log_retention.py
···
27
27
"talent_day_index",
28
28
"cogitate_history_cache",
29
29
"tokens",
30
+
"local_inference",
30
31
"awareness_logs",
31
32
"config_actions",
32
33
"facet_logs",
···
168
169
dry_run=dry_run,
169
170
class_name="tokens",
170
171
base=journal_path / "tokens",
172
+
pattern="*.jsonl",
173
+
)
174
+
_scan_dated_files(
175
+
journal_path,
176
+
cutoff,
177
+
result,
178
+
dry_run=dry_run,
179
+
class_name="local_inference",
180
+
base=journal_path / "health" / "local-inference",
171
181
pattern="*.jsonl",
172
182
)
173
183
_scan_dated_files(
+432
-56
solstone/think/providers/local.py
+432
-56
solstone/think/providers/local.py
···
12
12
import asyncio
13
13
import copy
14
14
import logging
15
+
import time
15
16
import traceback
17
+
import uuid
16
18
from collections.abc import Callable
17
19
from dataclasses import dataclass
18
20
from typing import Any
···
265
267
input_tokens = int(usage.get("prompt_tokens") or 0)
266
268
output_tokens = int(usage.get("completion_tokens") or 0)
267
269
total_tokens = int(usage.get("total_tokens") or input_tokens + output_tokens)
268
-
return {
270
+
normalized = {
269
271
"input_tokens": input_tokens,
270
272
"output_tokens": output_tokens,
271
273
"total_tokens": total_tokens,
272
274
}
275
+
prompt_details = usage.get("prompt_tokens_details")
276
+
if isinstance(prompt_details, dict):
277
+
cached_tokens = int(prompt_details.get("cached_tokens") or 0)
278
+
if cached_tokens:
279
+
normalized["cached_tokens"] = cached_tokens
280
+
return normalized
273
281
274
282
275
283
def _parse_response(data: dict[str, Any]) -> GenerateResult:
···
295
303
)
296
304
297
305
306
+
def _number(value: Any) -> int | float | None:
307
+
if isinstance(value, bool) or not isinstance(value, int | float):
308
+
return None
309
+
return value
310
+
311
+
312
+
def _server_response_fields(data: dict[str, Any]) -> dict[str, Any]:
313
+
"""Normalize content-free timing/cache/slot fields exposed by local servers."""
314
+
timings = data.get("timings")
315
+
timings = timings if isinstance(timings, dict) else {}
316
+
usage = data.get("usage")
317
+
usage = usage if isinstance(usage, dict) else {}
318
+
prompt_details = usage.get("prompt_tokens_details")
319
+
prompt_details = prompt_details if isinstance(prompt_details, dict) else {}
320
+
321
+
cached_tokens = _number(timings.get("cache_n"))
322
+
if cached_tokens is None:
323
+
cached_tokens = _number(prompt_details.get("cached_tokens"))
324
+
slot_id = _number(data.get("id_slot"))
325
+
if slot_id is None:
326
+
slot_id = _number(data.get("slot_id"))
327
+
if slot_id is None:
328
+
slot_id = _number(timings.get("slot_id"))
329
+
330
+
prompt_ms = _number(timings.get("prompt_ms"))
331
+
generation_ms = _number(timings.get("predicted_ms"))
332
+
server_total_ms = None
333
+
if prompt_ms is not None or generation_ms is not None:
334
+
server_total_ms = float(prompt_ms or 0) + float(generation_ms or 0)
335
+
336
+
return {
337
+
"prompt_eval_ms": prompt_ms,
338
+
"generation_ms": generation_ms,
339
+
"server_total_ms": server_total_ms,
340
+
"prompt_tokens": _number(usage.get("prompt_tokens"))
341
+
or _number(timings.get("prompt_n")),
342
+
"generated_tokens": _number(usage.get("completion_tokens"))
343
+
or _number(timings.get("predicted_n")),
344
+
"prompt_cached_tokens": cached_tokens,
345
+
"selected_slot": slot_id,
346
+
"prompt_cache_state": (
347
+
"warm"
348
+
if cached_tokens is not None and cached_tokens > 0
349
+
else "cold"
350
+
if cached_tokens is not None
351
+
else "unknown"
352
+
),
353
+
}
354
+
355
+
356
+
def _telemetry_record(
357
+
*,
358
+
request_id: str,
359
+
kind: str,
360
+
model: str,
361
+
profile: str,
362
+
capacity: int,
363
+
capacity_source: str,
364
+
started: float,
365
+
queue_wait_ms: float,
366
+
admission_slot: int | None,
367
+
retry_index: int | None,
368
+
outcome: str,
369
+
finish_reason: str | None = None,
370
+
response_data: dict[str, Any] | None = None,
371
+
reason_code: str | None = None,
372
+
) -> dict[str, Any]:
373
+
record: dict[str, Any] = {
374
+
"timestamp": time.time(),
375
+
"request_id": request_id,
376
+
"kind": kind,
377
+
"provider": "local",
378
+
"model": model,
379
+
"profile": profile,
380
+
"serving_capacity": capacity,
381
+
"capacity_source": capacity_source,
382
+
"admission_slot": admission_slot,
383
+
"queue_wait_ms": round(queue_wait_ms, 3),
384
+
"client_total_ms": round((time.monotonic() - started) * 1000.0, 3),
385
+
"retry_index": retry_index,
386
+
"outcome": outcome,
387
+
"finish_reason": finish_reason,
388
+
"reason_code": reason_code,
389
+
"timed_out": outcome == "timeout",
390
+
"cancelled": outcome == "cancelled",
391
+
}
392
+
if response_data is not None:
393
+
record.update(_server_response_fields(response_data))
394
+
return record
395
+
396
+
298
397
def _classify_byo_generate_error(exc: BaseException) -> LocalProviderError:
299
398
import httpx
300
399
···
320
419
)
321
420
322
421
422
+
def _remaining_timeout(started: float, timeout_s: float) -> float:
423
+
remaining = timeout_s - (time.monotonic() - started)
424
+
if remaining <= 0:
425
+
from solstone.think.providers.local_admission import LocalAdmissionTimeout
426
+
427
+
raise LocalAdmissionTimeout(
428
+
f"Local inference request exceeded its {timeout_s:.3f}s deadline."
429
+
)
430
+
return remaining
431
+
432
+
433
+
def _prepare_bundled_request(
434
+
*,
435
+
server: Any,
436
+
contents: str | list[Any],
437
+
system_instruction: str | None,
438
+
temperature: float,
439
+
max_output_tokens: int,
440
+
json_output: bool,
441
+
json_schema: dict | None,
442
+
) -> tuple[dict[str, Any], dict[str, Any] | None]:
443
+
from solstone.think.providers import local_budget
444
+
445
+
def counter(text: str) -> int:
446
+
return local_budget.count_tokens(text, server.base_url)
447
+
448
+
fitted_contents, input_budget = local_budget.fit_contents(
449
+
contents,
450
+
system_instruction,
451
+
max_output_tokens,
452
+
count=counter,
453
+
)
454
+
messages = _build_messages(fitted_contents, system_instruction)
455
+
return (
456
+
_build_request_body(
457
+
server.served_model_id,
458
+
messages,
459
+
temperature,
460
+
max_output_tokens,
461
+
json_output,
462
+
json_schema,
463
+
True,
464
+
),
465
+
input_budget,
466
+
)
467
+
468
+
469
+
def _raise_bundled_status(response: Any) -> None:
470
+
import httpx
471
+
472
+
try:
473
+
response.raise_for_status()
474
+
except httpx.HTTPStatusError as exc:
475
+
if _contains_any(response.text.lower(), _CONTEXT_WINDOW_PATTERNS):
476
+
raise ContextBudgetExceeded(
477
+
"Local request exceeded the model context window after fitting."
478
+
) from exc
479
+
raise
480
+
481
+
323
482
def run_generate(
324
483
contents: str | list[Any],
325
484
model: str,
···
332
491
timeout_s: float | None = None,
333
492
**kwargs: Any,
334
493
) -> GenerateResult:
335
-
del thinking_budget, kwargs
494
+
del thinking_budget
495
+
retry_index = int(kwargs.pop("inference_retry_index", 0) or 0)
336
496
endpoint = resolve_local_endpoint()
337
497
# Validate the requested logical id; served id comes from the server.
338
498
normalize_model_id(model)
339
499
messages = _build_messages(contents, system_instruction)
340
500
if endpoint.is_bundled:
341
501
from solstone.think.providers import local_server
502
+
from solstone.think.providers.local_admission import (
503
+
LocalAdmissionTimeout,
504
+
acquire_local_slot,
505
+
record_local_inference,
506
+
)
342
507
508
+
started = time.monotonic()
509
+
request_id = uuid.uuid4().hex
510
+
timeout = timeout_s or _DEFAULT_TIMEOUT
343
511
server = local_server.connect()
344
-
from solstone.think.providers import local_budget
345
-
346
-
def counter(text: str) -> int:
347
-
return local_budget.count_tokens(text, server.base_url)
348
-
349
-
contents, input_budget = local_budget.fit_contents(
350
-
contents,
351
-
system_instruction,
352
-
max_output_tokens,
353
-
count=counter,
354
-
)
355
-
messages = _build_messages(contents, system_instruction)
356
-
body = _build_request_body(
357
-
server.served_model_id,
358
-
messages,
359
-
temperature,
360
-
max_output_tokens,
361
-
json_output,
362
-
json_schema,
363
-
endpoint.is_bundled,
512
+
capacity = local_server.read_server_capacity()
513
+
body, input_budget = _prepare_bundled_request(
514
+
server=server,
515
+
contents=contents,
516
+
system_instruction=system_instruction,
517
+
temperature=temperature,
518
+
max_output_tokens=max_output_tokens,
519
+
json_output=json_output,
520
+
json_schema=json_schema,
364
521
)
365
522
366
523
import httpx
367
524
368
-
response = httpx.post(
369
-
f"{server.base_url}/v1/chat/completions",
370
-
json=body,
371
-
timeout=timeout_s or _DEFAULT_TIMEOUT,
372
-
)
525
+
permit = None
373
526
try:
374
-
response.raise_for_status()
375
-
except httpx.HTTPStatusError as exc:
376
-
if _contains_any(response.text.lower(), _CONTEXT_WINDOW_PATTERNS):
377
-
raise ContextBudgetExceeded(
378
-
"Local request exceeded the model context window after fitting."
379
-
) from exc
527
+
permit = acquire_local_slot(
528
+
capacity.parallel_slots,
529
+
_remaining_timeout(started, timeout),
530
+
)
531
+
with permit:
532
+
response = httpx.post(
533
+
f"{server.base_url}/v1/chat/completions",
534
+
json=body,
535
+
timeout=_remaining_timeout(started, timeout),
536
+
)
537
+
_raise_bundled_status(response)
538
+
response_data = response.json()
539
+
result = _parse_response(response_data)
540
+
telemetry = _telemetry_record(
541
+
request_id=request_id,
542
+
kind="generate",
543
+
model=LOCAL_MODEL,
544
+
profile=capacity.profile,
545
+
capacity=capacity.parallel_slots,
546
+
capacity_source=capacity.source,
547
+
started=started,
548
+
queue_wait_ms=permit.queue_wait_ms,
549
+
admission_slot=permit.slot_index,
550
+
retry_index=retry_index,
551
+
outcome="success",
552
+
finish_reason=result.get("finish_reason"),
553
+
response_data=response_data,
554
+
)
555
+
record_local_inference(telemetry)
556
+
result["inference"] = telemetry
557
+
if input_budget is not None:
558
+
result["input_budget"] = input_budget
559
+
return result
560
+
except BaseException as exc:
561
+
if isinstance(exc, KeyboardInterrupt | SystemExit):
562
+
raise
563
+
if permit is not None:
564
+
permit.release()
565
+
outcome = (
566
+
"timeout"
567
+
if isinstance(exc, (LocalAdmissionTimeout, httpx.TimeoutException))
568
+
else "cancelled"
569
+
if isinstance(exc, asyncio.CancelledError)
570
+
else "error"
571
+
)
572
+
record_local_inference(
573
+
_telemetry_record(
574
+
request_id=request_id,
575
+
kind="generate",
576
+
model=LOCAL_MODEL,
577
+
profile=capacity.profile,
578
+
capacity=capacity.parallel_slots,
579
+
capacity_source=capacity.source,
580
+
started=started,
581
+
queue_wait_ms=(
582
+
permit.queue_wait_ms
583
+
if permit is not None
584
+
else (time.monotonic() - started) * 1000.0
585
+
),
586
+
admission_slot=permit.slot_index if permit is not None else None,
587
+
retry_index=retry_index,
588
+
outcome=outcome,
589
+
reason_code=getattr(exc, "reason_code", None)
590
+
or classify_provider_error(exc, "local"),
591
+
)
592
+
)
380
593
raise
381
-
result = _parse_response(response.json())
382
-
if input_budget is not None:
383
-
result["input_budget"] = input_budget
384
-
return result
385
594
386
595
body = _build_request_body(
387
596
endpoint.served_model_id,
···
424
633
timeout_s: float | None = None,
425
634
**kwargs: Any,
426
635
) -> GenerateResult:
427
-
return await asyncio.to_thread(
428
-
run_generate,
429
-
contents,
430
-
model,
431
-
temperature,
432
-
max_output_tokens,
433
-
system_instruction,
434
-
json_output,
435
-
thinking_budget,
436
-
json_schema,
437
-
timeout_s,
438
-
**kwargs,
636
+
del thinking_budget
637
+
retry_index = int(kwargs.pop("inference_retry_index", 0) or 0)
638
+
endpoint = resolve_local_endpoint()
639
+
normalize_model_id(model)
640
+
messages = _build_messages(contents, system_instruction)
641
+
642
+
import httpx
643
+
644
+
if not endpoint.is_bundled:
645
+
body = _build_request_body(
646
+
endpoint.served_model_id,
647
+
messages,
648
+
temperature,
649
+
max_output_tokens,
650
+
json_output,
651
+
json_schema,
652
+
False,
653
+
)
654
+
post_kwargs: dict[str, Any] = {
655
+
"json": body,
656
+
"timeout": timeout_s or _DEFAULT_TIMEOUT,
657
+
}
658
+
if endpoint.credential:
659
+
post_kwargs["headers"] = {"Authorization": f"Bearer {endpoint.credential}"}
660
+
try:
661
+
async with httpx.AsyncClient() as client:
662
+
response = await client.post(
663
+
f"{endpoint.base_url}/v1/chat/completions", **post_kwargs
664
+
)
665
+
response.raise_for_status()
666
+
return _parse_response(response.json())
667
+
except asyncio.CancelledError:
668
+
raise
669
+
except Exception as exc:
670
+
raise _classify_byo_generate_error(exc) from exc
671
+
672
+
from solstone.think.providers import local_server
673
+
from solstone.think.providers.local_admission import (
674
+
LocalAdmissionTimeout,
675
+
acquire_local_slot_async,
676
+
record_local_inference,
439
677
)
440
678
679
+
started = time.monotonic()
680
+
request_id = uuid.uuid4().hex
681
+
timeout = timeout_s or _DEFAULT_TIMEOUT
682
+
server = local_server.connect()
683
+
capacity = local_server.read_server_capacity()
684
+
body, input_budget = await asyncio.to_thread(
685
+
_prepare_bundled_request,
686
+
server=server,
687
+
contents=contents,
688
+
system_instruction=system_instruction,
689
+
temperature=temperature,
690
+
max_output_tokens=max_output_tokens,
691
+
json_output=json_output,
692
+
json_schema=json_schema,
693
+
)
694
+
permit = None
695
+
try:
696
+
permit = await acquire_local_slot_async(
697
+
capacity.parallel_slots,
698
+
_remaining_timeout(started, timeout),
699
+
)
700
+
async with permit:
701
+
async with httpx.AsyncClient() as client:
702
+
response = await client.post(
703
+
f"{server.base_url}/v1/chat/completions",
704
+
json=body,
705
+
timeout=_remaining_timeout(started, timeout),
706
+
)
707
+
_raise_bundled_status(response)
708
+
response_data = response.json()
709
+
result = _parse_response(response_data)
710
+
telemetry = _telemetry_record(
711
+
request_id=request_id,
712
+
kind="generate",
713
+
model=LOCAL_MODEL,
714
+
profile=capacity.profile,
715
+
capacity=capacity.parallel_slots,
716
+
capacity_source=capacity.source,
717
+
started=started,
718
+
queue_wait_ms=permit.queue_wait_ms,
719
+
admission_slot=permit.slot_index,
720
+
retry_index=retry_index,
721
+
outcome="success",
722
+
finish_reason=result.get("finish_reason"),
723
+
response_data=response_data,
724
+
)
725
+
record_local_inference(telemetry)
726
+
result["inference"] = telemetry
727
+
if input_budget is not None:
728
+
result["input_budget"] = input_budget
729
+
return result
730
+
except BaseException as exc:
731
+
if isinstance(exc, KeyboardInterrupt | SystemExit):
732
+
raise
733
+
if permit is not None:
734
+
permit.release()
735
+
outcome = (
736
+
"cancelled"
737
+
if isinstance(exc, asyncio.CancelledError)
738
+
else "timeout"
739
+
if isinstance(exc, (LocalAdmissionTimeout, httpx.TimeoutException))
740
+
else "error"
741
+
)
742
+
record_local_inference(
743
+
_telemetry_record(
744
+
request_id=request_id,
745
+
kind="generate",
746
+
model=LOCAL_MODEL,
747
+
profile=capacity.profile,
748
+
capacity=capacity.parallel_slots,
749
+
capacity_source=capacity.source,
750
+
started=started,
751
+
queue_wait_ms=(
752
+
permit.queue_wait_ms
753
+
if permit is not None
754
+
else (time.monotonic() - started) * 1000.0
755
+
),
756
+
admission_slot=permit.slot_index if permit is not None else None,
757
+
retry_index=retry_index,
758
+
outcome=outcome,
759
+
reason_code=getattr(exc, "reason_code", None)
760
+
or classify_provider_error(exc, "local"),
761
+
)
762
+
)
763
+
raise
764
+
441
765
442
766
async def run_cogitate(
443
767
config: dict[str, Any],
444
768
on_event: Callable[[dict], None] | None = None,
445
769
) -> str:
446
770
from solstone.think.providers import local_server, openhands
771
+
from solstone.think.providers.local_admission import (
772
+
LocalAdmissionTimeout,
773
+
acquire_local_slot_async,
774
+
record_local_inference,
775
+
)
447
776
448
777
config = {**config, "model": normalize_model_id(config.get("model", LOCAL_MODEL))}
449
778
endpoint = resolve_local_endpoint()
779
+
started = time.monotonic()
780
+
request_id = uuid.uuid4().hex
781
+
server = None
782
+
capacity = None
783
+
permit = None
784
+
outcome = "success"
785
+
reason_code: str | None = None
450
786
try:
451
787
if endpoint.is_bundled:
452
-
local_server.connect()
788
+
server = local_server.connect()
789
+
capacity = local_server.read_server_capacity()
790
+
timeout = float(config.get("timeout_seconds", 600) or 600)
791
+
permit = await acquire_local_slot_async(
792
+
capacity.parallel_slots,
793
+
_remaining_timeout(started, timeout),
794
+
)
453
795
return await openhands.run_cogitate(config, on_event=on_event)
796
+
except asyncio.CancelledError:
797
+
outcome = "cancelled"
798
+
reason_code = "cancelled"
799
+
raise
454
800
except Exception as exc:
801
+
outcome = (
802
+
"timeout"
803
+
if isinstance(exc, LocalAdmissionTimeout)
804
+
or getattr(exc, "reason_code", None) == "wall_clock_exceeded"
805
+
else "error"
806
+
)
455
807
from solstone.think.talents import TalentHookError
456
808
457
809
if isinstance(exc, TalentHookError):
458
810
raise
459
811
460
-
reason_code = None
461
812
if not endpoint.is_bundled:
462
813
reason_code = classify_byo_cogitate_error(exc) or getattr(
463
814
exc, "reason_code", None
464
815
)
816
+
reason_code = (
817
+
reason_code
818
+
or getattr(exc, "reason_code", None)
819
+
or classify_provider_error(exc, "local")
820
+
)
465
821
if on_event and not getattr(exc, "_evented", False):
466
-
reason_code = (
467
-
reason_code
468
-
or getattr(exc, "reason_code", None)
469
-
or classify_provider_error(exc, "local")
470
-
)
471
822
error_text = str(exc)
472
823
trace_text = traceback.format_exc()
473
824
fixed_copy = local_endpoint_reason_copy(reason_code)
···
493
844
setattr(wrapped, "_evented", getattr(exc, "_evented", False))
494
845
raise wrapped from exc
495
846
raise
847
+
finally:
848
+
if permit is not None:
849
+
permit.release()
850
+
if server is not None and capacity is not None:
851
+
record_local_inference(
852
+
_telemetry_record(
853
+
request_id=request_id,
854
+
kind="cogitate",
855
+
model=LOCAL_MODEL,
856
+
profile=capacity.profile,
857
+
capacity=capacity.parallel_slots,
858
+
capacity_source=capacity.source,
859
+
started=started,
860
+
queue_wait_ms=(
861
+
permit.queue_wait_ms
862
+
if permit is not None
863
+
else (time.monotonic() - started) * 1000.0
864
+
),
865
+
admission_slot=permit.slot_index if permit is not None else None,
866
+
retry_index=None,
867
+
outcome=outcome,
868
+
finish_reason="stop" if outcome == "success" else None,
869
+
reason_code=reason_code,
870
+
)
871
+
)
496
872
497
873
498
874
def list_models(provider: str = "local") -> list[dict[str, Any]]:
+220
solstone/think/providers/local_admission.py
+220
solstone/think/providers/local_admission.py
···
1
+
# SPDX-License-Identifier: AGPL-3.0-only
2
+
# Copyright (c) 2026 sol pbc
3
+
4
+
"""Cross-process admission and content-free telemetry for local inference."""
5
+
6
+
from __future__ import annotations
7
+
8
+
import asyncio
9
+
import errno
10
+
import fcntl
11
+
import logging
12
+
import os
13
+
import time
14
+
import uuid
15
+
from dataclasses import dataclass
16
+
from pathlib import Path
17
+
from typing import IO, Any
18
+
19
+
from solstone.think.journal_io import append_jsonl
20
+
from solstone.think.utils import get_journal
21
+
22
+
LOG = logging.getLogger(__name__)
23
+
24
+
_POLL_INTERVAL_S = 0.025
25
+
26
+
27
+
class LocalAdmissionTimeout(TimeoutError):
28
+
"""No bundled-local inference slot became available before the deadline."""
29
+
30
+
reason_code = "local_queue_timeout"
31
+
32
+
33
+
@dataclass
34
+
class LocalPermit:
35
+
"""One flock-backed serving-capacity permit."""
36
+
37
+
slot_index: int
38
+
capacity: int
39
+
queue_wait_ms: float
40
+
_file: IO[str]
41
+
42
+
def release(self) -> None:
43
+
if self._file.closed:
44
+
return
45
+
try:
46
+
fcntl.flock(self._file, fcntl.LOCK_UN)
47
+
finally:
48
+
self._file.close()
49
+
50
+
def __enter__(self) -> LocalPermit:
51
+
return self
52
+
53
+
def __exit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
54
+
self.release()
55
+
56
+
async def __aenter__(self) -> LocalPermit:
57
+
return self
58
+
59
+
async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
60
+
self.release()
61
+
62
+
63
+
def _admission_dir() -> Path:
64
+
return Path(get_journal()) / "health" / "local-inference-admission"
65
+
66
+
67
+
@dataclass
68
+
class _WaitTicket:
69
+
path: Path
70
+
file: IO[str]
71
+
72
+
73
+
def _create_ticket(root: Path) -> _WaitTicket:
74
+
root.mkdir(parents=True, exist_ok=True)
75
+
identity = f"{time.monotonic_ns():020d}-{os.getpid()}-{uuid.uuid4().hex}"
76
+
path = root / f"wait-{identity}.ticket"
77
+
creating_path = root / f".creating-{identity}.ticket"
78
+
ticket_file = open(creating_path, "x+", encoding="utf-8")
79
+
fcntl.flock(ticket_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
80
+
creating_path.rename(path)
81
+
return _WaitTicket(path=path, file=ticket_file)
82
+
83
+
84
+
def _drop_ticket(ticket: _WaitTicket) -> None:
85
+
try:
86
+
ticket.path.unlink(missing_ok=True)
87
+
finally:
88
+
try:
89
+
fcntl.flock(ticket.file, fcntl.LOCK_UN)
90
+
finally:
91
+
ticket.file.close()
92
+
93
+
94
+
def _ticket_has_turn(root: Path, ticket: _WaitTicket) -> bool:
95
+
"""Return whether ticket is oldest, pruning tickets whose owners exited."""
96
+
while True:
97
+
waiting = sorted(root.glob("wait-*.ticket"))
98
+
if not waiting:
99
+
return False
100
+
first = waiting[0]
101
+
if first == ticket.path:
102
+
return True
103
+
try:
104
+
stale_file = open(first, "a+", encoding="utf-8")
105
+
except FileNotFoundError:
106
+
continue
107
+
try:
108
+
try:
109
+
fcntl.flock(stale_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
110
+
except OSError as exc:
111
+
if exc.errno in (errno.EACCES, errno.EAGAIN, errno.EWOULDBLOCK):
112
+
return False
113
+
raise
114
+
first.unlink(missing_ok=True)
115
+
finally:
116
+
stale_file.close()
117
+
118
+
119
+
def _try_acquire(capacity: int, started: float, root: Path) -> LocalPermit | None:
120
+
for slot_index in range(capacity):
121
+
lock_file = open(root / f"slot-{slot_index}.lock", "a+", encoding="utf-8")
122
+
try:
123
+
fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
124
+
except OSError as exc:
125
+
lock_file.close()
126
+
if exc.errno in (errno.EACCES, errno.EAGAIN, errno.EWOULDBLOCK):
127
+
continue
128
+
raise
129
+
return LocalPermit(
130
+
slot_index=slot_index,
131
+
capacity=capacity,
132
+
queue_wait_ms=(time.monotonic() - started) * 1000.0,
133
+
_file=lock_file,
134
+
)
135
+
return None
136
+
137
+
138
+
def _deadline(started: float, timeout_s: float | None) -> float | None:
139
+
if timeout_s is None:
140
+
return None
141
+
return started + max(0.0, timeout_s)
142
+
143
+
144
+
def acquire_local_slot(capacity: int, timeout_s: float | None) -> LocalPermit:
145
+
"""Wait synchronously for one bundled-local serving slot."""
146
+
if capacity < 1:
147
+
raise ValueError("local inference capacity must be at least one")
148
+
started = time.monotonic()
149
+
deadline = _deadline(started, timeout_s)
150
+
root = _admission_dir()
151
+
ticket = _create_ticket(root)
152
+
try:
153
+
while True:
154
+
if _ticket_has_turn(root, ticket):
155
+
permit = _try_acquire(capacity, started, root)
156
+
if permit is not None:
157
+
return permit
158
+
now = time.monotonic()
159
+
if deadline is not None and now >= deadline:
160
+
raise LocalAdmissionTimeout(
161
+
f"Local inference queue exceeded its {timeout_s:.3f}s deadline."
162
+
)
163
+
sleep_s = _POLL_INTERVAL_S
164
+
if deadline is not None:
165
+
sleep_s = min(sleep_s, max(0.0, deadline - now))
166
+
time.sleep(sleep_s)
167
+
finally:
168
+
_drop_ticket(ticket)
169
+
170
+
171
+
async def acquire_local_slot_async(
172
+
capacity: int, timeout_s: float | None
173
+
) -> LocalPermit:
174
+
"""Wait cancellation-safely for one bundled-local serving slot."""
175
+
if capacity < 1:
176
+
raise ValueError("local inference capacity must be at least one")
177
+
started = time.monotonic()
178
+
deadline = _deadline(started, timeout_s)
179
+
root = _admission_dir()
180
+
ticket = _create_ticket(root)
181
+
try:
182
+
while True:
183
+
if _ticket_has_turn(root, ticket):
184
+
permit = _try_acquire(capacity, started, root)
185
+
if permit is not None:
186
+
return permit
187
+
now = time.monotonic()
188
+
if deadline is not None and now >= deadline:
189
+
raise LocalAdmissionTimeout(
190
+
f"Local inference queue exceeded its {timeout_s:.3f}s deadline."
191
+
)
192
+
sleep_s = _POLL_INTERVAL_S
193
+
if deadline is not None:
194
+
sleep_s = min(sleep_s, max(0.0, deadline - now))
195
+
await asyncio.sleep(sleep_s)
196
+
finally:
197
+
_drop_ticket(ticket)
198
+
199
+
200
+
def record_local_inference(record: dict[str, Any]) -> None:
201
+
"""Durably append one prompt/output-free local inference record."""
202
+
try:
203
+
path = (
204
+
Path(get_journal())
205
+
/ "health"
206
+
/ "local-inference"
207
+
/ f"{time.strftime('%Y%m%d')}.jsonl"
208
+
)
209
+
append_jsonl(path, record)
210
+
except Exception:
211
+
LOG.warning("failed to record local inference telemetry", exc_info=True)
212
+
213
+
214
+
__all__ = [
215
+
"LocalAdmissionTimeout",
216
+
"LocalPermit",
217
+
"acquire_local_slot",
218
+
"acquire_local_slot_async",
219
+
"record_local_inference",
220
+
]
+42
-10
solstone/think/providers/local_server.py
+42
-10
solstone/think/providers/local_server.py
···
6
6
from __future__ import annotations
7
7
8
8
import logging
9
+
import sys
9
10
from dataclasses import dataclass
10
11
from pathlib import Path
11
12
from typing import Any
···
65
66
binary_path: str | None = None
66
67
model_path: str | None = None
67
68
served_model_id: str = LOCAL_MODEL
69
+
parallel_slots: int = 1
70
+
capacity_source: str = "default"
71
+
profile: str = "floor"
72
+
73
+
74
+
@dataclass(frozen=True)
75
+
class ServerCapacity:
76
+
parallel_slots: int
77
+
source: str
78
+
profile: str
68
79
69
80
70
81
def _base_url(port: int) -> str:
···
166
177
# server that never launched has no slots at all.
167
178
_UNKNOWN_SLOTS = 1
168
179
169
-
_PARALLEL_SLOTS_CACHE: int | None = None
180
+
_SERVER_CAPACITY_CACHE: ServerCapacity | None = None
170
181
171
182
172
183
def _extract_total_slots(props: dict[str, Any]) -> int | None:
···
186
197
return None
187
198
188
199
189
-
def _discover_parallel_slots() -> int:
200
+
def _profile_for_slots(slots: int) -> str:
201
+
if sys.platform == "darwin":
202
+
return "apple"
203
+
if slots == _CAPABLE_TIER.parallel_slots:
204
+
return _CAPABLE_TIER.name
205
+
if slots == _FLOOR_TIER.parallel_slots:
206
+
return _FLOOR_TIER.name
207
+
return "advertised"
208
+
209
+
210
+
def _discover_server_capacity() -> ServerCapacity:
190
211
port = read_service_port(_SERVICE_NAME)
191
212
props = fetch_props(port) if port is not None else None
192
213
if props is not None:
193
214
slots = _extract_total_slots(props)
194
215
if slots is not None:
195
-
return slots
216
+
return ServerCapacity(slots, "props", _profile_for_slots(slots))
196
217
197
218
context_tokens = read_local_context_window()
198
219
slots = _slots_from_launched_tier(context_tokens)
···
206
227
context_tokens,
207
228
source,
208
229
)
209
-
return slots
230
+
return ServerCapacity(slots, source, _profile_for_slots(slots))
231
+
232
+
233
+
def read_server_capacity() -> ServerCapacity:
234
+
"""Return the memoized bundled-server capacity and its evidence source."""
235
+
global _SERVER_CAPACITY_CACHE
236
+
if _SERVER_CAPACITY_CACHE is None:
237
+
_SERVER_CAPACITY_CACHE = _discover_server_capacity()
238
+
return _SERVER_CAPACITY_CACHE
210
239
211
240
212
241
def read_server_parallel_slots() -> int:
···
216
245
the context window persisted at launch, then to ``_UNKNOWN_SLOTS``.
217
246
Discovered once per process. Never raises; always returns >= 1.
218
247
"""
219
-
global _PARALLEL_SLOTS_CACHE
220
-
if _PARALLEL_SLOTS_CACHE is None:
221
-
_PARALLEL_SLOTS_CACHE = _discover_parallel_slots()
222
-
return _PARALLEL_SLOTS_CACHE
248
+
return read_server_capacity().parallel_slots
223
249
224
250
225
251
def reset_parallel_slots_cache() -> None:
226
252
"""Clear the per-process slot-discovery memo."""
227
-
global _PARALLEL_SLOTS_CACHE
228
-
_PARALLEL_SLOTS_CACHE = None
253
+
global _SERVER_CAPACITY_CACHE
254
+
_SERVER_CAPACITY_CACHE = None
229
255
230
256
231
257
def _resolve_served_model_id(health_body: dict[str, Any] | None) -> str | None:
···
263
289
served_model_id = _resolve_served_model_id(body)
264
290
if served_model_id is None:
265
291
raise LocalProviderError("local_model_not_ready", LOCAL_MODEL_NOT_READY_COPY)
292
+
capacity = read_server_capacity()
266
293
return LocalServerInfo(
267
294
model_id=LOCAL_MODEL,
268
295
port=port,
269
296
base_url=_base_url(port),
270
297
state=STATE_READY,
271
298
served_model_id=served_model_id,
299
+
parallel_slots=capacity.parallel_slots,
300
+
capacity_source=capacity.source,
301
+
profile=capacity.profile,
272
302
)
273
303
274
304
···
276
306
"LOCAL_MIN_CONTEXT_TOKENS",
277
307
"LOCAL_MODEL_NOT_READY_COPY",
278
308
"LocalServerInfo",
309
+
"ServerCapacity",
279
310
"ServerTier",
280
311
"STATE_IDLE",
281
312
"STATE_STARTING",
···
289
320
"probe_state",
290
321
"read_local_context_window",
291
322
"read_server_context_window",
323
+
"read_server_capacity",
292
324
"read_server_parallel_slots",
293
325
"reset_parallel_slots_cache",
294
326
"select_server_tier",
+1
solstone/think/talents.py
+1
solstone/think/talents.py
+80
tests/test_local.py
+80
tests/test_local.py
···
27
27
from solstone.think.talents import TalentHookError
28
28
29
29
30
+
@pytest.fixture(autouse=True)
31
+
def _isolate_local_admission(monkeypatch, tmp_path):
32
+
from solstone.think.providers import local_admission
33
+
34
+
monkeypatch.setattr(
35
+
local_admission,
36
+
"_admission_dir",
37
+
lambda: tmp_path / "local-inference-admission",
38
+
)
39
+
monkeypatch.setattr(local_admission, "record_local_inference", lambda _record: None)
40
+
41
+
30
42
def _provider():
31
43
providers_pkg = importlib.import_module("solstone.think.providers")
32
44
if hasattr(providers_pkg, "local_budget"):
···
1629
1641
1630
1642
1631
1643
def _local_journal(monkeypatch, tmp_path: Path) -> Path:
1644
+
from solstone.think.providers import local_server
1645
+
1632
1646
journal = tmp_path / "journal"
1633
1647
(journal / "health").mkdir(parents=True)
1634
1648
monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
1649
+
local_server.reset_parallel_slots_cache()
1635
1650
return journal
1636
1651
1637
1652
···
1652
1667
1653
1668
# /props is ground truth; it wins over the persisted tier.
1654
1669
assert local_server.read_server_parallel_slots() == 2
1670
+
assert local_server.read_server_capacity() == local_server.ServerCapacity(
1671
+
parallel_slots=2,
1672
+
source="props",
1673
+
profile="capable",
1674
+
)
1655
1675
1656
1676
1657
1677
def test_read_server_parallel_slots_no_port_returns_floor(
···
1674
1694
)
1675
1695
1676
1696
1697
+
def test_server_capacity_uses_explicit_apple_profile(monkeypatch, tmp_path):
1698
+
from solstone.think.providers import local_server
1699
+
1700
+
_local_journal(monkeypatch, tmp_path)
1701
+
monkeypatch.setattr(local_server.sys, "platform", "darwin")
1702
+
1703
+
assert local_server.read_server_capacity() == local_server.ServerCapacity(
1704
+
parallel_slots=1,
1705
+
source="default",
1706
+
profile="apple",
1707
+
)
1708
+
1709
+
1677
1710
@pytest.mark.parametrize("slots", [1, 2])
1678
1711
def test_read_server_parallel_slots_falls_back_to_launched_tier(
1679
1712
monkeypatch, tmp_path, slots
···
1735
1768
local_server.reset_parallel_slots_cache()
1736
1769
assert local_server.read_server_parallel_slots() == 2
1737
1770
assert len(calls) == 2
1771
+
1772
+
1773
+
def test_local_response_telemetry_normalizes_llama_and_mlx_fields():
1774
+
provider = _provider()
1775
+
1776
+
fields = provider._server_response_fields(
1777
+
{
1778
+
"timings": {
1779
+
"cache_n": 236,
1780
+
"prompt_n": 12,
1781
+
"prompt_ms": 30.5,
1782
+
"predicted_n": 35,
1783
+
"predicted_ms": 661.0,
1784
+
"slot_id": 1,
1785
+
},
1786
+
"usage": {
1787
+
"prompt_tokens": 248,
1788
+
"completion_tokens": 35,
1789
+
},
1790
+
}
1791
+
)
1792
+
1793
+
assert fields == {
1794
+
"prompt_eval_ms": 30.5,
1795
+
"generation_ms": 661.0,
1796
+
"server_total_ms": 691.5,
1797
+
"prompt_tokens": 248,
1798
+
"generated_tokens": 35,
1799
+
"prompt_cached_tokens": 236,
1800
+
"selected_slot": 1,
1801
+
"prompt_cache_state": "warm",
1802
+
}
1803
+
assert provider._extract_usage(
1804
+
{
1805
+
"usage": {
1806
+
"prompt_tokens": 248,
1807
+
"completion_tokens": 35,
1808
+
"total_tokens": 283,
1809
+
"prompt_tokens_details": {"cached_tokens": 236},
1810
+
}
1811
+
}
1812
+
) == {
1813
+
"input_tokens": 248,
1814
+
"output_tokens": 35,
1815
+
"total_tokens": 283,
1816
+
"cached_tokens": 236,
1817
+
}
+150
tests/test_local_admission.py
+150
tests/test_local_admission.py
···
1
+
# SPDX-License-Identifier: AGPL-3.0-only
2
+
# Copyright (c) 2026 sol pbc
3
+
4
+
from __future__ import annotations
5
+
6
+
import asyncio
7
+
import json
8
+
import threading
9
+
import time
10
+
11
+
import pytest
12
+
13
+
from solstone.think.providers.local_admission import (
14
+
LocalAdmissionTimeout,
15
+
acquire_local_slot,
16
+
acquire_local_slot_async,
17
+
record_local_inference,
18
+
)
19
+
20
+
21
+
def _isolated_journal(monkeypatch, tmp_path):
22
+
import solstone.think.utils as think_utils
23
+
24
+
monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
25
+
think_utils._journal_path_cache = None
26
+
27
+
28
+
def test_cross_thread_admission_never_exceeds_capacity(monkeypatch, tmp_path):
29
+
_isolated_journal(monkeypatch, tmp_path)
30
+
active = 0
31
+
peak = 0
32
+
lock = threading.Lock()
33
+
34
+
def work() -> None:
35
+
nonlocal active, peak
36
+
with acquire_local_slot(2, 2):
37
+
with lock:
38
+
active += 1
39
+
peak = max(peak, active)
40
+
time.sleep(0.04)
41
+
with lock:
42
+
active -= 1
43
+
44
+
threads = [threading.Thread(target=work) for _ in range(8)]
45
+
for thread in threads:
46
+
thread.start()
47
+
for thread in threads:
48
+
thread.join()
49
+
50
+
assert peak == 2
51
+
52
+
53
+
def test_failure_releases_permit(monkeypatch, tmp_path):
54
+
_isolated_journal(monkeypatch, tmp_path)
55
+
56
+
with pytest.raises(RuntimeError, match="boom"):
57
+
with acquire_local_slot(1, 0.5):
58
+
raise RuntimeError("boom")
59
+
60
+
with acquire_local_slot(1, 0.1) as permit:
61
+
assert permit.slot_index == 0
62
+
63
+
64
+
def test_sync_queue_timeout(monkeypatch, tmp_path):
65
+
_isolated_journal(monkeypatch, tmp_path)
66
+
first = acquire_local_slot(1, 0.1)
67
+
try:
68
+
with pytest.raises(LocalAdmissionTimeout):
69
+
acquire_local_slot(1, 0.03)
70
+
finally:
71
+
first.release()
72
+
73
+
74
+
def test_async_queued_cancellation_does_not_leak(monkeypatch, tmp_path):
75
+
_isolated_journal(monkeypatch, tmp_path)
76
+
77
+
async def exercise() -> None:
78
+
first = await acquire_local_slot_async(1, 0.5)
79
+
queued = asyncio.create_task(acquire_local_slot_async(1, 1.0))
80
+
await asyncio.sleep(0.05)
81
+
queued.cancel()
82
+
with pytest.raises(asyncio.CancelledError):
83
+
await queued
84
+
assert not list(
85
+
(tmp_path / "health" / "local-inference-admission").glob("wait-*.ticket")
86
+
)
87
+
first.release()
88
+
second = await acquire_local_slot_async(1, 0.1)
89
+
second.release()
90
+
91
+
asyncio.run(exercise())
92
+
93
+
94
+
def test_waiters_are_admitted_in_ticket_order(monkeypatch, tmp_path):
95
+
_isolated_journal(monkeypatch, tmp_path)
96
+
root = tmp_path / "health" / "local-inference-admission"
97
+
first = acquire_local_slot(1, 1)
98
+
order: list[int] = []
99
+
100
+
def wait(index: int) -> None:
101
+
with acquire_local_slot(1, 2):
102
+
order.append(index)
103
+
time.sleep(0.01)
104
+
105
+
threads = []
106
+
for index in range(5):
107
+
thread = threading.Thread(target=wait, args=(index,))
108
+
thread.start()
109
+
threads.append(thread)
110
+
deadline = time.monotonic() + 1
111
+
while len(list(root.glob("wait-*.ticket"))) < index + 1:
112
+
assert time.monotonic() < deadline
113
+
time.sleep(0.005)
114
+
115
+
first.release()
116
+
for thread in threads:
117
+
thread.join()
118
+
119
+
assert order == list(range(5))
120
+
121
+
122
+
def test_stale_ticket_from_exited_owner_is_pruned(monkeypatch, tmp_path):
123
+
_isolated_journal(monkeypatch, tmp_path)
124
+
root = tmp_path / "health" / "local-inference-admission"
125
+
root.mkdir(parents=True)
126
+
stale = root / "wait-00000000000000000000-1-stale.ticket"
127
+
stale.write_text("", encoding="utf-8")
128
+
129
+
with acquire_local_slot(1, 0.2):
130
+
assert not stale.exists()
131
+
132
+
133
+
def test_telemetry_is_durable_and_content_free(monkeypatch, tmp_path):
134
+
_isolated_journal(monkeypatch, tmp_path)
135
+
record_local_inference(
136
+
{
137
+
"timestamp": 1.0,
138
+
"request_id": "abc",
139
+
"provider": "local",
140
+
"model": "local/qwen3.5-4b",
141
+
"queue_wait_ms": 12.5,
142
+
"outcome": "success",
143
+
}
144
+
)
145
+
146
+
path = tmp_path / "health" / "local-inference" / time.strftime("%Y%m%d.jsonl")
147
+
row = json.loads(path.read_text(encoding="utf-8"))
148
+
assert row["request_id"] == "abc"
149
+
assert "prompt" not in row
150
+
assert "output" not in row
+13
tests/test_log_retention.py
+13
tests/test_log_retention.py
···
121
121
assert not (journal / "chronicle" / old_day / "task_log.txt").exists()
122
122
123
123
124
+
def test_local_inference_telemetry_follows_journal_log_retention(journal):
125
+
old_day = _day(31)
126
+
recent_day = _day(1)
127
+
old = _write(journal / "health" / "local-inference" / f"{old_day}.jsonl")
128
+
recent = _write(journal / "health" / "local-inference" / f"{recent_day}.jsonl")
129
+
130
+
result = prune(config=LogRetentionConfig(days=30))
131
+
132
+
assert not old.exists()
133
+
assert recent.exists()
134
+
assert result.by_class["local_inference"]["files_deleted"] == 1
135
+
136
+
124
137
def test_ac4_dry_run_reports_candidates_without_deleting_or_audit(journal):
125
138
old_day = _day(31)
126
139
old_token = _write(journal / "tokens" / f"{old_day}.jsonl", "old")