solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4from __future__ import annotations
5
6import ast
7import asyncio
8import concurrent.futures
9import subprocess
10import sys
11import threading
12import time
13from collections import OrderedDict
14from collections.abc import Iterator
15from dataclasses import replace
16from pathlib import Path
17from typing import Any
18from unittest.mock import MagicMock
19
20import pytest
21
22from solstone.think import supervisor
23from solstone.think.providers.artifact_proof import ReadinessOutcome
24from solstone.think.providers.brain_state import build_active_brain_fingerprint
25from solstone.think.providers.install_state import (
26 IN_FLIGHT_STATES,
27 TERMINAL_STATES,
28 InstallState,
29)
30from solstone.think.providers.runtime_health import (
31 RUNTIME_PHASES,
32 ReasonCode,
33 RuntimeHealthRecord,
34 RuntimePhase,
35 read_retry_token,
36 read_runtime_health,
37 request_retry_token,
38 request_runtime_retry,
39 write_runtime_health,
40)
41
42
43@pytest.fixture
44def provider_cache_reset() -> Iterator[None]:
45 from solstone.think.providers import local_server, local_vulkan
46
47 local_vulkan.reset_detect_cache()
48 local_server.reset_parallel_slots_cache()
49 try:
50 yield
51 finally:
52 local_vulkan.reset_detect_cache()
53 local_server.reset_parallel_slots_cache()
54
55
56@pytest.fixture(autouse=True)
57def runtime_state_reset(
58 tmp_path, monkeypatch, provider_cache_reset, set_test_journal_path
59):
60 import solstone.think.utils as think_utils
61
62 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path / "journal"))
63 think_utils._journal_path_cache = None
64 states = {
65 "local": supervisor.ProviderRuntimeState("local"),
66 "parakeet": supervisor.ProviderRuntimeState("parakeet"),
67 }
68 monkeypatch.setattr(supervisor, "_provider_runtime_states", states)
69 monkeypatch.setattr(
70 supervisor,
71 "_recovery_state",
72 {
73 "local": supervisor.ProviderRecoveryState(),
74 "parakeet": supervisor.ProviderRecoveryState(),
75 },
76 )
77 monkeypatch.setattr(supervisor, "_provider_runtime_executor", None)
78 monkeypatch.setattr(
79 supervisor,
80 "_wedge_state",
81 {
82 "providers": OrderedDict(),
83 "failures": set(),
84 "cooldown_until": 0.0,
85 "awaiting_recovery": False,
86 },
87 )
88 monkeypatch.setattr(supervisor, "_provider_startup_gate", None)
89 monkeypatch.setattr(supervisor, "_parakeet_admission_retry_epoch", 0)
90 monkeypatch.setattr(supervisor, "_task_queue", None)
91 supervisor._SERVICE_STATE.clear()
92 yield
93 executor = supervisor._provider_runtime_executor
94 if executor is not None:
95 executor.shutdown(wait=True, cancel_futures=True)
96 think_utils._journal_path_cache = None
97
98
99class _InlineExecutor:
100 def submit(self, fn, *args, **kwargs):
101 future: concurrent.futures.Future = concurrent.futures.Future()
102 try:
103 future.set_result(fn(*args, **kwargs))
104 except BaseException as exc:
105 future.set_exception(exc)
106 return future
107
108
109def _future_with(result: Any) -> concurrent.futures.Future:
110 future: concurrent.futures.Future = concurrent.futures.Future()
111 future.set_result(result)
112 return future
113
114
115class _FakeTaskQueue:
116 def __init__(self) -> None:
117 self.ready_calls = 0
118 self.submitted: list[tuple[list[str], tuple[Any, ...], dict[str, Any]]] = []
119
120 def set_ready(self) -> None:
121 self.ready_calls += 1
122
123 def submit(self, cmd: list[str], *args: Any, **kwargs: Any) -> None:
124 self.submitted.append((cmd, args, kwargs))
125
126
127class _FakeReservation:
128 def __init__(self, port: int = 45123):
129 self.port = port
130 self.closed = False
131
132 def release_for_spawn(self) -> int:
133 self.closed = True
134 return self.port
135
136 def close(self) -> None:
137 self.closed = True
138
139
140class _FakeProcess:
141 pid = 12345
142 returncode = None
143
144 def poll(self) -> None:
145 return None
146
147
148class _FakeManaged:
149 def __init__(self, name: str = supervisor.LOCAL_SERVER_PROCESS_NAME) -> None:
150 self.name = name
151 self.ref = f"ref-{name}"
152 self.process = _FakeProcess()
153 self.cleanup = MagicMock()
154 self.terminate = MagicMock()
155
156 def is_running(self) -> bool:
157 return True
158
159
160class _DeadManaged(_FakeManaged):
161 def is_running(self) -> bool:
162 return False
163
164
165def _hold_local_slot_in_child(root: Path):
166 ready = root / "slot-holder.ready"
167 code = (
168 "import fcntl, pathlib, sys, time\n"
169 "root = pathlib.Path(sys.argv[1])\n"
170 "ready = pathlib.Path(sys.argv[2])\n"
171 "root.mkdir(parents=True, exist_ok=True)\n"
172 "f = open(root / 'slot-0.lock', 'a+', encoding='utf-8')\n"
173 "fcntl.flock(f, fcntl.LOCK_EX)\n"
174 "ready.write_text('1', encoding='utf-8')\n"
175 "time.sleep(60)\n"
176 )
177 proc = subprocess.Popen(
178 [sys.executable, "-c", code, str(root), str(ready)],
179 )
180 deadline = time.monotonic() + 2.0
181 while time.monotonic() < deadline:
182 if ready.exists():
183 return proc
184 time.sleep(0.01)
185 proc.terminate()
186 proc.wait(timeout=2)
187 raise AssertionError("local admission slot holder did not become ready")
188
189
190def _readiness(
191 status: str,
192 reason_code: str,
193 *,
194 install_state: InstallState = "idle",
195 host: dict[str, Any] | None = None,
196) -> ReadinessOutcome:
197 return ReadinessOutcome(
198 provider="parakeet",
199 status=status,
200 reason_code=reason_code,
201 target={},
202 install={"install_state": install_state},
203 host=host or {},
204 artifacts={},
205 proof={},
206 )
207
208
209def _local_readiness(
210 status: str = "ready",
211 reason_code: str = "ready",
212 *,
213 install_state: InstallState = "idle",
214 host_reason: str = "gpu_unavailable",
215) -> ReadinessOutcome:
216 return ReadinessOutcome(
217 provider="local",
218 status=status,
219 reason_code=reason_code,
220 target={"model_id": supervisor.LOCAL_MODEL},
221 install={"install_state": install_state},
222 host=(
223 {
224 "backend": "vulkan",
225 "backend_reason": "test",
226 }
227 if status == "ready"
228 else {"reason": host_reason}
229 ),
230 artifacts=(
231 {
232 "binary_path": "/tmp/llama-server",
233 "model_path": "/tmp/model.gguf",
234 "mmproj_path": None,
235 }
236 if status == "ready"
237 else {}
238 ),
239 proof={},
240 )
241
242
243def _status_snapshot(
244 *,
245 install_state: InstallState = "downloading",
246 attempt_id: str | None = "attempt-live",
247 target_fingerprint_sha256: str | None = "fp-local",
248 revision: int = 7,
249) -> dict[str, Any]:
250 return {
251 "revision": revision,
252 "install_state": install_state,
253 "attempt_id": attempt_id,
254 "target_fingerprint_sha256": target_fingerprint_sha256,
255 }
256
257
258def _local_install_progress_detail(
259 *,
260 readiness_state: InstallState,
261 attempt_id: str,
262 revision: int,
263) -> dict[str, Any]:
264 return {
265 "readiness_status": "missing-or-mismatched",
266 "readiness_reason_code": "manifest_missing",
267 "install_state": readiness_state,
268 "install_acquisition_allowed": False,
269 "install_attempt_id": attempt_id,
270 "install_revision": revision,
271 }
272
273
274def _local_artifact_missing_detail(install_state: InstallState) -> dict[str, Any]:
275 return {
276 "readiness_status": "missing-or-mismatched",
277 "readiness_reason_code": "manifest_missing",
278 "install_state": install_state,
279 "install_acquisition_allowed": install_state == "idle",
280 }
281
282
283def _assert_local_observation(
284 observation: supervisor.ProviderTruthObservation | None,
285 *,
286 phase: RuntimePhase,
287 reason_code: ReasonCode,
288 desired_json: str,
289 desired_sha: str,
290 boot_required: bool,
291 detail: dict[str, Any],
292) -> None:
293 assert observation is not None
294 assert observation.phase == phase
295 assert observation.reason_code == reason_code
296 assert observation.desired_fingerprint_json == desired_json
297 assert observation.desired_fingerprint_sha256 == desired_sha
298 assert observation.boot_required is boot_required
299 assert observation.detail == detail
300 assert observation.plan is None
301
302
303def _patch_lease_state(monkeypatch, state: str) -> None:
304 monkeypatch.setattr(
305 "solstone.think.providers.install_lease.probe_install_lease_state",
306 lambda _provider: state,
307 )
308
309
310def _local_plan() -> supervisor.LocalServerLaunchPlan:
311 return supervisor.LocalServerLaunchPlan(
312 backend="vulkan",
313 desired_fingerprint_json='{"provider":"local"}',
314 desired_fingerprint_sha256="fp-local",
315 binary_path=Path("/tmp/llama-server"),
316 model_path=Path("/tmp/model.gguf"),
317 context_tokens=16384,
318 parallel_slots=1,
319 prompt_cache_mib=0,
320 env_updates={"GGML_VK_VISIBLE_DEVICES": "0"},
321 )
322
323
324def _cuda_plan() -> supervisor.LocalServerLaunchPlan:
325 return supervisor.LocalServerLaunchPlan(
326 backend="cuda",
327 desired_fingerprint_json='{"provider":"local","backend":"cuda"}',
328 desired_fingerprint_sha256="fp-local-cuda",
329 binary_path=Path("/tmp/llama-server"),
330 model_path=Path("/tmp/model.gguf"),
331 lib_dir=Path("/tmp/cuda/lib"),
332 gpu_index=0,
333 gpu_vram_mib=24576,
334 context_tokens=32768,
335 parallel_slots=1,
336 prompt_cache_mib=0,
337 visible_devices_env="CUDA_VISIBLE_DEVICES",
338 visible_devices_value="0",
339 env_updates={"CUDA_VISIBLE_DEVICES": "0", "LD_LIBRARY_PATH": "/tmp/cuda/lib"},
340 )
341
342
343def _mlx_plan() -> supervisor.LocalServerLaunchPlan:
344 return supervisor.LocalServerLaunchPlan(
345 backend="mlx",
346 desired_fingerprint_json='{"provider":"local","backend":"mlx"}',
347 desired_fingerprint_sha256="fp-local-mlx",
348 model_id=supervisor.LOCAL_MODEL,
349 runtime_dir=Path("/tmp/mlx-runtime"),
350 )
351
352
353def _parakeet_plan(backend: str = "cpu") -> supervisor.ParakeetServerLaunchPlan:
354 return supervisor.ParakeetServerLaunchPlan(
355 binary_backend=backend,
356 env_updates={"GGML_VK_VISIBLE_DEVICES": "0"} if backend == "vulkan" else {},
357 gpu_index=0 if backend == "vulkan" else None,
358 binary_path=Path(f"/tmp/parakeet-{backend}"),
359 model_path=Path("/tmp/parakeet-model.bin"),
360 threads=4,
361 library_dirs=(),
362 desired_fingerprint_json='{"provider":"parakeet"}',
363 desired_fingerprint_sha256="fp-parakeet",
364 placement="gpu" if backend == "vulkan" else "cpu",
365 )
366
367
368@pytest.mark.parametrize("install_state", sorted(IN_FLIGHT_STATES))
369def test_local_readiness_block_observation_upgrades_live_inflight_states(
370 monkeypatch,
371 install_state: InstallState,
372) -> None:
373 target = {"provider": "local", "target": "live-install"}
374 desired_json, desired_sha = supervisor._target_fingerprint_pair(target)
375 before_status = _status_snapshot(
376 install_state=install_state,
377 attempt_id=f"attempt-{install_state}",
378 target_fingerprint_sha256=desired_sha,
379 )
380 after_status = dict(before_status)
381 statuses = [before_status, after_status]
382 target_calls = 0
383
384 def read_status(*, name: str):
385 assert name == "local"
386 return statuses.pop(0)
387
388 def target_fingerprint() -> dict[str, str]:
389 nonlocal target_calls
390 target_calls += 1
391 return target
392
393 monkeypatch.setattr(supervisor, "read_install_status", read_status)
394 _patch_lease_state(monkeypatch, "held")
395
396 observation = supervisor._local_readiness_block_observation(
397 readiness=_local_readiness(
398 "missing-or-mismatched",
399 "manifest_missing",
400 install_state=install_state,
401 ),
402 fingerprint_json=desired_json,
403 fingerprint_sha256_value=desired_sha,
404 boot_required=True,
405 target_fingerprint=target_fingerprint,
406 )
407
408 _assert_local_observation(
409 observation,
410 phase="artifact-not-ready",
411 reason_code="install-in-progress",
412 desired_json=desired_json,
413 desired_sha=desired_sha,
414 boot_required=True,
415 detail=_local_install_progress_detail(
416 readiness_state=install_state,
417 attempt_id=str(before_status["attempt_id"]),
418 revision=int(before_status["revision"]),
419 ),
420 )
421 assert statuses == []
422 assert target_calls == 1
423
424
425@pytest.mark.parametrize("install_state", sorted(TERMINAL_STATES))
426def test_local_readiness_block_observation_returns_shared_object_for_terminal_pregate(
427 monkeypatch,
428 install_state: InstallState,
429) -> None:
430 sentinel = supervisor.ProviderTruthObservation(
431 provider="local",
432 phase="artifact-not-ready",
433 reason_code="artifact-missing",
434 detail={"sentinel": True},
435 )
436 monkeypatch.setattr(
437 supervisor,
438 "_readiness_block_observation",
439 lambda **_kwargs: sentinel,
440 )
441 monkeypatch.setattr(
442 supervisor,
443 "read_install_status",
444 lambda **_kwargs: pytest.fail("canonical status read not expected"),
445 )
446 monkeypatch.setattr(
447 "solstone.think.providers.install_lease.probe_install_lease_state",
448 lambda _provider: pytest.fail("lease probe not expected"),
449 )
450
451 result = supervisor._local_readiness_block_observation(
452 readiness=_local_readiness(
453 "missing-or-mismatched",
454 "manifest_missing",
455 install_state=install_state,
456 ),
457 fingerprint_json='{"provider":"local"}',
458 fingerprint_sha256_value="fp-local",
459 boot_required=True,
460 target_fingerprint=lambda: pytest.fail("target recompute not expected"),
461 )
462
463 assert result is sentinel
464
465
466@pytest.mark.parametrize(
467 "case",
468 [
469 "canonical-state",
470 "attempt-none",
471 "attempt-empty",
472 "target-mismatch",
473 "lease-free",
474 "lease-missing",
475 ],
476)
477def test_local_readiness_block_observation_returns_shared_object_for_failed_terms(
478 monkeypatch,
479 case: str,
480) -> None:
481 target = {"provider": "local", "target": "live-install"}
482 _desired_json, desired_sha = supervisor._target_fingerprint_pair(target)
483 before_status = _status_snapshot(target_fingerprint_sha256=desired_sha)
484 lease_state = "held"
485 if case == "canonical-state":
486 before_status["install_state"] = "installed"
487 elif case == "attempt-none":
488 before_status["attempt_id"] = None
489 elif case == "attempt-empty":
490 before_status["attempt_id"] = ""
491 elif case == "target-mismatch":
492 before_status["target_fingerprint_sha256"] = "fp-other"
493 elif case == "lease-free":
494 lease_state = "free"
495 elif case == "lease-missing":
496 lease_state = "missing"
497 else:
498 raise AssertionError(case)
499
500 sentinel = supervisor.ProviderTruthObservation(
501 provider="local",
502 phase="artifact-not-ready",
503 reason_code="artifact-missing",
504 detail={"sentinel": True},
505 )
506 monkeypatch.setattr(
507 supervisor,
508 "_readiness_block_observation",
509 lambda **_kwargs: sentinel,
510 )
511 monkeypatch.setattr(
512 supervisor, "read_install_status", lambda *, name: before_status
513 )
514 _patch_lease_state(monkeypatch, lease_state)
515
516 result = supervisor._local_readiness_block_observation(
517 readiness=_local_readiness(
518 "missing-or-mismatched",
519 "manifest_missing",
520 install_state="downloading",
521 ),
522 fingerprint_json='{"provider":"local"}',
523 fingerprint_sha256_value=desired_sha,
524 boot_required=True,
525 target_fingerprint=lambda: pytest.fail("target recompute not expected"),
526 )
527
528 assert result is sentinel
529
530
531def test_local_readiness_block_observation_maps_lease_probe_oserror(
532 monkeypatch,
533) -> None:
534 target = {"provider": "local", "target": "live-install"}
535 desired_json, desired_sha = supervisor._target_fingerprint_pair(target)
536 before_status = _status_snapshot(target_fingerprint_sha256=desired_sha)
537 monkeypatch.setattr(
538 supervisor, "read_install_status", lambda *, name: before_status
539 )
540
541 def fail_probe(_provider: str) -> str:
542 raise OSError("lease probe failed")
543
544 monkeypatch.setattr(
545 "solstone.think.providers.install_lease.probe_install_lease_state",
546 fail_probe,
547 )
548
549 observation = supervisor._local_readiness_block_observation(
550 readiness=_local_readiness(
551 "missing-or-mismatched",
552 "manifest_missing",
553 install_state="downloading",
554 ),
555 fingerprint_json=desired_json,
556 fingerprint_sha256_value=desired_sha,
557 boot_required=True,
558 target_fingerprint=lambda: pytest.fail("target recompute not expected"),
559 )
560
561 _assert_local_observation(
562 observation,
563 phase="state-unavailable",
564 reason_code="proof-observation-unavailable",
565 desired_json=desired_json,
566 desired_sha=desired_sha,
567 boot_required=True,
568 detail={
569 "readiness_status": "missing-or-mismatched",
570 "readiness_reason_code": "manifest_missing",
571 "error": "lease probe failed",
572 },
573 )
574
575
576@pytest.mark.parametrize(
577 "race",
578 ["desired", "revision", "attempt", "install-state", "install-target"],
579)
580def test_local_readiness_block_observation_reports_races(
581 monkeypatch,
582 race: str,
583) -> None:
584 target = {"provider": "local", "target": "live-install"}
585 desired_json, desired_sha = supervisor._target_fingerprint_pair(target)
586 after_target = target
587 before_status = _status_snapshot(target_fingerprint_sha256=desired_sha)
588 after_status = dict(before_status)
589 if race == "desired":
590 after_target = {"provider": "local", "target": "new-install"}
591 elif race == "revision":
592 after_status["revision"] = int(before_status["revision"]) + 1
593 elif race == "attempt":
594 after_status["attempt_id"] = "attempt-new"
595 elif race == "install-state":
596 after_status["install_state"] = "verifying"
597 elif race == "install-target":
598 after_status["target_fingerprint_sha256"] = "fp-new-target"
599 else:
600 raise AssertionError(race)
601 statuses = [before_status, after_status]
602
603 def read_status(*, name: str):
604 assert name == "local"
605 return statuses.pop(0)
606
607 monkeypatch.setattr(supervisor, "read_install_status", read_status)
608 _patch_lease_state(monkeypatch, "held")
609
610 observation = supervisor._local_readiness_block_observation(
611 readiness=_local_readiness(
612 "missing-or-mismatched",
613 "manifest_missing",
614 install_state="downloading",
615 ),
616 fingerprint_json=desired_json,
617 fingerprint_sha256_value=desired_sha,
618 boot_required=True,
619 target_fingerprint=lambda: after_target,
620 )
621
622 assert observation is not None
623 assert observation.phase == "observing"
624 assert observation.reason_code == "observation-raced"
625 assert observation.boot_required is True
626 assert observation.plan is None
627 _after_json, after_sha = supervisor._target_fingerprint_pair(after_target)
628 assert observation.detail == {
629 "before": supervisor._local_install_progress_identity(
630 desired_fingerprint_sha256=desired_sha,
631 install_status=before_status,
632 ),
633 "after": supervisor._local_install_progress_identity(
634 desired_fingerprint_sha256=after_sha,
635 install_status=after_status,
636 ),
637 }
638 assert statuses == []
639
640
641def _begin_downloading_local_install(target: dict[str, Any]) -> dict[str, Any]:
642 from solstone.think.providers.install_state import begin_or_replace_install_attempt
643
644 return begin_or_replace_install_attempt(
645 "local",
646 target,
647 initial_state="downloading",
648 )
649
650
651def test_mlx_local_observer_preserves_live_install_progress(
652 monkeypatch,
653) -> None:
654 from solstone.think.providers import mlx_install
655 from solstone.think.providers.install_lease import acquire_install_lease
656
657 target = {"provider": "local", "runtime": "mlx", "model": "mlx-test"}
658 desired_json, desired_sha = supervisor._target_fingerprint_pair(target)
659 status = _begin_downloading_local_install(target)
660 readiness = _local_readiness(
661 "missing-or-mismatched",
662 "manifest_missing",
663 install_state=status["install_state"],
664 )
665 forbidden_calls: list[str] = []
666
667 def fail_mlx_install(*_args, **_kwargs):
668 forbidden_calls.append("install_local_mlx")
669 pytest.fail("install_local_mlx not expected")
670
671 monkeypatch.setattr(mlx_install, "install_local_mlx", fail_mlx_install)
672 monkeypatch.setattr(mlx_install, "target_fingerprint", lambda: target)
673 monkeypatch.setattr(mlx_install, "inspect_readiness", lambda: readiness)
674 lease = acquire_install_lease("local")
675 assert lease is not None
676 try:
677 observation = supervisor._observe_mlx_local_provider_truth()
678 finally:
679 lease.release()
680
681 _assert_local_observation(
682 observation,
683 phase="artifact-not-ready",
684 reason_code="install-in-progress",
685 desired_json=desired_json,
686 desired_sha=desired_sha,
687 boot_required=True,
688 detail=_local_install_progress_detail(
689 readiness_state=status["install_state"],
690 attempt_id=str(status["attempt_id"]),
691 revision=int(status["revision"]),
692 ),
693 )
694 assert forbidden_calls == []
695
696 observation = supervisor._observe_mlx_local_provider_truth()
697
698 _assert_local_observation(
699 observation,
700 phase="artifact-not-ready",
701 reason_code="artifact-missing",
702 desired_json=desired_json,
703 desired_sha=desired_sha,
704 boot_required=True,
705 detail=_local_artifact_missing_detail(status["install_state"]),
706 )
707 assert forbidden_calls == []
708
709
710def test_linux_local_observer_preserves_live_install_progress(
711 monkeypatch,
712) -> None:
713 from solstone.think.providers import local_cuda, local_install, local_server
714 from solstone.think.providers.install_lease import acquire_install_lease
715
716 target = {"provider": "local", "runtime": "llama.cpp", "target": "linux-test"}
717 desired_json, desired_sha = supervisor._target_fingerprint_pair(target)
718 status = _begin_downloading_local_install(target)
719 readiness = _local_readiness(
720 "missing-or-mismatched",
721 "manifest_missing",
722 install_state=status["install_state"],
723 )
724 forbidden_calls: list[str] = []
725
726 def fail_local_install(*_args, **_kwargs):
727 forbidden_calls.append("install_local")
728 pytest.fail("install_local not expected")
729
730 def fail_select_server_tier(*_args, **_kwargs):
731 forbidden_calls.append("select_server_tier")
732 pytest.fail("select_server_tier not expected")
733
734 def fail_probe_nvidia_gpu(*_args, **_kwargs):
735 forbidden_calls.append("probe_nvidia_gpu")
736 pytest.fail("probe_nvidia_gpu not expected")
737
738 monkeypatch.setattr(local_install, "install_local", fail_local_install)
739 monkeypatch.setattr(local_server, "select_server_tier", fail_select_server_tier)
740 monkeypatch.setattr(local_cuda, "probe_nvidia_gpu", fail_probe_nvidia_gpu)
741 monkeypatch.setattr(local_install, "target_fingerprint", lambda _model: target)
742 monkeypatch.setattr(local_install, "inspect_readiness", lambda _model: readiness)
743 lease = acquire_install_lease("local")
744 assert lease is not None
745 try:
746 observation = supervisor._observe_linux_local_provider_truth()
747 finally:
748 lease.release()
749
750 _assert_local_observation(
751 observation,
752 phase="artifact-not-ready",
753 reason_code="install-in-progress",
754 desired_json=desired_json,
755 desired_sha=desired_sha,
756 boot_required=True,
757 detail=_local_install_progress_detail(
758 readiness_state=status["install_state"],
759 attempt_id=str(status["attempt_id"]),
760 revision=int(status["revision"]),
761 ),
762 )
763 assert forbidden_calls == []
764
765 observation = supervisor._observe_linux_local_provider_truth()
766
767 _assert_local_observation(
768 observation,
769 phase="artifact-not-ready",
770 reason_code="artifact-missing",
771 desired_json=desired_json,
772 desired_sha=desired_sha,
773 boot_required=True,
774 detail=_local_artifact_missing_detail(status["install_state"]),
775 )
776 assert forbidden_calls == []
777
778
779def test_linux_local_observer_missing_lease_does_not_create_or_mutate_status(
780 monkeypatch,
781) -> None:
782 from solstone.think.providers import local_install
783 from solstone.think.providers.install_lease import lease_path
784 from solstone.think.providers.install_state import provider_status_path
785
786 target = {"provider": "local", "runtime": "llama.cpp", "target": "linux-test"}
787 desired_json, desired_sha = supervisor._target_fingerprint_pair(target)
788 status = _begin_downloading_local_install(target)
789 readiness = _local_readiness(
790 "missing-or-mismatched",
791 "manifest_missing",
792 install_state=status["install_state"],
793 )
794 monkeypatch.setattr(local_install, "target_fingerprint", lambda _model: target)
795 monkeypatch.setattr(local_install, "inspect_readiness", lambda _model: readiness)
796 status_path = provider_status_path("local")
797 lease = lease_path("local")
798 assert not lease.exists()
799 before_bytes = status_path.read_bytes()
800 before_mtime = status_path.stat().st_mtime_ns
801
802 observation = supervisor._observe_linux_local_provider_truth()
803
804 _assert_local_observation(
805 observation,
806 phase="artifact-not-ready",
807 reason_code="artifact-missing",
808 desired_json=desired_json,
809 desired_sha=desired_sha,
810 boot_required=True,
811 detail=_local_artifact_missing_detail(status["install_state"]),
812 )
813 assert not lease.exists()
814 assert status_path.read_bytes() == before_bytes
815 assert status_path.stat().st_mtime_ns == before_mtime
816
817
818def test_local_ready_side_effects_submit_brain_refresh_once(monkeypatch) -> None:
819 from solstone.think.providers import local_server
820
821 plan = _local_plan()
822 state = supervisor.ProviderRuntimeState("local")
823 state.latest_plan = plan
824 queue = _FakeTaskQueue()
825 context_windows: list[int] = []
826
827 monkeypatch.setattr(supervisor, "_task_queue", queue)
828 monkeypatch.setattr(local_server, "reset_parallel_slots_cache", lambda: None)
829 monkeypatch.setattr(
830 local_server,
831 "write_local_context_window",
832 lambda tokens: context_windows.append(tokens),
833 )
834
835 supervisor._write_provider_ready_side_effects(
836 state,
837 supervisor.ProviderLaunchOutcome(
838 status="ready",
839 reason_code="probe-ready",
840 detail={},
841 ),
842 )
843
844 assert context_windows == [plan.context_tokens]
845 assert queue.submitted == [
846 (
847 [
848 "journal",
849 "brain",
850 "refresh",
851 "--expected-fingerprint",
852 plan.desired_fingerprint_sha256,
853 ],
854 (),
855 {},
856 )
857 ]
858
859
860def test_parakeet_ready_side_effects_do_not_submit_brain_refresh(monkeypatch) -> None:
861 from solstone.think.providers import parakeet_server
862
863 plan = _parakeet_plan("vulkan")
864 state = supervisor.ProviderRuntimeState("parakeet")
865 state.latest_plan = plan
866 queue = _FakeTaskQueue()
867 placements: list[str] = []
868
869 monkeypatch.setattr(supervisor, "_task_queue", queue)
870 monkeypatch.setattr(
871 parakeet_server,
872 "write_parakeet_placement",
873 lambda placement: placements.append(placement),
874 )
875
876 supervisor._write_provider_ready_side_effects(
877 state,
878 supervisor.ProviderLaunchOutcome(
879 status="ready",
880 reason_code="probe-ready",
881 detail={"placement": "gpu"},
882 ),
883 )
884
885 assert placements == ["gpu"]
886 assert queue.submitted == []
887
888
889def test_local_ready_without_local_launch_plan_does_not_submit_brain_refresh(
890 monkeypatch,
891) -> None:
892 state = supervisor.ProviderRuntimeState("local")
893 state.latest_plan = _parakeet_plan()
894 queue = _FakeTaskQueue()
895 monkeypatch.setattr(supervisor, "_task_queue", queue)
896
897 supervisor._write_provider_ready_side_effects(
898 state,
899 supervisor.ProviderLaunchOutcome(
900 status="ready",
901 reason_code="probe-ready",
902 detail={},
903 ),
904 )
905
906 assert queue.submitted == []
907
908
909def test_local_ready_side_effects_allow_missing_task_queue(monkeypatch) -> None:
910 from solstone.think.providers import local_server
911
912 plan = _local_plan()
913 state = supervisor.ProviderRuntimeState("local")
914 state.latest_plan = plan
915 context_windows: list[int] = []
916
917 monkeypatch.setattr(supervisor, "_task_queue", None)
918 monkeypatch.setattr(local_server, "reset_parallel_slots_cache", lambda: None)
919 monkeypatch.setattr(
920 local_server,
921 "write_local_context_window",
922 lambda tokens: context_windows.append(tokens),
923 )
924
925 supervisor._write_provider_ready_side_effects(
926 state,
927 supervisor.ProviderLaunchOutcome(
928 status="ready",
929 reason_code="probe-ready",
930 detail={},
931 ),
932 )
933
934 assert context_windows == [plan.context_tokens]
935
936
937def test_supervisor_bundled_fingerprint_matches_brain_preflight(monkeypatch) -> None:
938 target = {
939 "provider": "local",
940 "runtime": "llama.cpp",
941 "backend": "vulkan",
942 "runtime_pin": {"kind": "test-runtime", "revision": "1"},
943 "model_pin": {"kind": "test-model", "revision": "1"},
944 }
945 if sys.platform == "darwin":
946 from solstone.think.providers import mlx_install
947
948 monkeypatch.setattr(mlx_install, "target_fingerprint", lambda: target)
949 else:
950 from solstone.think.providers import local_install
951
952 monkeypatch.setattr(
953 local_install,
954 "target_fingerprint",
955 lambda _model_id=supervisor.LOCAL_MODEL: target,
956 )
957
958 _target_json, supervisor_sha = supervisor._target_fingerprint_pair(target)
959 result = build_active_brain_fingerprint(
960 {
961 "providers": {"active": {"provider": "local", "model": "local/bundled"}},
962 "env": {},
963 },
964 hmac_key=b"supervisor-brain-fingerprint-test",
965 )
966
967 assert result["ok"] is True
968 assert result["bundled_runtime_fingerprint_sha256"] == supervisor_sha
969
970
971def _runtime_record(
972 provider: str,
973 *,
974 phase: RuntimePhase,
975 fingerprint: str | None,
976 generation: int,
977 attempt: int,
978 process: dict[str, Any] | None = None,
979) -> RuntimeHealthRecord:
980 record = read_runtime_health(provider)
981 return {
982 **record,
983 "phase": phase,
984 "reason_code": None,
985 "detail": {},
986 "desired_fingerprint_sha256": fingerprint,
987 "incarnation": supervisor._PROVIDER_INCARNATION,
988 "generation": generation,
989 "attempt": attempt,
990 "process": process,
991 "updated_at": "2026-07-19T00:00:00+00:00",
992 "owner": {"test": "provider-runtime"},
993 }
994
995
996@pytest.mark.parametrize(
997 ("status", "expected_phase", "expected_reason"),
998 [
999 ("missing-or-mismatched", "artifact-not-ready", "artifact-missing"),
1000 (
1001 "proof-unavailable",
1002 "state-unavailable",
1003 "proof-observation-unavailable",
1004 ),
1005 ],
1006)
1007def test_readiness_block_table_maps_artifact_and_proof_statuses(
1008 status: str,
1009 expected_phase: RuntimePhase,
1010 expected_reason: ReasonCode,
1011) -> None:
1012 observation = supervisor._readiness_block_observation(
1013 provider="parakeet",
1014 readiness=_readiness(status, "manifest_missing"),
1015 fingerprint_json='{"provider":"parakeet"}',
1016 fingerprint_sha256_value="fp-parakeet",
1017 boot_required=True,
1018 )
1019
1020 assert observation is not None
1021 assert observation.phase == expected_phase
1022 assert observation.reason_code == expected_reason
1023
1024
1025@pytest.mark.parametrize(
1026 ("readiness_reason", "expected_reason"),
1027 [
1028 ("platform_unsupported", "platform-unsupported"),
1029 ("package_unavailable", "package-unavailable"),
1030 ("ram_insufficient", "ram-insufficient"),
1031 ("gpu_probe_failed", "gpu-probe-failed"),
1032 ("gpu_unavailable", "gpu-unavailable"),
1033 ],
1034)
1035def test_host_ineligible_reason_table(readiness_reason: str, expected_reason: str):
1036 observation = supervisor._readiness_block_observation(
1037 provider="local",
1038 readiness=_readiness(
1039 "host-ineligible",
1040 readiness_reason,
1041 host={"reason": readiness_reason},
1042 ),
1043 fingerprint_json='{"provider":"local"}',
1044 fingerprint_sha256_value="fp-local",
1045 boot_required=True,
1046 )
1047
1048 assert observation is not None
1049 assert observation.phase == "host-blocked"
1050 assert observation.reason_code == expected_reason
1051
1052
1053@pytest.mark.parametrize(
1054 "install_state",
1055 [
1056 "idle",
1057 "resolving",
1058 "downloading",
1059 "verifying",
1060 "installing",
1061 "installed",
1062 "failed",
1063 ],
1064)
1065def test_parakeet_missing_artifact_table_permits_acquisition_only_when_idle(
1066 install_state: InstallState,
1067) -> None:
1068 observation = supervisor._readiness_block_observation(
1069 provider="parakeet",
1070 readiness=_readiness(
1071 "missing-or-mismatched",
1072 "manifest_missing",
1073 install_state=install_state,
1074 ),
1075 fingerprint_json='{"provider":"parakeet"}',
1076 fingerprint_sha256_value="fp-parakeet",
1077 boot_required=True,
1078 )
1079
1080 assert observation is not None
1081 assert observation.phase == "artifact-not-ready"
1082 assert observation.detail["install_acquisition_allowed"] is (
1083 install_state == "idle"
1084 )
1085
1086
1087@pytest.mark.parametrize("active_provider", ["local", "cloud", None])
1088@pytest.mark.parametrize("endpoint_bundled", [True, False])
1089@pytest.mark.parametrize(
1090 ("readiness_status", "readiness_reason", "expected_phase"),
1091 [
1092 ("ready", "ready", "starting"),
1093 ("missing-or-mismatched", "manifest_missing", "artifact-not-ready"),
1094 (
1095 "proof-unavailable",
1096 "proof_cache_unavailable",
1097 "state-unavailable",
1098 ),
1099 ("host-ineligible", "gpu_unavailable", "host-blocked"),
1100 ],
1101)
1102def test_local_desired_state_table(
1103 monkeypatch,
1104 active_provider: str | None,
1105 endpoint_bundled: bool,
1106 readiness_status: str,
1107 readiness_reason: str,
1108 expected_phase: RuntimePhase,
1109) -> None:
1110 from solstone.think.providers import local_install, local_server, local_vulkan
1111
1112 config = {"providers": {"active": {"provider": active_provider}}}
1113 inspect_calls = 0
1114
1115 def inspect_readiness(_model_id):
1116 nonlocal inspect_calls
1117 inspect_calls += 1
1118 return _local_readiness(readiness_status, readiness_reason)
1119
1120 monkeypatch.setattr(supervisor, "_is_remote_mode", False)
1121 monkeypatch.setattr(supervisor.sys, "platform", "linux")
1122 monkeypatch.setattr(supervisor, "read_journal_config", lambda: config)
1123 monkeypatch.setattr(
1124 supervisor,
1125 "is_local_provider_needed",
1126 lambda config_arg=None: active_provider == "local",
1127 )
1128 monkeypatch.setattr(
1129 "solstone.think.providers.local_endpoint.resolve_local_endpoint",
1130 lambda: type("Endpoint", (), {"is_bundled": endpoint_bundled})(),
1131 )
1132 monkeypatch.setattr(
1133 local_install,
1134 "target_fingerprint",
1135 lambda _model_id: {"provider": "local", "target": "one"},
1136 )
1137 monkeypatch.setattr(local_install, "inspect_readiness", inspect_readiness)
1138 monkeypatch.setattr(local_install, "gpu_device_override", lambda: None)
1139 monkeypatch.setattr(
1140 local_vulkan,
1141 "detect_gpus",
1142 lambda: [
1143 local_vulkan.VulkanDevice(0, "GPU", local_vulkan.VK_TYPE_DISCRETE, 12288)
1144 ],
1145 )
1146 monkeypatch.setattr(
1147 local_vulkan, "select_device", lambda devices, **_kw: devices[0]
1148 )
1149 monkeypatch.setattr(local_vulkan, "device_local_used_mib", lambda _index: 0)
1150 monkeypatch.setattr(local_server, "select_server_tier", lambda _vram: _FakeTier())
1151
1152 observation = supervisor._observe_local_provider_truth()
1153
1154 if active_provider != "local" or not endpoint_bundled:
1155 assert observation.phase == "not-desired"
1156 assert observation.reason_code == "provider-not-needed"
1157 assert inspect_calls == 0
1158 else:
1159 assert observation.phase == expected_phase
1160 assert inspect_calls == 1
1161
1162
1163def test_local_desired_state_table_remote_mode_skips_bundled_observation(
1164 monkeypatch,
1165) -> None:
1166 from solstone.think.providers import local_install
1167
1168 monkeypatch.setattr(supervisor, "_is_remote_mode", True)
1169 monkeypatch.setattr(
1170 local_install,
1171 "inspect_readiness",
1172 lambda _model_id: pytest.fail("remote mode must not inspect bundled local"),
1173 )
1174
1175 observation = supervisor._observe_local_provider_truth()
1176
1177 assert observation.phase == "not-desired"
1178 assert observation.reason_code == "provider-not-needed"
1179 assert observation.detail["remote_mode"] is True
1180
1181
1182def test_local_desired_state_table_darwin_uses_mlx_observation(monkeypatch) -> None:
1183 from solstone.think.providers import local_install, mlx_install
1184
1185 monkeypatch.setattr(supervisor, "_is_remote_mode", False)
1186 monkeypatch.setattr(supervisor.sys, "platform", "darwin")
1187 monkeypatch.setattr(supervisor, "read_journal_config", lambda: {})
1188 monkeypatch.setattr(
1189 supervisor, "is_local_provider_needed", lambda _config=None: True
1190 )
1191 monkeypatch.setattr(
1192 "solstone.think.providers.local_endpoint.resolve_local_endpoint",
1193 lambda: type("Endpoint", (), {"is_bundled": True})(),
1194 )
1195 monkeypatch.setattr(
1196 local_install,
1197 "inspect_readiness",
1198 lambda _model_id: pytest.fail("darwin local observation must use MLX"),
1199 )
1200 monkeypatch.setattr(
1201 mlx_install,
1202 "target_fingerprint",
1203 lambda: {"provider": "local", "backend": "mlx"},
1204 )
1205 monkeypatch.setattr(
1206 mlx_install,
1207 "inspect_readiness",
1208 lambda: ReadinessOutcome(
1209 provider="local",
1210 status="ready",
1211 reason_code="ready",
1212 target={"model_id": supervisor.LOCAL_MODEL},
1213 install={"install_state": "idle"},
1214 host={"backend": "mlx"},
1215 artifacts={"runtime_dir": "/tmp/mlx-runtime"},
1216 proof={},
1217 ),
1218 )
1219
1220 observation = supervisor._observe_local_provider_truth()
1221
1222 assert observation.phase == "starting"
1223 assert observation.plan is not None
1224 assert observation.plan.backend == "mlx"
1225
1226
1227@pytest.mark.parametrize(
1228 (
1229 "remote_mode",
1230 "platform_can_host",
1231 "latch",
1232 "expected_phase",
1233 "expected_reason",
1234 "readiness_calls",
1235 ),
1236 [
1237 (
1238 True,
1239 True,
1240 None,
1241 "not-desired",
1242 "provider-not-needed",
1243 0,
1244 ),
1245 (
1246 False,
1247 False,
1248 None,
1249 "not-desired",
1250 "provider-not-needed",
1251 0,
1252 ),
1253 (
1254 False,
1255 True,
1256 {
1257 "desired": False,
1258 "blocked": False,
1259 "reason_code": "provider-not-needed",
1260 },
1261 "not-desired",
1262 "provider-not-needed",
1263 0,
1264 ),
1265 (
1266 False,
1267 True,
1268 {
1269 "desired": False,
1270 "blocked": True,
1271 "reason_code": "host-admission-blocked",
1272 },
1273 "host-blocked",
1274 "host-admission-blocked",
1275 0,
1276 ),
1277 (
1278 False,
1279 True,
1280 {
1281 "desired": True,
1282 "blocked": False,
1283 "reason_code": "provider-not-needed",
1284 },
1285 "starting",
1286 "launch-requested",
1287 1,
1288 ),
1289 ],
1290)
1291def test_parakeet_desired_state_table_remote_platform_and_stt(
1292 monkeypatch,
1293 remote_mode: bool,
1294 platform_can_host: bool,
1295 latch: dict[str, Any] | None,
1296 expected_phase: RuntimePhase,
1297 expected_reason: ReasonCode,
1298 readiness_calls: int,
1299) -> None:
1300 from solstone.think.providers import parakeet_install
1301
1302 calls = 0
1303 monkeypatch.setattr(supervisor, "_is_remote_mode", remote_mode)
1304 monkeypatch.setattr(
1305 supervisor, "_parakeet_platform_can_host", lambda: platform_can_host
1306 )
1307 monkeypatch.setattr(supervisor, "read_journal_config", lambda: {"transcribe": {}})
1308 monkeypatch.setattr(
1309 supervisor,
1310 "_parakeet_stt_admission_latch",
1311 lambda _transcribe, _confidential: latch,
1312 )
1313 monkeypatch.setattr(
1314 parakeet_install,
1315 "target_fingerprint",
1316 lambda *, journal_path=None: {"provider": "parakeet", "target": "one"},
1317 )
1318
1319 def inspect_readiness(_journal_path=None):
1320 nonlocal calls
1321 calls += 1
1322 return ReadinessOutcome(
1323 provider="parakeet",
1324 status="ready",
1325 reason_code="ready",
1326 target={},
1327 install={"install_state": "idle"},
1328 host={},
1329 artifacts={
1330 "binary_path_cpu": "/tmp/parakeet-cpu",
1331 "binary_path_vulkan": "/tmp/parakeet-vulkan",
1332 "model_path": "/tmp/parakeet-model.bin",
1333 },
1334 proof={},
1335 )
1336
1337 monkeypatch.setattr(parakeet_install, "inspect_readiness", inspect_readiness)
1338 monkeypatch.setattr(supervisor, "_configured_parakeet_device", lambda: "cpu")
1339 monkeypatch.setattr(
1340 supervisor,
1341 "_resolve_parakeet_backend",
1342 lambda _device, _selected: ("cpu", {}, None),
1343 )
1344 monkeypatch.setattr(supervisor, "parakeet_physical_thread_count", lambda: 4)
1345 monkeypatch.setattr(supervisor, "_parakeet_runtime_library_dirs", lambda: [])
1346
1347 observation = supervisor._observe_parakeet_provider_truth()
1348
1349 assert observation.phase == expected_phase
1350 assert observation.reason_code == expected_reason
1351 assert calls == readiness_calls
1352
1353
1354class _FakeTier:
1355 name = "floor"
1356 context_tokens = 16384
1357 parallel_slots = 1
1358 prompt_cache_mib = 0
1359
1360
1361def test_inactive_local_projection_publishes_not_desired_without_attempt(
1362 monkeypatch,
1363) -> None:
1364 observation = supervisor.ProviderTruthObservation(
1365 provider="local",
1366 phase="not-desired",
1367 reason_code="provider-not-needed",
1368 detail={"projection": {"status": "reconciling"}},
1369 )
1370 starts: list[int] = []
1371 monkeypatch.setattr(supervisor, "_provider_executor", lambda: _InlineExecutor())
1372 monkeypatch.setattr(
1373 supervisor, "_observe_local_provider_truth", lambda: observation
1374 )
1375 monkeypatch.setattr(
1376 supervisor,
1377 "_provider_start_worker",
1378 lambda provider, plan_arg, fence, _cancel_event: starts.append(fence.attempt),
1379 )
1380
1381 asyncio.run(supervisor._reconcile_local_provider_runtime([]))
1382 asyncio.run(supervisor._reconcile_local_provider_runtime([]))
1383
1384 state = supervisor._provider_runtime_states["local"]
1385 record = read_runtime_health("local")
1386 assert state.latest_phase == "not-desired"
1387 assert state.retry.attempt_count == 0
1388 assert starts == []
1389 assert record["phase"] == "not-desired"
1390 assert record["detail"]["projection"]["status"] == "reconciling"
1391
1392
1393def test_byo_local_performs_no_bundled_observation(monkeypatch):
1394 from solstone.think.providers import local_install
1395
1396 monkeypatch.setattr(supervisor, "_is_remote_mode", False)
1397 monkeypatch.setattr(supervisor, "read_journal_config", lambda: {})
1398 monkeypatch.setattr(
1399 supervisor, "is_local_provider_needed", lambda _config=None: True
1400 )
1401 monkeypatch.setattr(
1402 "solstone.think.providers.local_endpoint.resolve_local_endpoint",
1403 lambda: type("Endpoint", (), {"is_bundled": False})(),
1404 )
1405 monkeypatch.setattr(
1406 local_install,
1407 "inspect_readiness",
1408 lambda _model_id: pytest.fail(
1409 "BYO endpoint must not inspect bundled readiness"
1410 ),
1411 )
1412
1413 observation = supervisor._observe_local_provider_truth()
1414
1415 assert observation.phase == "not-desired"
1416 assert observation.detail["projection"]["status"] == "reconciling"
1417
1418
1419def test_truth_observation_runs_off_tick_and_single_flight(monkeypatch):
1420 started = threading.Event()
1421 release = threading.Event()
1422 calls = 0
1423
1424 def slow_observer():
1425 nonlocal calls
1426 calls += 1
1427 started.set()
1428 release.wait(timeout=5)
1429 return supervisor.ProviderTruthObservation(
1430 provider="local",
1431 phase="not-desired",
1432 reason_code="provider-not-needed",
1433 detail={},
1434 )
1435
1436 monkeypatch.setattr(supervisor, "_observe_local_provider_truth", slow_observer)
1437
1438 asyncio.run(supervisor._reconcile_local_provider_runtime([]))
1439
1440 assert started.wait(timeout=1)
1441 state = supervisor._provider_runtime_states["local"]
1442 assert state.truth_future is not None
1443 assert not state.truth_future.done()
1444
1445 asyncio.run(supervisor._reconcile_local_provider_runtime([]))
1446
1447 assert calls == 1
1448 release.set()
1449 state.truth_future.result(timeout=1)
1450 asyncio.run(supervisor._reconcile_local_provider_runtime([]))
1451
1452 assert state.latest_phase == "not-desired"
1453 assert read_runtime_health("local")["phase"] == "not-desired"
1454
1455
1456def test_no_op_tick_waits_for_truth_cadence_and_backoff_deadline(monkeypatch) -> None:
1457 plan = _local_plan()
1458 state = supervisor._provider_runtime_states["local"]
1459 state.latest_phase = "backoff"
1460 state.latest_plan = plan
1461 state.desired_fingerprint = plan.desired_fingerprint_sha256
1462 state.retry = supervisor.ProviderRetryState(
1463 attempt_count=1,
1464 next_at=200.0,
1465 desired_fingerprint=plan.desired_fingerprint_sha256,
1466 )
1467 state.next_truth_at = 160.0
1468
1469 monkeypatch.setattr(supervisor.time, "monotonic", lambda: 100.0)
1470 monkeypatch.setattr(
1471 supervisor,
1472 "_provider_start_worker",
1473 lambda *_args: pytest.fail("backoff deadline has not arrived"),
1474 )
1475 monkeypatch.setattr(
1476 supervisor,
1477 "_observe_local_provider_truth",
1478 lambda: pytest.fail("truth cadence has not arrived"),
1479 )
1480
1481 asyncio.run(supervisor._reconcile_local_provider_runtime([]))
1482
1483 assert state.start_future is None
1484 assert state.truth_future is None
1485 assert state.latest_phase == "backoff"
1486
1487
1488def test_start_worker_is_single_flight(monkeypatch) -> None:
1489 plan = _local_plan()
1490 pending: concurrent.futures.Future = concurrent.futures.Future()
1491 submitted = 0
1492 state = supervisor._provider_runtime_states["local"]
1493 state.latest_phase = "starting"
1494 state.latest_plan = plan
1495 state.desired_fingerprint = plan.desired_fingerprint_sha256
1496 state.retry.desired_fingerprint = plan.desired_fingerprint_sha256
1497 state.next_truth_at = 9999.0
1498
1499 class _PendingExecutor:
1500 def submit(self, *_args, **_kwargs):
1501 nonlocal submitted
1502 submitted += 1
1503 return pending
1504
1505 monkeypatch.setattr(supervisor.time, "monotonic", lambda: 100.0)
1506 monkeypatch.setattr(supervisor, "_provider_executor", lambda: _PendingExecutor())
1507
1508 asyncio.run(supervisor._reconcile_local_provider_runtime([]))
1509 asyncio.run(supervisor._reconcile_local_provider_runtime([]))
1510
1511 assert submitted == 1
1512 assert state.start_future is pending
1513 assert state.retry.attempt_count == 1
1514
1515
1516def test_ready_episode_reobserves_on_sixty_second_cadence(monkeypatch) -> None:
1517 now = 100.0
1518 observations = 0
1519 state = supervisor._provider_runtime_states["local"]
1520 state.latest_phase = "ready"
1521 state.desired_fingerprint = "fp-local"
1522 state.next_truth_at = now + supervisor.PROVIDER_STABLE_READY_REFRESH_SECONDS
1523
1524 def monotonic() -> float:
1525 return now
1526
1527 def observe():
1528 nonlocal observations
1529 observations += 1
1530 return supervisor.ProviderTruthObservation(
1531 provider="local",
1532 phase="ready",
1533 reason_code="ready-existing-owned-process",
1534 detail={},
1535 desired_fingerprint_sha256="fp-local",
1536 boot_required=True,
1537 )
1538
1539 monkeypatch.setattr(supervisor.time, "monotonic", monotonic)
1540 monkeypatch.setattr(supervisor, "_provider_executor", lambda: _InlineExecutor())
1541 monkeypatch.setattr(supervisor, "_observe_local_provider_truth", observe)
1542
1543 asyncio.run(supervisor._reconcile_local_provider_runtime([]))
1544
1545 assert observations == 0
1546
1547 now = 160.0
1548 asyncio.run(supervisor._reconcile_local_provider_runtime([]))
1549 asyncio.run(supervisor._reconcile_local_provider_runtime([]))
1550
1551 assert observations == 1
1552 assert state.latest_phase in {"ready", "ready-proof-unavailable"}
1553
1554
1555@pytest.mark.parametrize(
1556 ("provider", "plan", "managed_name"),
1557 [
1558 ("local", _local_plan(), supervisor.LOCAL_SERVER_PROCESS_NAME),
1559 ("parakeet", _parakeet_plan("vulkan"), supervisor.PARAKEET_SERVER_PROCESS_NAME),
1560 ],
1561)
1562def test_ready_truth_refresh_keeps_same_target_process_authoritative(
1563 monkeypatch,
1564 provider: str,
1565 plan: supervisor.LocalServerLaunchPlan | supervisor.ParakeetServerLaunchPlan,
1566 managed_name: str,
1567) -> None:
1568 state = supervisor._provider_runtime_states[provider]
1569 managed = _FakeManaged(managed_name)
1570 process = {
1571 "name": managed.name,
1572 "pid": managed.process.pid,
1573 "ref": managed.ref,
1574 "port": 45678,
1575 }
1576 _set_provider_ready(provider, state, plan)
1577 state.next_truth_at = 0.0
1578 state.next_probe_at = 10**12
1579 write_runtime_health(
1580 _runtime_record(
1581 provider,
1582 phase="ready",
1583 fingerprint=plan.desired_fingerprint_sha256,
1584 generation=state.generation,
1585 attempt=state.retry.attempt_count,
1586 process=process,
1587 )
1588 )
1589 observation = supervisor.ProviderTruthObservation(
1590 provider=provider,
1591 phase="starting",
1592 reason_code="launch-requested",
1593 detail={"source": "stable-refresh"},
1594 desired_fingerprint_json=plan.desired_fingerprint_json,
1595 desired_fingerprint_sha256=plan.desired_fingerprint_sha256,
1596 plan=plan,
1597 boot_required=True,
1598 )
1599 monkeypatch.setattr(supervisor.time, "monotonic", lambda: 100.0)
1600 monkeypatch.setattr(supervisor, "_provider_executor", lambda: _InlineExecutor())
1601 monkeypatch.setattr(
1602 supervisor,
1603 "_observe_provider_truth",
1604 lambda provider_arg: observation,
1605 )
1606
1607 asyncio.run(supervisor._reconcile_provider_runtime(provider, [managed]))
1608 submitted_health = read_runtime_health(provider)
1609 assert state.latest_phase in {"ready", "ready-proof-unavailable"}
1610 assert submitted_health["phase"] == "ready"
1611 assert submitted_health["process"] == process
1612
1613 asyncio.run(supervisor._reconcile_provider_runtime(provider, [managed]))
1614 refreshed_health = read_runtime_health(provider)
1615 assert state.latest_phase == "ready"
1616 assert state.latest_plan is plan
1617 assert state.pending_stop_request is None
1618 assert state.stop_cleanup_future is None
1619 assert refreshed_health["phase"] == "ready"
1620 assert refreshed_health["process"] == process
1621 managed.terminate.assert_not_called()
1622 managed.cleanup.assert_not_called()
1623
1624 state.truth_fence = supervisor._provider_fence(state, state.retry.attempt_count)
1625 state.truth_future = _future_with(observation)
1626 state.generation += 1
1627 assert supervisor._handle_provider_truth_result(state) is True
1628 fenced_health = read_runtime_health(provider)
1629 assert fenced_health["reason_code"] == "stale-result-ignored"
1630 assert fenced_health["process"] == process
1631
1632
1633def test_ready_truth_refresh_to_not_desired_retains_stop_ownership(
1634 monkeypatch,
1635) -> None:
1636 plan = _local_plan()
1637 managed = _FakeManaged()
1638 state = supervisor._provider_runtime_states["local"]
1639 process = {
1640 "name": managed.name,
1641 "pid": managed.process.pid,
1642 "ref": managed.ref,
1643 "port": 45678,
1644 }
1645 _set_provider_ready("local", state, plan)
1646 state.next_truth_at = 0.0
1647 state.next_probe_at = 10**12
1648 write_runtime_health(
1649 _runtime_record(
1650 "local",
1651 phase="ready",
1652 fingerprint=plan.desired_fingerprint_sha256,
1653 generation=state.generation,
1654 attempt=state.retry.attempt_count,
1655 process=process,
1656 )
1657 )
1658 observation = supervisor.ProviderTruthObservation(
1659 provider="local",
1660 phase="not-desired",
1661 reason_code="provider-not-needed",
1662 detail={"active_provider": "openai"},
1663 boot_required=False,
1664 )
1665 monkeypatch.setattr(supervisor.time, "monotonic", lambda: 100.0)
1666 monkeypatch.setattr(supervisor, "_provider_executor", lambda: _InlineExecutor())
1667 monkeypatch.setattr(
1668 supervisor,
1669 "_observe_provider_truth",
1670 lambda _provider: observation,
1671 )
1672
1673 supervisor._submit_provider_truth_if_needed(state)
1674 assert state.latest_phase == "ready"
1675 assert read_runtime_health("local")["process"] == process
1676
1677 assert supervisor._handle_provider_truth_result(state) is True
1678 assert state.latest_phase == "stop-deferred"
1679 assert state.pending_stop_target_phase == "not-desired"
1680 assert state.pending_stop_admission_exclusive is True
1681
1682 request = supervisor._deferred_stop_request(state, [managed])
1683 assert request is not None
1684 supervisor._set_provider_pending_stop_request(state, request)
1685 state.next_truth_at = 0.0
1686 supervisor._submit_provider_truth_if_needed(state)
1687 assert state.latest_phase == "stop-deferred"
1688 assert supervisor._handle_provider_truth_result(state) is True
1689
1690 health = read_runtime_health("local")
1691 assert state.latest_phase == "stop-deferred"
1692 assert state.pending_stop_request is request
1693 assert state.pending_stop_request.managed is managed
1694 assert health["phase"] == "stop-deferred"
1695 assert health["process"] == process
1696
1697
1698def test_retry_token_resets_live_target_without_launching(monkeypatch) -> None:
1699 plan = _local_plan()
1700 observations = 0
1701 state = supervisor._provider_runtime_states["local"]
1702 state.latest_phase = "ready"
1703 state.latest_plan = plan
1704 state.desired_fingerprint = plan.desired_fingerprint_sha256
1705 state.retry = supervisor.ProviderRetryState(
1706 attempt_count=4,
1707 next_at=9999.0,
1708 desired_fingerprint=plan.desired_fingerprint_sha256,
1709 )
1710 state.next_truth_at = 9999.0
1711 managed = _FakeManaged()
1712 request_retry_token(
1713 "local",
1714 desired_fingerprint_sha256=plan.desired_fingerprint_sha256,
1715 owner={"test": "retry"},
1716 )
1717 monkeypatch.setattr(
1718 supervisor,
1719 "_provider_start_worker",
1720 lambda *_args: pytest.fail("live retry token must not duplicate launch"),
1721 )
1722
1723 def observe():
1724 nonlocal observations
1725 observations += 1
1726 return supervisor.ProviderTruthObservation(
1727 provider="local",
1728 phase="ready",
1729 reason_code="ready-existing-owned-process",
1730 detail={},
1731 desired_fingerprint_sha256=plan.desired_fingerprint_sha256,
1732 boot_required=True,
1733 )
1734
1735 monkeypatch.setattr(supervisor, "_provider_executor", lambda: _InlineExecutor())
1736 monkeypatch.setattr(supervisor, "_observe_local_provider_truth", observe)
1737
1738 asyncio.run(supervisor._reconcile_local_provider_runtime([managed]))
1739 asyncio.run(supervisor._reconcile_local_provider_runtime([managed]))
1740
1741 assert state.retry.attempt_count == 0
1742 assert state.latest_phase == "stopped"
1743 assert observations == 1
1744
1745
1746def test_owner_retry_is_nonterminal_before_token_is_cleared(monkeypatch) -> None:
1747 state = supervisor._provider_runtime_states["local"]
1748 state.latest_phase = "failed"
1749 state.desired_fingerprint = "fp-local"
1750 state.retry = supervisor.ProviderRetryState(
1751 attempt_count=len(supervisor.PROVIDER_RETRY_SCHEDULE_SECONDS),
1752 desired_fingerprint="fp-local",
1753 )
1754 failed = supervisor._write_provider_runtime(
1755 state,
1756 phase="failed",
1757 reason_code="launch-budget-exhausted",
1758 detail={},
1759 )
1760 assert failed is not None
1761 token = request_runtime_retry(
1762 "local",
1763 expected_health_revision=failed["revision"],
1764 expected_retry_revision=0,
1765 desired_fingerprint_sha256="fp-local",
1766 )
1767
1768 real_consume = supervisor.consume_retry_token
1769
1770 def assert_nonterminal_before_consume(*args, **kwargs):
1771 health = read_runtime_health("local")
1772 assert health["phase"] == "observing"
1773 with pytest.raises(
1774 supervisor.RuntimeHealthConflictError,
1775 match="terminal failure",
1776 ):
1777 request_runtime_retry(
1778 "local",
1779 expected_health_revision=health["revision"],
1780 expected_retry_revision=token["revision"],
1781 desired_fingerprint_sha256="fp-local",
1782 )
1783 return real_consume(*args, **kwargs)
1784
1785 monkeypatch.setattr(
1786 supervisor, "consume_retry_token", assert_nonterminal_before_consume
1787 )
1788
1789 supervisor._handle_provider_retry_token(state)
1790
1791 assert read_retry_token("local")["token_id"] is None
1792 assert state.retry.attempt_count == 0
1793
1794
1795@pytest.mark.parametrize(
1796 "observation",
1797 [
1798 supervisor.ProviderTruthObservation(
1799 provider="local",
1800 phase="starting",
1801 reason_code="launch-requested",
1802 detail={},
1803 desired_fingerprint_json='{"provider":"local","target":"new"}',
1804 desired_fingerprint_sha256="fp-new",
1805 plan=_local_plan(),
1806 boot_required=True,
1807 ),
1808 supervisor.ProviderTruthObservation(
1809 provider="local",
1810 phase="not-desired",
1811 reason_code="provider-not-needed",
1812 detail={},
1813 ),
1814 supervisor.ProviderTruthObservation(
1815 provider="local",
1816 phase="state-corrupt",
1817 reason_code="record-malformed",
1818 detail={},
1819 desired_fingerprint_sha256="fp-local",
1820 boot_required=True,
1821 ),
1822 supervisor.ProviderTruthObservation(
1823 provider="local",
1824 phase="state-unavailable",
1825 reason_code="record-unavailable",
1826 detail={},
1827 desired_fingerprint_sha256="fp-local",
1828 boot_required=True,
1829 ),
1830 ],
1831)
1832def test_truth_change_signals_pending_start_cancel_event(
1833 observation: supervisor.ProviderTruthObservation,
1834) -> None:
1835 state = supervisor._provider_runtime_states["local"]
1836 state.latest_phase = "starting"
1837 state.desired_fingerprint = "fp-local"
1838 state.retry.desired_fingerprint = "fp-local"
1839 state.start_fence = supervisor._provider_fence(state, 0)
1840 state.start_future = concurrent.futures.Future()
1841 state.start_cancel_event = threading.Event()
1842 state.truth_fence = supervisor._provider_fence(state, 0)
1843 state.truth_future = _future_with(observation)
1844
1845 assert supervisor._handle_provider_truth_result(state) is True
1846
1847 assert state.start_cancel_event is not None
1848 assert state.start_cancel_event.is_set()
1849
1850
1851@pytest.mark.parametrize(
1852 ("provider", "phase", "plan", "observation"),
1853 [
1854 (
1855 "local",
1856 "starting",
1857 _cuda_plan(),
1858 supervisor.ProviderTruthObservation(
1859 provider="local",
1860 phase="host-blocked",
1861 reason_code="gpu-unavailable",
1862 detail={"host": {"reason": "transient cuda pressure"}},
1863 desired_fingerprint_sha256="fp-local-cuda",
1864 boot_required=True,
1865 ),
1866 ),
1867 (
1868 "parakeet",
1869 "starting",
1870 _parakeet_plan("cpu"),
1871 supervisor.ProviderTruthObservation(
1872 provider="parakeet",
1873 phase="starting",
1874 reason_code="launch-requested",
1875 detail={"placement": "gpu"},
1876 desired_fingerprint_sha256="fp-parakeet",
1877 plan=_parakeet_plan("vulkan"),
1878 boot_required=True,
1879 ),
1880 ),
1881 (
1882 "parakeet",
1883 "ready",
1884 _parakeet_plan("cpu"),
1885 supervisor.ProviderTruthObservation(
1886 provider="parakeet",
1887 phase="starting",
1888 reason_code="launch-requested",
1889 detail={"placement": "gpu"},
1890 desired_fingerprint_sha256="fp-parakeet",
1891 plan=_parakeet_plan("vulkan"),
1892 boot_required=True,
1893 ),
1894 ),
1895 ],
1896)
1897def test_same_target_transient_observation_keeps_captured_plan_authoritative(
1898 provider: str,
1899 phase: RuntimePhase,
1900 plan: supervisor.LocalServerLaunchPlan | supervisor.ParakeetServerLaunchPlan,
1901 observation: supervisor.ProviderTruthObservation,
1902) -> None:
1903 state = supervisor._provider_runtime_states[provider]
1904 state.latest_phase = phase
1905 state.latest_plan = plan
1906 state.desired_fingerprint = plan.desired_fingerprint_sha256
1907 state.retry.desired_fingerprint = plan.desired_fingerprint_sha256
1908 state.truth_fence = supervisor._provider_fence(state, 0)
1909 state.truth_future = _future_with(observation)
1910 if phase == "starting":
1911 state.start_fence = supervisor._provider_fence(state, 0)
1912 state.start_future = concurrent.futures.Future()
1913 state.start_cancel_event = threading.Event()
1914
1915 assert supervisor._handle_provider_truth_result(state) is True
1916
1917 if state.start_cancel_event is not None:
1918 assert not state.start_cancel_event.is_set()
1919 assert state.latest_plan is plan
1920 assert state.latest_phase == phase
1921
1922
1923def test_owned_local_host_blocked_defers_admission_exclusive_stop() -> None:
1924 plan = _cuda_plan()
1925 state = supervisor._provider_runtime_states["local"]
1926 state.latest_phase = "ready"
1927 state.latest_plan = plan
1928 state.desired_fingerprint = plan.desired_fingerprint_sha256
1929 state.retry.desired_fingerprint = plan.desired_fingerprint_sha256
1930 state.truth_fence = supervisor._provider_fence(state, 0)
1931 state.truth_future = _future_with(
1932 supervisor.ProviderTruthObservation(
1933 provider="local",
1934 phase="host-blocked",
1935 reason_code="gpu-unavailable",
1936 detail={"host": {"reason": "transient cuda pressure"}},
1937 desired_fingerprint_sha256=plan.desired_fingerprint_sha256,
1938 boot_required=True,
1939 )
1940 )
1941
1942 assert supervisor._handle_provider_truth_result(state) is True
1943
1944 assert state.latest_plan is plan
1945 assert state.latest_phase == "stop-deferred"
1946 assert state.pending_stop_admission_exclusive is True
1947 assert state.pending_stop_target_phase == "host-blocked"
1948
1949
1950def test_parakeet_stt_admission_latch_survives_ready_probe_and_restart(
1951 monkeypatch,
1952) -> None:
1953 monkeypatch.setattr(supervisor.sys, "platform", "linux")
1954 monkeypatch.setattr(supervisor.platform, "machine", lambda: "x86_64")
1955 monkeypatch.setattr(supervisor, "local_stt_backend", lambda: "parakeet")
1956 monkeypatch.setattr(supervisor, "stt_local_floor_bytes", lambda: 4 * 1024**3)
1957 transcribe: dict[str, Any] = {}
1958 admission_input = supervisor._parakeet_stt_admission_input(transcribe, False)
1959 _input_json, input_sha = supervisor._target_fingerprint_pair(admission_input)
1960 latch = {
1961 "input_json": "{}",
1962 "input_sha256": input_sha,
1963 "retry_epoch": 0,
1964 "choice": "parakeet",
1965 "desired": True,
1966 "blocked": False,
1967 "reason_code": "launch-requested",
1968 }
1969 plan = _parakeet_plan("cpu")
1970 state = supervisor._provider_runtime_states["parakeet"]
1971 state.desired_fingerprint = plan.desired_fingerprint_sha256
1972 state.generation = 1
1973 state.retry.attempt_count = 1
1974 write_runtime_health(
1975 {
1976 **read_runtime_health("parakeet"),
1977 "phase": "starting",
1978 "reason_code": "launch-requested",
1979 "detail": {"stt_admission_latch": latch},
1980 "desired_fingerprint_sha256": plan.desired_fingerprint_sha256,
1981 "incarnation": supervisor._PROVIDER_INCARNATION,
1982 "generation": 1,
1983 "attempt": 1,
1984 "process": None,
1985 "updated_at": "2026-07-19T00:00:00+00:00",
1986 "owner": {"test": "latch"},
1987 }
1988 )
1989
1990 supervisor._write_provider_runtime(
1991 state,
1992 phase="ready",
1993 reason_code="probe-ready",
1994 detail={"backend": "cpu", "port": 45678},
1995 process={
1996 "name": supervisor.PARAKEET_SERVER_PROCESS_NAME,
1997 "pid": 12345,
1998 "ref": "ref-parakeet",
1999 "port": 45678,
2000 },
2001 )
2002 supervisor._write_provider_runtime(
2003 state,
2004 phase="ready-proof-unavailable",
2005 reason_code="proof-observation-unavailable",
2006 detail={"health_state": "failed"},
2007 process={
2008 "name": supervisor.PARAKEET_SERVER_PROCESS_NAME,
2009 "pid": 12345,
2010 "ref": "ref-parakeet",
2011 "port": 45678,
2012 },
2013 )
2014 monkeypatch.setattr(
2015 supervisor,
2016 "read_available_bytes",
2017 lambda: pytest.fail("valid latch must avoid point-in-time RAM recheck"),
2018 )
2019
2020 recovered = supervisor._parakeet_stt_admission_latch(transcribe, False)
2021
2022 assert recovered == latch
2023 assert read_runtime_health("parakeet")["detail"]["stt_admission_latch"] == latch
2024
2025
2026def test_handle_shutdown_signals_pending_provider_start(monkeypatch) -> None:
2027 state = supervisor._provider_runtime_states["local"]
2028 event = threading.Event()
2029 state.start_cancel_event = event
2030 state.start_fence = supervisor.ProviderFence(
2031 incarnation=supervisor._PROVIDER_INCARNATION,
2032 generation=1,
2033 fingerprint="fp-local",
2034 attempt=1,
2035 )
2036 monkeypatch.setattr(supervisor, "_managed_procs", [])
2037 monkeypatch.setattr(supervisor, "shutdown_requested", False)
2038
2039 try:
2040 with pytest.raises(KeyboardInterrupt):
2041 supervisor.handle_shutdown(15, None)
2042 finally:
2043 supervisor.shutdown_requested = False
2044
2045 assert event.is_set()
2046
2047
2048def test_local_probe_worker_reports_loading_without_opening_socket(monkeypatch) -> None:
2049 from solstone.think.providers import local_server
2050
2051 calls: list[int] = []
2052
2053 def fake_probe(port: int):
2054 calls.append(port)
2055 return local_server.STATE_LOADING, None
2056
2057 monkeypatch.setattr(local_server, "_probe_health", fake_probe)
2058
2059 outcome = supervisor._provider_probe_worker(
2060 "local",
2061 45678,
2062 supervisor.ProviderFence(
2063 incarnation=supervisor._PROVIDER_INCARNATION,
2064 generation=1,
2065 fingerprint="fp-local",
2066 attempt=1,
2067 ),
2068 )
2069
2070 assert calls == [45678]
2071 assert outcome.status == "not-ready"
2072 assert outcome.reason_code == "proof-observation-unavailable"
2073 assert outcome.detail["health_state"] == local_server.STATE_LOADING
2074
2075
2076def test_parakeet_probe_worker_has_no_loading_state(monkeypatch) -> None:
2077 from solstone.think.providers import parakeet_server
2078
2079 calls: list[int] = []
2080
2081 def fake_probe(port: int):
2082 calls.append(port)
2083 return parakeet_server.STATE_FAILED, "warming"
2084
2085 monkeypatch.setattr(parakeet_server, "_probe_health", fake_probe)
2086
2087 outcome = supervisor._provider_probe_worker(
2088 "parakeet",
2089 45679,
2090 supervisor.ProviderFence(
2091 incarnation=supervisor._PROVIDER_INCARNATION,
2092 generation=1,
2093 fingerprint="fp-parakeet",
2094 attempt=1,
2095 ),
2096 )
2097
2098 assert calls == [45679]
2099 assert outcome.status == "unavailable"
2100 assert outcome.reason_code == "proof-observation-unavailable"
2101 assert outcome.detail["health_state"] == parakeet_server.STATE_FAILED
2102
2103
2104def test_probe_slot_transitions_ready_and_proof_unavailable_with_fakes(
2105 monkeypatch,
2106) -> None:
2107 from solstone.think.providers import local_server
2108
2109 managed = _FakeManaged()
2110 state = supervisor._provider_runtime_states["local"]
2111 plan = _local_plan()
2112 _set_provider_ready("local", state, plan)
2113 state.next_truth_at = 10**12
2114 state.next_probe_at = 0.0
2115 supervisor.write_service_port("local", 45678)
2116 write_runtime_health(
2117 _runtime_record(
2118 "local",
2119 phase="ready",
2120 fingerprint=plan.desired_fingerprint_sha256,
2121 generation=1,
2122 attempt=1,
2123 process={
2124 "name": managed.name,
2125 "pid": managed.process.pid,
2126 "ref": managed.ref,
2127 "port": 45678,
2128 },
2129 )
2130 )
2131 observations = [
2132 (local_server.STATE_LOADING, None),
2133 (local_server.STATE_READY, None),
2134 ]
2135 calls: list[int] = []
2136
2137 def fake_probe(port: int):
2138 calls.append(port)
2139 return observations.pop(0)
2140
2141 monkeypatch.setattr(local_server, "_probe_health", fake_probe)
2142 monkeypatch.setattr(supervisor, "_provider_executor", lambda: _InlineExecutor())
2143 monkeypatch.setattr(supervisor.time, "monotonic", lambda: 100.0)
2144
2145 asyncio.run(supervisor._reconcile_local_provider_runtime([managed]))
2146 asyncio.run(supervisor._reconcile_local_provider_runtime([managed]))
2147
2148 assert state.latest_phase == "ready-proof-unavailable"
2149 assert calls == [45678]
2150 assert read_runtime_health("local")["process"]["ref"] == managed.ref
2151
2152 state.next_probe_at = 0.0
2153 asyncio.run(supervisor._reconcile_local_provider_runtime([managed]))
2154 asyncio.run(supervisor._reconcile_local_provider_runtime([managed]))
2155
2156 assert state.latest_phase == "ready"
2157 assert calls == [45678, 45678]
2158
2159
2160def test_wedge_threshold_records_recycle_token_without_sync_termination(
2161 monkeypatch,
2162) -> None:
2163 from solstone.think.providers import local_endpoint, local_server
2164 from solstone.think.providers.local_endpoint import LocalEndpoint
2165
2166 state = supervisor._provider_runtime_states["local"]
2167 plan = _local_plan()
2168 _set_provider_ready("local", state, plan)
2169 state.generation = 4
2170 state.next_truth_at = 9999.0
2171 managed = _FakeManaged()
2172 supervisor._managed_procs = [managed]
2173 supervisor._SERVICE_STATE[managed.name] = {
2174 "restart": False,
2175 "shutdown_timeout": 15,
2176 }
2177 monkeypatch.setattr(
2178 local_endpoint,
2179 "resolve_local_endpoint",
2180 lambda: LocalEndpoint("", "", None, True),
2181 )
2182 monkeypatch.setattr(supervisor, "read_service_port", lambda service: 45678)
2183 monkeypatch.setattr(
2184 local_server,
2185 "_probe_health",
2186 lambda port: (local_server.STATE_READY, None),
2187 )
2188 monkeypatch.setattr(
2189 supervisor,
2190 "_restart_service",
2191 lambda *_args, **_kwargs: pytest.fail("wedge must not restart synchronously"),
2192 )
2193 monkeypatch.setattr(
2194 supervisor,
2195 "_start_termination_thread",
2196 lambda *_args, **_kwargs: pytest.fail("wedge must not terminate synchronously"),
2197 )
2198
2199 for idx in range(supervisor.LOCAL_WEDGE_THRESHOLD):
2200 use_id = f"wedge-{idx}"
2201 supervisor._handle_cortex_outcome(
2202 {
2203 "tract": "cortex",
2204 "event": "start",
2205 "use_id": use_id,
2206 "provider": "local",
2207 }
2208 )
2209 supervisor._handle_cortex_outcome(
2210 {
2211 "tract": "cortex",
2212 "event": "error",
2213 "use_id": use_id,
2214 "reason_code": "provider_unavailable",
2215 }
2216 )
2217
2218 token = read_retry_token("local")
2219 assert token["reason_code"] == "local-wedge-provider-unavailable"
2220 assert token["desired_fingerprint_sha256"] == plan.desired_fingerprint_sha256
2221 assert state.generation == 5
2222 assert state.latest_phase == "retry-requested"
2223 assert state.next_truth_at == 0.0
2224 assert supervisor._recovery_state["local"].down_generation == 5
2225 assert supervisor._SERVICE_STATE[managed.name] == {
2226 "restart": False,
2227 "shutdown_timeout": 15,
2228 }
2229 managed.terminate.assert_not_called()
2230
2231
2232def _plan_for_backend(
2233 backend: str,
2234) -> supervisor.LocalServerLaunchPlan | supervisor.ParakeetServerLaunchPlan:
2235 if backend == "vulkan":
2236 return _local_plan()
2237 if backend == "cuda":
2238 return _cuda_plan()
2239 if backend == "mlx":
2240 return _mlx_plan()
2241 if backend == "parakeet-vulkan":
2242 return _parakeet_plan("vulkan")
2243 return _parakeet_plan("cpu")
2244
2245
2246def _process_name_for_backend(backend: str) -> str:
2247 if backend == "mlx":
2248 return supervisor.MLX_SERVER_PROCESS_NAME
2249 if backend.startswith("parakeet"):
2250 return supervisor.PARAKEET_SERVER_PROCESS_NAME
2251 return supervisor.LOCAL_SERVER_PROCESS_NAME
2252
2253
2254def _launch_backend_for_test(
2255 backend: str,
2256 plan: supervisor.LocalServerLaunchPlan | supervisor.ParakeetServerLaunchPlan,
2257 reservation: _FakeReservation,
2258 cancel_event: threading.Event,
2259) -> supervisor.ProviderLaunchOutcome:
2260 if backend.startswith("parakeet"):
2261 return supervisor.start_parakeet_server(
2262 plan,
2263 reservation,
2264 cancel_event,
2265 )
2266 return supervisor.start_local_server(plan, reservation, cancel_event)
2267
2268
2269@pytest.mark.parametrize(
2270 "backend",
2271 ["vulkan", "cuda", "mlx", "parakeet-cpu", "parakeet-vulkan"],
2272)
2273@pytest.mark.parametrize("cancel_point", ["before-probe", "ready", "wait"])
2274def test_start_worker_cancellation_cleans_child_at_warmup_boundaries(
2275 monkeypatch,
2276 backend: str,
2277 cancel_point: str,
2278) -> None:
2279 from solstone.think.providers import local_server, local_vulkan, parakeet_server
2280
2281 plan = _plan_for_backend(backend)
2282 cancel_event = threading.Event()
2283 probe_entered = threading.Event()
2284 managed = _FakeManaged(_process_name_for_backend(backend))
2285 service_name = managed.name
2286 ports: list[tuple[str, int]] = []
2287 placements: list[str] = []
2288
2289 monkeypatch.setattr(supervisor.time, "monotonic", lambda: 0.0)
2290 monkeypatch.setattr(
2291 supervisor,
2292 "write_service_port",
2293 lambda service, port: ports.append((service, port)),
2294 )
2295 monkeypatch.setattr(
2296 local_server,
2297 "write_local_context_window",
2298 lambda _tokens: None,
2299 )
2300 monkeypatch.setattr(
2301 local_server,
2302 "fetch_props",
2303 lambda _port: pytest.fail("cancelled launch must not fetch props"),
2304 )
2305 monkeypatch.setattr(
2306 local_vulkan,
2307 "device_local_used_mib",
2308 lambda _index: pytest.fail("cancelled launch must not inspect post-ready VRAM"),
2309 )
2310 monkeypatch.setattr(
2311 parakeet_server,
2312 "write_parakeet_placement",
2313 lambda placement: placements.append(placement),
2314 )
2315
2316 def launch_process(name, _cmd, **_kwargs):
2317 supervisor._SERVICE_STATE[name] = {
2318 "restart": True,
2319 "shutdown_timeout": 15,
2320 }
2321 if cancel_point == "before-probe":
2322 cancel_event.set()
2323 return managed
2324
2325 def local_probe(_port):
2326 probe_entered.set()
2327 if cancel_point == "ready":
2328 cancel_event.set()
2329 return local_server.STATE_READY, None
2330 return local_server.STATE_STARTING, None
2331
2332 def parakeet_probe(_port):
2333 probe_entered.set()
2334 if cancel_point == "ready":
2335 cancel_event.set()
2336 return parakeet_server.STATE_READY, None
2337 return parakeet_server.STATE_FAILED, "warming"
2338
2339 monkeypatch.setattr(supervisor, "_launch_process", launch_process)
2340 monkeypatch.setattr(local_server, "_probe_health", local_probe)
2341 monkeypatch.setattr(parakeet_server, "_probe_health", parakeet_probe)
2342
2343 outcome_box: dict[str, supervisor.ProviderLaunchOutcome] = {}
2344
2345 def run_launch() -> None:
2346 outcome_box["outcome"] = _launch_backend_for_test(
2347 backend,
2348 plan,
2349 _FakeReservation(port=45678),
2350 cancel_event,
2351 )
2352
2353 if cancel_point == "wait":
2354 thread = threading.Thread(target=run_launch)
2355 thread.start()
2356 assert probe_entered.wait(timeout=1.0)
2357 cancel_event.set()
2358 thread.join(timeout=1.0)
2359 assert not thread.is_alive()
2360 else:
2361 run_launch()
2362
2363 outcome = outcome_box["outcome"]
2364 assert outcome.status == "launch-failed"
2365 assert outcome.detail["cancelled"] is True
2366 assert outcome.managed is None
2367 managed.terminate.assert_called_once()
2368 managed.cleanup.assert_called_once_with()
2369 assert service_name not in supervisor._SERVICE_STATE
2370 assert ports == []
2371 assert placements == []
2372 if cancel_point == "before-probe":
2373 assert not probe_entered.is_set()
2374
2375
2376def test_cancelled_launch_outcome_preserves_handle_when_cleanup_raises(
2377 monkeypatch,
2378) -> None:
2379 managed = _FakeManaged()
2380 monkeypatch.setattr(
2381 supervisor,
2382 "_terminate_cleanup_handle",
2383 lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("cleanup failed")),
2384 )
2385
2386 outcome = supervisor._cancelled_launch_outcome(
2387 "local",
2388 backend="cuda",
2389 port=45678,
2390 managed=managed,
2391 reason="test cancellation",
2392 )
2393
2394 assert outcome.status == "launch-failed"
2395 assert outcome.managed is managed
2396 assert outcome.detail["cancelled"] is True
2397 assert outcome.detail["cleanup_failed"] is True
2398 assert outcome.detail["cleanup_deferred_to"] == "cleanup-failed-reconciler"
2399
2400
2401def test_cancelled_ready_result_is_cleaned_without_port_publication(monkeypatch):
2402 plan = _local_plan()
2403 state = supervisor._provider_runtime_states["local"]
2404 state.latest_phase = "starting"
2405 state.latest_plan = plan
2406 state.desired_fingerprint = plan.desired_fingerprint_sha256
2407 state.retry.attempt_count = 1
2408 state.generation = 1
2409 fence = supervisor._provider_fence(state, 1)
2410 managed = _FakeManaged()
2411 cancel_event = threading.Event()
2412 cancel_event.set()
2413 ports: list[tuple[str, int]] = []
2414 state.start_fence = fence
2415 state.start_cancel_event = cancel_event
2416 state.start_future = _future_with(
2417 supervisor.ProviderLaunchOutcome(
2418 status="ready",
2419 reason_code="probe-ready",
2420 detail={"port": 45678},
2421 managed=managed,
2422 )
2423 )
2424 monkeypatch.setattr(
2425 supervisor,
2426 "write_service_port",
2427 lambda service, port: ports.append((service, port)),
2428 )
2429
2430 assert supervisor._handle_provider_start_result(state, []) is True
2431
2432 assert ports == []
2433 managed.terminate.assert_called_once()
2434 managed.cleanup.assert_called_once_with()
2435 assert state.latest_phase == "backoff"
2436
2437
2438def test_superseded_start_cleanup_failure_is_adopted(monkeypatch) -> None:
2439 plan = _local_plan()
2440 state = supervisor._provider_runtime_states["local"]
2441 state.latest_plan = plan
2442 state.latest_phase = "ready"
2443 state.desired_fingerprint = "fp-new"
2444 state.retry.attempt_count = 2
2445 state.generation = 2
2446 old_managed = _FakeManaged()
2447 state.start_fence = supervisor.ProviderFence(
2448 incarnation=supervisor._PROVIDER_INCARNATION,
2449 generation=1,
2450 fingerprint="fp-old",
2451 attempt=1,
2452 )
2453 state.start_future = _future_with(
2454 supervisor.ProviderLaunchOutcome(
2455 status="warmup-timeout",
2456 reason_code="warmup-timeout",
2457 detail={"port": 11111},
2458 managed=old_managed,
2459 )
2460 )
2461 monkeypatch.setattr(
2462 supervisor,
2463 "_terminate_cleanup_handle",
2464 lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("cleanup failed")),
2465 )
2466
2467 assert supervisor._handle_provider_start_result(state, []) is True
2468
2469 assert state.latest_phase == "cleanup-failed"
2470 assert state.pending_stop_request is not None
2471 assert state.pending_stop_request.managed is old_managed
2472
2473
2474def test_pending_stop_request_assignment_uses_single_chokepoint() -> None:
2475 tree = ast.parse(Path(supervisor.__file__).read_text(encoding="utf-8"))
2476 offenders: list[tuple[str, int]] = []
2477 stack: list[str] = []
2478
2479 class Visitor(ast.NodeVisitor):
2480 def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
2481 stack.append(node.name)
2482 self.generic_visit(node)
2483 stack.pop()
2484
2485 def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
2486 stack.append(node.name)
2487 self.generic_visit(node)
2488 stack.pop()
2489
2490 def visit_Assign(self, node: ast.Assign) -> None:
2491 for target in node.targets:
2492 self._check_target(target, node.lineno)
2493 self.generic_visit(node)
2494
2495 def visit_AnnAssign(self, node: ast.AnnAssign) -> None:
2496 self._check_target(node.target, node.lineno)
2497 self.generic_visit(node)
2498
2499 def _check_target(self, target: ast.expr, lineno: int) -> None:
2500 if not (
2501 isinstance(target, ast.Attribute)
2502 and target.attr == "pending_stop_request"
2503 ):
2504 return
2505 owner = stack[-1] if stack else "<module>"
2506 if owner != "_set_provider_pending_stop_request":
2507 offenders.append((owner, lineno))
2508
2509 Visitor().visit(tree)
2510
2511 assert offenders == []
2512
2513
2514def test_pending_cleanup_survives_truth_phase_change_and_blocks_start(
2515 monkeypatch,
2516) -> None:
2517 plan = _local_plan()
2518 managed = _FakeManaged()
2519 state = supervisor._provider_runtime_states["local"]
2520 state.latest_phase = "cleanup-failed"
2521 state.latest_plan = None
2522 state.desired_fingerprint = plan.desired_fingerprint_sha256
2523 state.retry.desired_fingerprint = plan.desired_fingerprint_sha256
2524 state.retry.attempt_count = 1
2525 state.generation = 1
2526 state.pending_stop_request = supervisor._make_stop_request(
2527 state,
2528 managed,
2529 reason_code="cleanup-attempt-failed",
2530 detail={"source": "preserved-handle"},
2531 target_phase="stopped",
2532 target_reason_code="cleanup-succeeded",
2533 )
2534 state.cleanup_attempt_count = 1
2535 state.cleanup_next_at = 50.0
2536 state.truth_fence = supervisor._provider_fence(state, 1)
2537 state.truth_future = _future_with(
2538 supervisor.ProviderTruthObservation(
2539 provider="local",
2540 phase="starting",
2541 reason_code="launch-requested",
2542 detail={"backend": plan.backend},
2543 desired_fingerprint_json=plan.desired_fingerprint_json,
2544 desired_fingerprint_sha256=plan.desired_fingerprint_sha256,
2545 plan=plan,
2546 boot_required=True,
2547 )
2548 )
2549 start_submits: list[str] = []
2550 monkeypatch.setattr(
2551 supervisor,
2552 "_provider_start_worker",
2553 lambda *_args: start_submits.append("start"),
2554 )
2555
2556 assert supervisor._handle_provider_truth_result(state) is True
2557 assert state.latest_phase == "starting"
2558 assert state.pending_stop_request is not None
2559 assert state.pending_stop_request.managed is managed
2560
2561 monkeypatch.setattr(supervisor.time, "monotonic", lambda: 10.0)
2562 assert supervisor._submit_provider_stop_cleanup_if_needed(state, []) is True
2563 assert state.stop_cleanup_future is None
2564 supervisor._submit_provider_start_if_needed(state, [])
2565 assert state.start_future is None
2566 assert start_submits == []
2567
2568 monkeypatch.setattr(supervisor.time, "monotonic", lambda: 50.0)
2569 monkeypatch.setattr(supervisor, "_provider_executor", lambda: _InlineExecutor())
2570 assert supervisor._submit_provider_stop_cleanup_if_needed(state, []) is True
2571 assert supervisor._handle_provider_stop_cleanup_result(state, []) is True
2572
2573 managed.terminate.assert_called_once()
2574 managed.cleanup.assert_called_once_with()
2575 assert state.pending_stop_request is None
2576 assert state.latest_phase == "stopped"
2577 assert start_submits == []
2578
2579
2580def test_cancelled_stop_cleanup_preserves_unresolved_handle(monkeypatch) -> None:
2581 plan = _local_plan()
2582 managed = _FakeManaged()
2583 state = supervisor._provider_runtime_states["local"]
2584 state.latest_phase = "stopping"
2585 state.latest_plan = plan
2586 state.desired_fingerprint = plan.desired_fingerprint_sha256
2587 state.retry.desired_fingerprint = plan.desired_fingerprint_sha256
2588 state.retry.attempt_count = 1
2589 state.generation = 1
2590 state.pending_stop_request = supervisor._make_stop_request(
2591 state,
2592 managed,
2593 reason_code="cleanup-attempt-failed",
2594 detail={"source": "preserved-handle"},
2595 target_phase="stopped",
2596 target_reason_code="cleanup-succeeded",
2597 )
2598 state.stop_cleanup_fence = supervisor._provider_fence(state, 1)
2599 state.stop_cleanup_future = _future_with(
2600 supervisor.ProviderStopCleanupOutcome(
2601 status="cancelled",
2602 reason_code="stale-result-ignored",
2603 detail={"cancelled": True},
2604 managed=managed,
2605 )
2606 )
2607 monkeypatch.setattr(supervisor.time, "monotonic", lambda: 100.0)
2608
2609 assert supervisor._handle_provider_stop_cleanup_result(state, []) is True
2610
2611 assert state.latest_phase == "cleanup-failed"
2612 assert state.pending_stop_request is not None
2613 assert state.pending_stop_request.managed is managed
2614 assert state.cleanup_attempt_count == 1
2615 assert state.cleanup_next_at == 102.0
2616 supervisor._submit_provider_start_if_needed(state, [])
2617 assert state.start_future is None
2618
2619
2620def test_late_probe_cannot_declare_ready_with_cleanup_outstanding(
2621 monkeypatch,
2622) -> None:
2623 plan = _local_plan()
2624 managed = _FakeManaged()
2625 state = supervisor._provider_runtime_states["local"]
2626 state.latest_phase = "cleanup-failed"
2627 state.latest_plan = plan
2628 state.desired_fingerprint = plan.desired_fingerprint_sha256
2629 state.retry.desired_fingerprint = plan.desired_fingerprint_sha256
2630 state.retry.attempt_count = 1
2631 state.generation = 1
2632 state.pending_stop_request = supervisor._make_stop_request(
2633 state,
2634 managed,
2635 reason_code="cleanup-attempt-failed",
2636 detail={"source": "preserved-handle"},
2637 target_phase="stopped",
2638 target_reason_code="cleanup-succeeded",
2639 )
2640 state.probe_fence = supervisor._provider_fence(state, 1)
2641 state.probe_future = _future_with(
2642 supervisor.ProviderProbeOutcome(
2643 status="ready",
2644 reason_code="probe-ready",
2645 detail={"port": 45678},
2646 )
2647 )
2648 monkeypatch.setattr(supervisor.time, "monotonic", lambda: 100.0)
2649
2650 assert supervisor._handle_provider_probe_result(state) is True
2651
2652 assert state.latest_phase == "cleanup-failed"
2653 assert state.pending_stop_request is not None
2654 assert state.pending_stop_request.managed is managed
2655 assert state.next_probe_at == 100.0 + supervisor.PROVIDER_PROBE_INTERVAL_SECONDS
2656 assert read_runtime_health("local")["reason_code"] == "stale-result-ignored"
2657
2658
2659def test_deferred_stop_preserves_existing_cleanup_request() -> None:
2660 old_plan = _local_plan()
2661 managed = _FakeManaged()
2662 state = supervisor._provider_runtime_states["local"]
2663 _set_provider_ready("local", state, old_plan)
2664 state.pending_stop_request = supervisor._make_stop_request(
2665 state,
2666 managed,
2667 reason_code="cleanup-attempt-failed",
2668 detail={"source": "preserved-handle"},
2669 target_phase="stopped",
2670 target_reason_code="cleanup-succeeded",
2671 )
2672 state.cleanup_attempt_count = 2
2673 state.cleanup_next_at = 50.0
2674 observation = supervisor.ProviderTruthObservation(
2675 provider="local",
2676 phase="not-desired",
2677 reason_code="provider-not-needed",
2678 detail={"active_provider": "cloud"},
2679 desired_fingerprint_sha256=old_plan.desired_fingerprint_sha256,
2680 boot_required=False,
2681 )
2682
2683 supervisor._defer_provider_stop_for_observation(
2684 state,
2685 observation,
2686 reason_code="admission-exclusive-stop",
2687 admission_exclusive=True,
2688 )
2689
2690 assert state.latest_phase == "stop-deferred"
2691 assert state.pending_stop_request is not None
2692 assert state.pending_stop_request.managed is managed
2693 assert state.pending_stop_request.target_phase == "stop-deferred"
2694 assert state.pending_stop_request.target_detail["target_phase"] == "not-desired"
2695 assert state.cleanup_attempt_count == 2
2696 assert state.cleanup_next_at == 50.0
2697
2698
2699def test_unresolvable_cleanup_stays_owned_visible_and_blocks_start(
2700 monkeypatch,
2701) -> None:
2702 plan = _local_plan()
2703 managed = _FakeManaged()
2704 state = supervisor._provider_runtime_states["local"]
2705 state.latest_phase = "cleanup-failed"
2706 state.latest_plan = plan
2707 state.desired_fingerprint = plan.desired_fingerprint_sha256
2708 state.retry.desired_fingerprint = plan.desired_fingerprint_sha256
2709 state.retry.attempt_count = 1
2710 state.generation = 1
2711 state.pending_stop_request = supervisor._make_stop_request(
2712 state,
2713 managed,
2714 reason_code="cleanup-attempt-failed",
2715 detail={"source": "preserved-handle"},
2716 target_phase="stopped",
2717 target_reason_code="cleanup-succeeded",
2718 )
2719 state.cleanup_attempt_count = 0
2720 state.cleanup_next_at = 0.0
2721 starts: list[str] = []
2722 monkeypatch.setattr(supervisor, "_provider_executor", lambda: _InlineExecutor())
2723 monkeypatch.setattr(supervisor.time, "monotonic", lambda: 100.0)
2724 monkeypatch.setattr(
2725 supervisor,
2726 "_terminate_cleanup_handle",
2727 lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("still alive")),
2728 )
2729 monkeypatch.setattr(
2730 supervisor,
2731 "_provider_start_worker",
2732 lambda *_args: starts.append("start"),
2733 )
2734
2735 assert supervisor._submit_provider_stop_cleanup_if_needed(state, []) is True
2736 assert supervisor._handle_provider_stop_cleanup_result(state, []) is True
2737 supervisor._submit_provider_start_if_needed(state, [])
2738
2739 assert state.latest_phase == "cleanup-failed"
2740 assert state.pending_stop_request is not None
2741 assert state.pending_stop_request.managed is managed
2742 assert state.cleanup_attempt_count == 1
2743 assert state.cleanup_next_at == 102.0
2744 assert state.start_future is None
2745 assert starts == []
2746
2747
2748def test_cancelled_ready_cleanup_failure_is_adopted(monkeypatch) -> None:
2749 plan = _local_plan()
2750 state = supervisor._provider_runtime_states["local"]
2751 state.latest_phase = "starting"
2752 state.latest_plan = plan
2753 state.desired_fingerprint = plan.desired_fingerprint_sha256
2754 state.retry.attempt_count = 1
2755 state.generation = 1
2756 fence = supervisor._provider_fence(state, 1)
2757 managed = _FakeManaged()
2758 cancel_event = threading.Event()
2759 cancel_event.set()
2760 state.start_fence = fence
2761 state.start_cancel_event = cancel_event
2762 state.start_future = _future_with(
2763 supervisor.ProviderLaunchOutcome(
2764 status="ready",
2765 reason_code="probe-ready",
2766 detail={"port": 45678},
2767 managed=managed,
2768 )
2769 )
2770 monkeypatch.setattr(
2771 supervisor,
2772 "_terminate_cleanup_handle",
2773 lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("cleanup failed")),
2774 )
2775
2776 assert supervisor._handle_provider_start_result(state, []) is True
2777
2778 assert state.latest_phase == "cleanup-failed"
2779 assert state.pending_stop_request is not None
2780 assert state.pending_stop_request.managed is managed
2781
2782
2783def test_missing_ready_port_cleanup_failure_is_adopted(monkeypatch) -> None:
2784 plan = _local_plan()
2785 state = supervisor._provider_runtime_states["local"]
2786 state.latest_phase = "starting"
2787 state.latest_plan = plan
2788 state.desired_fingerprint = plan.desired_fingerprint_sha256
2789 state.retry.attempt_count = 1
2790 state.generation = 1
2791 managed = _FakeManaged()
2792 state.start_fence = supervisor._provider_fence(state, 1)
2793 state.start_future = _future_with(
2794 supervisor.ProviderLaunchOutcome(
2795 status="ready",
2796 reason_code="probe-ready",
2797 detail={},
2798 managed=managed,
2799 )
2800 )
2801 monkeypatch.setattr(
2802 supervisor,
2803 "_terminate_cleanup_handle",
2804 lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("cleanup failed")),
2805 )
2806
2807 assert supervisor._handle_provider_start_result(state, []) is True
2808
2809 assert state.latest_phase == "cleanup-failed"
2810 assert state.pending_stop_request is not None
2811 assert state.pending_stop_request.managed is managed
2812
2813
2814def test_port_publication_cleanup_failure_is_adopted(monkeypatch) -> None:
2815 plan = _local_plan()
2816 state = supervisor._provider_runtime_states["local"]
2817 state.latest_phase = "starting"
2818 state.latest_plan = plan
2819 state.desired_fingerprint = plan.desired_fingerprint_sha256
2820 state.retry.attempt_count = 1
2821 state.generation = 1
2822 managed = _FakeManaged()
2823 state.start_fence = supervisor._provider_fence(state, 1)
2824 state.start_future = _future_with(
2825 supervisor.ProviderLaunchOutcome(
2826 status="ready",
2827 reason_code="probe-ready",
2828 detail={"port": 45678},
2829 managed=managed,
2830 )
2831 )
2832 monkeypatch.setattr(
2833 supervisor,
2834 "write_service_port",
2835 lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("disk full")),
2836 )
2837 monkeypatch.setattr(
2838 supervisor,
2839 "_terminate_cleanup_handle",
2840 lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("cleanup failed")),
2841 )
2842
2843 assert supervisor._handle_provider_start_result(state, []) is True
2844
2845 assert state.latest_phase == "cleanup-failed"
2846 assert state.pending_stop_request is not None
2847 assert state.pending_stop_request.managed is managed
2848
2849
2850def test_observation_raced_when_target_fingerprint_changes_between_reads(
2851 monkeypatch,
2852) -> None:
2853 from solstone.think.providers import local_install, local_server, local_vulkan
2854
2855 fingerprints = iter(
2856 [
2857 {"provider": "local", "target": "one"},
2858 {"provider": "local", "target": "two"},
2859 ]
2860 )
2861 monkeypatch.setattr(supervisor, "_is_remote_mode", False)
2862 monkeypatch.setattr(supervisor.sys, "platform", "linux")
2863 monkeypatch.setattr(supervisor, "read_journal_config", lambda: {})
2864 monkeypatch.setattr(
2865 supervisor, "is_local_provider_needed", lambda _config=None: True
2866 )
2867 monkeypatch.setattr(
2868 "solstone.think.providers.local_endpoint.resolve_local_endpoint",
2869 lambda: type("Endpoint", (), {"is_bundled": True})(),
2870 )
2871 monkeypatch.setattr(
2872 local_install,
2873 "target_fingerprint",
2874 lambda _model_id: next(fingerprints),
2875 )
2876 monkeypatch.setattr(
2877 local_install,
2878 "inspect_readiness",
2879 lambda _model_id: _local_readiness(),
2880 )
2881 monkeypatch.setattr(local_install, "gpu_device_override", lambda: None)
2882 monkeypatch.setattr(
2883 local_vulkan,
2884 "detect_gpus",
2885 lambda: [
2886 local_vulkan.VulkanDevice(0, "GPU", local_vulkan.VK_TYPE_DISCRETE, 12288)
2887 ],
2888 )
2889 monkeypatch.setattr(
2890 local_vulkan, "select_device", lambda devices, **_kw: devices[0]
2891 )
2892 monkeypatch.setattr(local_vulkan, "device_local_used_mib", lambda _index: 0)
2893 monkeypatch.setattr(local_server, "select_server_tier", lambda _vram: _FakeTier())
2894 monkeypatch.setattr(
2895 supervisor,
2896 "_launch_process",
2897 lambda *_args, **_kwargs: pytest.fail("observation race must not launch"),
2898 )
2899
2900 observation = supervisor._observe_local_provider_truth()
2901
2902 assert observation.phase == "observing"
2903 assert observation.reason_code == "observation-raced"
2904
2905
2906def test_observation_raced_when_device_changes_during_plan_construction(
2907 monkeypatch,
2908) -> None:
2909 from solstone.think.providers import local_install, local_server, local_vulkan
2910
2911 devices = [
2912 local_vulkan.VulkanDevice(0, "GPU-A", local_vulkan.VK_TYPE_DISCRETE, 12288),
2913 local_vulkan.VulkanDevice(1, "GPU-B", local_vulkan.VK_TYPE_DISCRETE, 16384),
2914 ]
2915 selections = iter([devices[0], devices[0], devices[1]])
2916 monkeypatch.setattr(supervisor, "_is_remote_mode", False)
2917 monkeypatch.setattr(supervisor.sys, "platform", "linux")
2918 monkeypatch.setattr(supervisor, "read_journal_config", lambda: {})
2919 monkeypatch.setattr(
2920 supervisor, "is_local_provider_needed", lambda _config=None: True
2921 )
2922 monkeypatch.setattr(
2923 "solstone.think.providers.local_endpoint.resolve_local_endpoint",
2924 lambda: type("Endpoint", (), {"is_bundled": True})(),
2925 )
2926 monkeypatch.setattr(
2927 local_install,
2928 "target_fingerprint",
2929 lambda _model_id: {"provider": "local", "target": "one"},
2930 )
2931 monkeypatch.setattr(
2932 local_install,
2933 "inspect_readiness",
2934 lambda _model_id: _local_readiness(),
2935 )
2936 monkeypatch.setattr(local_install, "gpu_device_override", lambda: None)
2937 monkeypatch.setattr(local_vulkan, "detect_gpus", lambda: devices)
2938 monkeypatch.setattr(
2939 local_vulkan, "select_device", lambda _devices, **_kw: next(selections)
2940 )
2941 monkeypatch.setattr(local_vulkan, "device_local_used_mib", lambda _index: 0)
2942 monkeypatch.setattr(local_server, "select_server_tier", lambda _vram: _FakeTier())
2943
2944 observation = supervisor._observe_local_provider_truth()
2945
2946 assert observation.phase == "observing"
2947 assert observation.reason_code == "observation-raced"
2948
2949
2950def test_discarded_truth_result_reobserves_immediately_after_retry_fence_change(
2951 monkeypatch,
2952) -> None:
2953 plan = _local_plan()
2954 now = 100.0
2955 truth_submits = 0
2956 first_truth: concurrent.futures.Future = concurrent.futures.Future()
2957 state = supervisor._provider_runtime_states["local"]
2958 state.latest_phase = "ready"
2959 state.latest_plan = plan
2960 state.desired_fingerprint = plan.desired_fingerprint_sha256
2961 state.retry.desired_fingerprint = plan.desired_fingerprint_sha256
2962 state.next_truth_at = 0.0
2963
2964 class _RaceExecutor:
2965 def submit(self, fn, *args, **kwargs):
2966 nonlocal truth_submits
2967 if fn is supervisor._observe_provider_truth:
2968 truth_submits += 1
2969 if truth_submits == 1:
2970 return first_truth
2971 return _future_with(
2972 supervisor.ProviderTruthObservation(
2973 provider="local",
2974 phase="not-desired",
2975 reason_code="provider-not-needed",
2976 detail={},
2977 )
2978 )
2979 assert fn is supervisor._provider_start_worker
2980 return _future_with(
2981 supervisor.ProviderLaunchOutcome(
2982 status="launch-failed",
2983 reason_code="launch-failed",
2984 detail={},
2985 )
2986 )
2987
2988 monkeypatch.setattr(supervisor, "_provider_executor", lambda: _RaceExecutor())
2989 monkeypatch.setattr(supervisor.time, "monotonic", lambda: now)
2990
2991 asyncio.run(supervisor._reconcile_local_provider_runtime([]))
2992
2993 assert truth_submits == 1
2994 assert state.truth_future is first_truth
2995 assert state.next_truth_at == (
2996 now + supervisor.PROVIDER_TRUTH_OBSERVATION_INTERVAL_SECONDS
2997 )
2998
2999 state.latest_phase = "backoff"
3000 state.retry.next_at = now
3001 supervisor._submit_provider_start_if_needed(state, [])
3002 assert state.retry.attempt_count == 1
3003 first_truth.set_result(
3004 supervisor.ProviderTruthObservation(
3005 provider="local",
3006 phase="not-desired",
3007 reason_code="provider-not-needed",
3008 detail={"active_provider": "cloud"},
3009 )
3010 )
3011
3012 asyncio.run(supervisor._reconcile_local_provider_runtime([]))
3013
3014 assert truth_submits == 2
3015 assert state.truth_future is not first_truth
3016 assert state.next_truth_at == (
3017 now + supervisor.PROVIDER_TRUTH_OBSERVATION_INTERVAL_SECONDS
3018 )
3019
3020
3021def _set_provider_ready(
3022 provider: str,
3023 state: supervisor.ProviderRuntimeState,
3024 plan: supervisor.LocalServerLaunchPlan | supervisor.ParakeetServerLaunchPlan,
3025) -> None:
3026 state.latest_phase = "ready"
3027 state.latest_plan = plan
3028 state.desired_fingerprint = plan.desired_fingerprint_sha256
3029 state.retry.desired_fingerprint = plan.desired_fingerprint_sha256
3030 state.generation = 1
3031 state.retry.attempt_count = 1
3032 del provider
3033
3034
3035def test_admission_exclusive_stop_defers_then_stops_when_slot_frees(
3036 monkeypatch,
3037) -> None:
3038 from solstone.think.providers import local_admission
3039
3040 plan = _local_plan()
3041 managed = _FakeManaged()
3042 state = supervisor._provider_runtime_states["local"]
3043 _set_provider_ready("local", state, plan)
3044 request = supervisor._make_stop_request(
3045 state,
3046 managed,
3047 reason_code="admission-exclusive-stop",
3048 detail={},
3049 target_phase="not-desired",
3050 target_reason_code="provider-not-needed",
3051 admission_exclusive=True,
3052 )
3053 cancel_event = threading.Event()
3054 holder = _hold_local_slot_in_child(local_admission._admission_dir())
3055
3056 assert supervisor.PROVIDER_ADMISSION_STOP_TIMEOUT_S == 5.0
3057 monkeypatch.setattr(supervisor, "PROVIDER_ADMISSION_STOP_TIMEOUT_S", 0.0)
3058
3059 try:
3060 outcome = supervisor._provider_stop_cleanup_worker(
3061 "local",
3062 request,
3063 supervisor._provider_fence(state, 1),
3064 cancel_event,
3065 )
3066
3067 assert outcome.status == "stop-deferred"
3068 assert managed.terminate.call_count == 0
3069 assert list(local_admission._admission_dir().glob("wait-*.ticket")) == []
3070 finally:
3071 holder.terminate()
3072 holder.wait(timeout=2)
3073
3074 outcome = supervisor._provider_stop_cleanup_worker(
3075 "local",
3076 request,
3077 supervisor._provider_fence(state, 1),
3078 cancel_event,
3079 )
3080
3081 assert outcome.status == "stopped"
3082 managed.terminate.assert_called_once()
3083 managed.cleanup.assert_called_once_with()
3084
3085
3086def test_admission_exclusive_stop_uses_launch_captured_capacity(monkeypatch) -> None:
3087 from solstone.think.providers import local_admission
3088
3089 capacities: list[int] = []
3090 original_acquire = local_admission.acquire_local_slot
3091 plan = replace(_local_plan(), parallel_slots=3)
3092 managed = _FakeManaged()
3093 state = supervisor._provider_runtime_states["local"]
3094 _set_provider_ready("local", state, plan)
3095 state.latest_phase = "stop-deferred"
3096 state.pending_stop_target_phase = "not-desired"
3097 state.pending_stop_target_reason_code = "provider-not-needed"
3098 state.pending_stop_admission_exclusive = True
3099 state.next_truth_at = 10**12
3100 procs = [managed]
3101
3102 def acquire(capacity, timeout_s, **kwargs):
3103 capacities.append(capacity)
3104 return original_acquire(capacity, timeout_s, **kwargs)
3105
3106 monkeypatch.setattr(local_admission, "acquire_local_slot", acquire)
3107 monkeypatch.setattr(supervisor, "_provider_executor", lambda: _InlineExecutor())
3108
3109 asyncio.run(supervisor._reconcile_local_provider_runtime(procs))
3110 asyncio.run(supervisor._reconcile_local_provider_runtime(procs))
3111
3112 assert capacities == [3]
3113 assert state.latest_phase == "not-desired"
3114
3115
3116def test_admission_exclusive_stop_rechecks_reactivation_after_acquisition(
3117 monkeypatch,
3118) -> None:
3119 from solstone.think.providers import local_admission
3120
3121 original_acquire = local_admission.acquire_local_slot
3122 managed = _FakeManaged()
3123 state = supervisor._provider_runtime_states["local"]
3124 plan = _local_plan()
3125 _set_provider_ready("local", state, plan)
3126 request = supervisor._make_stop_request(
3127 state,
3128 managed,
3129 reason_code="admission-exclusive-stop",
3130 detail={},
3131 target_phase="not-desired",
3132 target_reason_code="provider-not-needed",
3133 admission_exclusive=True,
3134 )
3135 cancel_event = threading.Event()
3136
3137 def acquire(*args, **kwargs):
3138 permit = original_acquire(*args, **kwargs)
3139 cancel_event.set()
3140 return permit
3141
3142 monkeypatch.setattr(local_admission, "acquire_local_slot", acquire)
3143
3144 outcome = supervisor._provider_stop_cleanup_worker(
3145 "local",
3146 request,
3147 supervisor._provider_fence(state, 1),
3148 cancel_event,
3149 )
3150
3151 assert outcome.status == "cancelled"
3152 assert outcome.managed is managed
3153 managed.terminate.assert_not_called()
3154 managed.cleanup.assert_not_called()
3155
3156
3157@pytest.mark.parametrize(
3158 ("provider", "old_managed", "new_observation"),
3159 [
3160 (
3161 "local",
3162 _FakeManaged(supervisor.MLX_SERVER_PROCESS_NAME),
3163 supervisor.ProviderTruthObservation(
3164 provider="local",
3165 phase="starting",
3166 reason_code="launch-requested",
3167 detail={},
3168 desired_fingerprint_json='{"provider":"local","backend":"cuda"}',
3169 desired_fingerprint_sha256="fp-local-cuda",
3170 plan=_cuda_plan(),
3171 boot_required=True,
3172 ),
3173 ),
3174 (
3175 "parakeet",
3176 _FakeManaged(supervisor.PARAKEET_SERVER_PROCESS_NAME),
3177 supervisor.ProviderTruthObservation(
3178 provider="parakeet",
3179 phase="starting",
3180 reason_code="launch-requested",
3181 detail={},
3182 desired_fingerprint_json='{"provider":"parakeet","target":"new"}',
3183 desired_fingerprint_sha256="fp-parakeet-new",
3184 plan=replace(
3185 _parakeet_plan("cpu"),
3186 desired_fingerprint_sha256="fp-parakeet-new",
3187 ),
3188 boot_required=True,
3189 ),
3190 ),
3191 ],
3192)
3193def test_stop_before_replace_runs_before_replacement_start(
3194 monkeypatch,
3195 provider: str,
3196 old_managed: _FakeManaged,
3197 new_observation: supervisor.ProviderTruthObservation,
3198) -> None:
3199 state = supervisor._provider_runtime_states[provider]
3200 old_plan = _local_plan() if provider == "local" else _parakeet_plan("cpu")
3201 _set_provider_ready(provider, state, old_plan)
3202 state.next_truth_at = 10**12
3203 state.truth_fence = supervisor._provider_fence(state, 1)
3204 state.truth_future = _future_with(new_observation)
3205 order: list[str] = []
3206 procs = [old_managed]
3207
3208 def cleanup(managed, *, reason, state_name=None):
3209 order.append(f"cleanup:{managed.name}:{reason}")
3210 managed.terminate()
3211 managed.cleanup()
3212 managed.is_running = lambda: False
3213
3214 def start_worker(provider_arg, _plan_arg, _fence, _cancel_event):
3215 order.append(f"start:{provider_arg}")
3216 return supervisor.ProviderLaunchOutcome(
3217 status="launch-failed",
3218 reason_code="launch-failed",
3219 detail={},
3220 )
3221
3222 monkeypatch.setattr(supervisor, "_provider_executor", lambda: _InlineExecutor())
3223 monkeypatch.setattr(supervisor, "_terminate_cleanup_handle", cleanup)
3224 monkeypatch.setattr(supervisor, "_provider_start_worker", start_worker)
3225
3226 asyncio.run(supervisor._reconcile_provider_runtime(provider, procs))
3227 assert order == ["cleanup:" + old_managed.name + ":target-changed"]
3228
3229 for _ in range(3):
3230 if order[-1] == f"start:{provider}":
3231 break
3232 asyncio.run(supervisor._reconcile_provider_runtime(provider, procs))
3233 assert order[-1] == f"start:{provider}"
3234 assert old_managed.terminate.call_count == 1
3235
3236
3237def test_local_artifact_failure_before_replacement_keeps_old_child_and_retries(
3238 monkeypatch,
3239 tmp_path: Path,
3240) -> None:
3241 from solstone.think.providers import local_server
3242 from solstone.think.providers.install_state import (
3243 begin_or_replace_install_attempt,
3244 read_install_status,
3245 transition_state,
3246 write_install_status,
3247 )
3248
3249 state = supervisor._provider_runtime_states["local"]
3250 old_managed = _FakeManaged(supervisor.LOCAL_SERVER_PROCESS_NAME)
3251 prior_tree = tmp_path / "prior-runtime"
3252 prior_tree.mkdir()
3253 prior_binary = prior_tree / "llama-server"
3254 prior_binary.write_text("old runtime", encoding="utf-8")
3255 old_plan = replace(
3256 _local_plan(),
3257 binary_path=prior_binary,
3258 desired_fingerprint_json='{"provider":"local","backend":"vulkan"}',
3259 desired_fingerprint_sha256="fp-local-vulkan-old",
3260 )
3261 _set_provider_ready("local", state, old_plan)
3262 write_runtime_health(
3263 _runtime_record(
3264 "local",
3265 phase="ready",
3266 fingerprint=old_plan.desired_fingerprint_sha256,
3267 generation=1,
3268 attempt=1,
3269 process={
3270 "name": old_managed.name,
3271 "pid": old_managed.process.pid,
3272 "ref": old_managed.ref,
3273 "port": 45678,
3274 },
3275 )
3276 )
3277 failed_target = {"provider": "local", "backend": "cuda"}
3278 attempt = begin_or_replace_install_attempt(
3279 "local",
3280 failed_target,
3281 initial_state="downloading",
3282 )
3283 write_install_status(
3284 transition_state(
3285 attempt,
3286 new_state="failed",
3287 error="cuda_runtime_incomplete",
3288 error_code="cuda_runtime_incomplete",
3289 )
3290 )
3291 failed_observation = supervisor.ProviderTruthObservation(
3292 provider="local",
3293 phase="artifact-not-ready",
3294 reason_code="artifact-missing",
3295 detail={
3296 "readiness_status": "missing-or-mismatched",
3297 "readiness_reason_code": "cuda_runtime_incomplete",
3298 "install_state": "failed",
3299 "install_acquisition_allowed": False,
3300 },
3301 desired_fingerprint_json='{"provider":"local","backend":"cuda"}',
3302 desired_fingerprint_sha256="fp-local-cuda",
3303 plan=None,
3304 boot_required=True,
3305 )
3306 state.next_truth_at = 10**12
3307 state.truth_fence = supervisor._provider_fence(state, 1)
3308 state.truth_future = _future_with(failed_observation)
3309 state.probe_fence = supervisor._provider_fence(state, 1)
3310 state.probe_future = _future_with(
3311 supervisor._probe_outcome("ready", "probe-ready", {"port": 45678})
3312 )
3313 monkeypatch.setattr(supervisor, "_provider_executor", lambda: _InlineExecutor())
3314
3315 asyncio.run(supervisor._reconcile_provider_runtime("local", [old_managed]))
3316
3317 old_managed.terminate.assert_not_called()
3318 old_managed.cleanup.assert_not_called()
3319 assert prior_binary.read_text(encoding="utf-8") == "old runtime"
3320 status = read_install_status(name="local")
3321 assert status["install_state"] == "failed"
3322 assert status["error_code"] == "cuda_runtime_incomplete"
3323 assert state.latest_phase in {"ready", "ready-proof-unavailable"}
3324 assert state.desired_fingerprint == old_plan.desired_fingerprint_sha256
3325 record = read_runtime_health("local")
3326 assert record["phase"] == "artifact-not-ready"
3327 assert record["reason_code"] == "artifact-missing"
3328 assert record["process"]["ref"] == old_managed.ref
3329 assert state.probe_future is None
3330
3331 probe_calls: list[int] = []
3332
3333 def failed_probe(port: int) -> tuple[str, str]:
3334 probe_calls.append(port)
3335 return local_server.STATE_FAILED, "timed out"
3336
3337 monkeypatch.setattr(local_server, "_probe_health", failed_probe)
3338 state.next_probe_at = 0.0
3339 supervisor.write_service_port("local", 45678)
3340
3341 asyncio.run(supervisor._reconcile_provider_runtime("local", [old_managed]))
3342 assert state.probe_future is not None
3343 asyncio.run(supervisor._reconcile_provider_runtime("local", [old_managed]))
3344
3345 assert probe_calls == [45678]
3346 record = read_runtime_health("local")
3347 assert record["phase"] == "ready-proof-unavailable"
3348 assert record["reason_code"] == "proof-observation-unavailable"
3349 assert record["detail"]["error"] == "timed out"
3350 assert record["process"]["ref"] == old_managed.ref
3351
3352 retry_observation = supervisor.ProviderTruthObservation(
3353 provider="local",
3354 phase="starting",
3355 reason_code="launch-requested",
3356 detail={},
3357 desired_fingerprint_json='{"provider":"local","backend":"cuda"}',
3358 desired_fingerprint_sha256="fp-local-cuda",
3359 plan=_cuda_plan(),
3360 boot_required=True,
3361 )
3362 state.truth_fence = supervisor._provider_fence(state, 1)
3363 state.truth_future = _future_with(retry_observation)
3364 order: list[str] = []
3365
3366 def cleanup(managed, *, reason, state_name=None):
3367 del state_name
3368 order.append(f"cleanup:{managed.name}:{reason}")
3369 managed.terminate()
3370 managed.cleanup()
3371 managed.is_running = lambda: False
3372
3373 def start_worker(provider_arg, _plan_arg, _fence, _cancel_event):
3374 order.append(f"start:{provider_arg}")
3375 return supervisor.ProviderLaunchOutcome(
3376 status="launch-failed",
3377 reason_code="launch-failed",
3378 detail={},
3379 )
3380
3381 monkeypatch.setattr(supervisor, "_terminate_cleanup_handle", cleanup)
3382 monkeypatch.setattr(supervisor, "_provider_start_worker", start_worker)
3383
3384 asyncio.run(supervisor._reconcile_provider_runtime("local", [old_managed]))
3385 assert order == ["cleanup:" + old_managed.name + ":target-changed"]
3386
3387 for _ in range(3):
3388 if order[-1] == "start:local":
3389 break
3390 asyncio.run(supervisor._reconcile_provider_runtime("local", [old_managed]))
3391 assert order[-1] == "start:local"
3392
3393
3394def test_matching_target_duplicate_convergence_keeps_owner_and_stops_stale(
3395 monkeypatch,
3396) -> None:
3397 keep = _FakeManaged(supervisor.LOCAL_SERVER_PROCESS_NAME)
3398 stale = _FakeManaged(supervisor.MLX_SERVER_PROCESS_NAME)
3399 state = supervisor._provider_runtime_states["local"]
3400 plan = _local_plan()
3401 _set_provider_ready("local", state, plan)
3402 state.next_truth_at = 10**12
3403 procs = [keep, stale]
3404 write_runtime_health(
3405 _runtime_record(
3406 "local",
3407 phase="ready",
3408 fingerprint=plan.desired_fingerprint_sha256,
3409 generation=1,
3410 attempt=1,
3411 process={
3412 "name": keep.name,
3413 "pid": keep.process.pid,
3414 "ref": keep.ref,
3415 "port": 45678,
3416 },
3417 )
3418 )
3419 starts: list[int] = []
3420 stopped: list[_FakeManaged] = []
3421 monkeypatch.setattr(supervisor, "_provider_executor", lambda: _InlineExecutor())
3422 monkeypatch.setattr(
3423 supervisor,
3424 "_provider_start_worker",
3425 lambda *_args: starts.append(1),
3426 )
3427 monkeypatch.setattr(
3428 supervisor,
3429 "_terminate_cleanup_handle",
3430 lambda managed, *, reason, state_name=None: (
3431 stopped.append(managed),
3432 managed.terminate(),
3433 managed.cleanup(),
3434 setattr(managed, "is_running", lambda: False),
3435 ),
3436 )
3437
3438 asyncio.run(supervisor._reconcile_local_provider_runtime(procs))
3439 asyncio.run(supervisor._reconcile_local_provider_runtime(procs))
3440
3441 assert stopped == [stale]
3442 assert keep not in stopped
3443 assert starts == []
3444
3445
3446def test_late_cleanup_cannot_clear_newer_generation_port_file() -> None:
3447 plan = _local_plan()
3448 state = supervisor._provider_runtime_states["local"]
3449 state.latest_phase = "stopping"
3450 state.latest_plan = plan
3451 state.desired_fingerprint = plan.desired_fingerprint_sha256
3452 state.generation = 2
3453 state.retry.attempt_count = 2
3454 newer = _FakeManaged()
3455 supervisor.write_service_port("local", 22222)
3456 write_runtime_health(
3457 _runtime_record(
3458 "local",
3459 phase="ready",
3460 fingerprint=plan.desired_fingerprint_sha256,
3461 generation=2,
3462 attempt=2,
3463 process={
3464 "name": newer.name,
3465 "pid": newer.process.pid,
3466 "ref": newer.ref,
3467 "port": 22222,
3468 },
3469 )
3470 )
3471 old = _FakeManaged()
3472 old_fence = supervisor.ProviderFence(
3473 incarnation=supervisor._PROVIDER_INCARNATION,
3474 generation=1,
3475 fingerprint=plan.desired_fingerprint_sha256,
3476 attempt=1,
3477 )
3478 state.pending_stop_request = supervisor._make_stop_request(
3479 state,
3480 old,
3481 reason_code="target-changed",
3482 detail={},
3483 )
3484 state.stop_cleanup_fence = old_fence
3485 state.stop_cleanup_future = _future_with(
3486 supervisor.ProviderStopCleanupOutcome(
3487 status="stopped",
3488 reason_code="cleanup-succeeded",
3489 detail={"port": 11111},
3490 )
3491 )
3492
3493 assert supervisor._handle_provider_stop_cleanup_result(state, [newer]) is True
3494
3495 assert supervisor.read_service_port("local") == 22222
3496
3497
3498def test_fenced_out_stop_cleanup_failure_preserves_handle() -> None:
3499 plan = _local_plan()
3500 state = supervisor._provider_runtime_states["local"]
3501 state.latest_phase = "ready"
3502 state.latest_plan = plan
3503 state.desired_fingerprint = plan.desired_fingerprint_sha256
3504 state.generation = 2
3505 state.retry.attempt_count = 2
3506 stale_managed = _FakeManaged()
3507 old_fence = supervisor.ProviderFence(
3508 incarnation=supervisor._PROVIDER_INCARNATION,
3509 generation=1,
3510 fingerprint=plan.desired_fingerprint_sha256,
3511 attempt=1,
3512 )
3513 state.pending_stop_request = supervisor._make_stop_request(
3514 state,
3515 stale_managed,
3516 reason_code="target-changed",
3517 detail={},
3518 )
3519 state.stop_cleanup_fence = old_fence
3520 state.stop_cleanup_future = _future_with(
3521 supervisor.ProviderStopCleanupOutcome(
3522 status="cleanup-failed",
3523 reason_code="cleanup-attempt-failed",
3524 detail={"error": "terminate failed"},
3525 managed=stale_managed,
3526 )
3527 )
3528
3529 assert supervisor._handle_provider_stop_cleanup_result(state, []) is True
3530
3531 assert state.latest_phase == "cleanup-failed"
3532 assert state.pending_stop_request is not None
3533 assert state.pending_stop_request.managed is stale_managed
3534
3535
3536def test_cleanup_failed_cadence_consumes_no_launch_budget(monkeypatch) -> None:
3537 now = 100.0
3538 plan = _local_plan()
3539 managed = _FakeManaged()
3540 state = supervisor._provider_runtime_states["local"]
3541 _set_provider_ready("local", state, plan)
3542 state.retry.attempt_count = 4
3543 state.pending_stop_request = supervisor._make_stop_request(
3544 state,
3545 managed,
3546 reason_code="target-changed",
3547 detail={},
3548 )
3549 delays: list[float] = []
3550
3551 def monotonic() -> float:
3552 return now
3553
3554 monkeypatch.setattr(supervisor.time, "monotonic", monotonic)
3555
3556 for _ in range(5):
3557 request = state.pending_stop_request
3558 assert request is not None
3559 supervisor._schedule_cleanup_failed_retry(
3560 state,
3561 request,
3562 supervisor.ProviderStopCleanupOutcome(
3563 status="cleanup-failed",
3564 reason_code="cleanup-attempt-failed",
3565 detail={},
3566 managed=managed,
3567 ),
3568 )
3569 delays.append(state.cleanup_next_at - now)
3570
3571 assert delays == [2.0, 4.0, 8.0, 16.0, 30.0]
3572 assert state.retry.attempt_count == 4
3573
3574
3575def test_cleanup_failed_rechecks_dead_child_and_recovers(monkeypatch) -> None:
3576 plan = _local_plan()
3577 dead = _DeadManaged()
3578 state = supervisor._provider_runtime_states["local"]
3579 _set_provider_ready("local", state, plan)
3580 state.latest_phase = "cleanup-failed"
3581 state.latest_plan = None
3582 state.next_truth_at = 10**12
3583 state.pending_stop_request = supervisor._make_stop_request(
3584 state,
3585 dead,
3586 reason_code="target-changed",
3587 detail={},
3588 )
3589 state.cleanup_next_at = 0.0
3590 monkeypatch.setattr(supervisor, "_provider_executor", lambda: _InlineExecutor())
3591 monkeypatch.setattr(supervisor.time, "monotonic", lambda: 10.0)
3592
3593 procs = [dead]
3594 asyncio.run(supervisor._reconcile_local_provider_runtime(procs))
3595 asyncio.run(supervisor._reconcile_local_provider_runtime(procs))
3596
3597 assert state.latest_phase == "stopped"
3598 dead.cleanup.assert_called_once_with()
3599
3600
3601def test_pending_cleanup_dead_child_converges_after_truth_changed_phase(
3602 monkeypatch,
3603) -> None:
3604 plan = _local_plan()
3605 dead = _DeadManaged()
3606 state = supervisor._provider_runtime_states["local"]
3607 state.latest_phase = "starting"
3608 state.latest_plan = plan
3609 state.desired_fingerprint = plan.desired_fingerprint_sha256
3610 state.retry.desired_fingerprint = plan.desired_fingerprint_sha256
3611 state.retry.attempt_count = 1
3612 state.generation = 1
3613 state.pending_stop_request = supervisor._make_stop_request(
3614 state,
3615 dead,
3616 reason_code="cleanup-attempt-failed",
3617 detail={"source": "preserved-handle"},
3618 target_phase="stopped",
3619 target_reason_code="cleanup-succeeded",
3620 )
3621 state.cleanup_attempt_count = 1
3622 state.cleanup_next_at = 0.0
3623 monkeypatch.setattr(supervisor, "_provider_executor", lambda: _InlineExecutor())
3624 monkeypatch.setattr(supervisor.time, "monotonic", lambda: 10.0)
3625
3626 assert supervisor._submit_provider_stop_cleanup_if_needed(state, []) is True
3627 assert supervisor._handle_provider_stop_cleanup_result(state, [dead]) is True
3628
3629 assert state.pending_stop_request is None
3630 assert state.latest_phase == "stopped"
3631 dead.cleanup.assert_called_once_with()
3632
3633
3634def test_preserved_cancel_cleanup_handle_is_adopted_not_orphaned() -> None:
3635 plan = _local_plan()
3636 managed = _FakeManaged()
3637 state = supervisor._provider_runtime_states["local"]
3638 state.latest_phase = "starting"
3639 state.latest_plan = plan
3640 state.desired_fingerprint = plan.desired_fingerprint_sha256
3641 state.generation = 1
3642 state.retry.attempt_count = 1
3643 state.start_fence = supervisor._provider_fence(state, 1)
3644 state.start_future = _future_with(
3645 supervisor.ProviderLaunchOutcome(
3646 status="launch-failed",
3647 reason_code="launch-failed",
3648 detail={
3649 "cleanup_failed": True,
3650 "cleanup_deferred_to": "cleanup-failed-reconciler",
3651 },
3652 managed=managed,
3653 )
3654 )
3655
3656 assert supervisor._handle_provider_start_result(state, []) is True
3657
3658 assert state.latest_phase == "cleanup-failed"
3659 assert state.pending_stop_request is not None
3660 assert state.pending_stop_request.managed is managed
3661 managed.terminate.assert_not_called()
3662 managed.cleanup.assert_not_called()
3663
3664
3665def test_shutdown_stop_cleanup_signal_does_not_run_cleanup(monkeypatch) -> None:
3666 state = supervisor._provider_runtime_states["local"]
3667 event = threading.Event()
3668 state.stop_cleanup_cancel_event = event
3669 state.stop_cleanup_fence = supervisor.ProviderFence(
3670 incarnation=supervisor._PROVIDER_INCARNATION,
3671 generation=1,
3672 fingerprint="fp-local",
3673 attempt=1,
3674 )
3675 monkeypatch.setattr(
3676 supervisor,
3677 "_terminate_cleanup_handle",
3678 lambda *_args, **_kwargs: pytest.fail(
3679 "shutdown signal must not cleanup inline"
3680 ),
3681 )
3682
3683 supervisor._cancel_all_provider_stops("test shutdown")
3684
3685 assert event.is_set()
3686
3687
3688@pytest.mark.parametrize(
3689 ("initial_phase", "observation", "expected_phase", "expected_fingerprint"),
3690 [
3691 (
3692 "ready",
3693 supervisor.ProviderTruthObservation(
3694 provider="local",
3695 phase="not-desired",
3696 reason_code="provider-not-needed",
3697 detail={"active_provider": "cloud"},
3698 ),
3699 "not-desired",
3700 None,
3701 ),
3702 (
3703 "backoff",
3704 supervisor.ProviderTruthObservation(
3705 provider="local",
3706 phase="not-desired",
3707 reason_code="provider-not-needed",
3708 detail={"active_provider": "cloud"},
3709 ),
3710 "not-desired",
3711 None,
3712 ),
3713 (
3714 "ready",
3715 supervisor.ProviderTruthObservation(
3716 provider="local",
3717 phase="starting",
3718 reason_code="launch-requested",
3719 detail={"target": "new"},
3720 desired_fingerprint_json='{"provider":"local","target":"new"}',
3721 desired_fingerprint_sha256="fp-new",
3722 plan=_local_plan(),
3723 boot_required=True,
3724 ),
3725 "starting",
3726 "fp-new",
3727 ),
3728 (
3729 "host-blocked",
3730 supervisor.ProviderTruthObservation(
3731 provider="local",
3732 phase="starting",
3733 reason_code="launch-requested",
3734 detail={"target": "new"},
3735 desired_fingerprint_json='{"provider":"local","target":"new"}',
3736 desired_fingerprint_sha256="fp-new",
3737 plan=_local_plan(),
3738 boot_required=True,
3739 ),
3740 "starting",
3741 "fp-new",
3742 ),
3743 (
3744 "failed",
3745 supervisor.ProviderTruthObservation(
3746 provider="local",
3747 phase="starting",
3748 reason_code="launch-requested",
3749 detail={"target": "new"},
3750 desired_fingerprint_json='{"provider":"local","target":"new"}',
3751 desired_fingerprint_sha256="fp-new",
3752 plan=_local_plan(),
3753 boot_required=True,
3754 ),
3755 "starting",
3756 "fp-new",
3757 ),
3758 (
3759 "artifact-not-ready",
3760 supervisor.ProviderTruthObservation(
3761 provider="local",
3762 phase="starting",
3763 reason_code="launch-requested",
3764 detail={"target": "new"},
3765 desired_fingerprint_json='{"provider":"local","target":"new"}',
3766 desired_fingerprint_sha256="fp-new",
3767 plan=_local_plan(),
3768 boot_required=True,
3769 ),
3770 "starting",
3771 "fp-new",
3772 ),
3773 ],
3774)
3775def test_continuous_reobservation_converges_from_steady_phases(
3776 monkeypatch,
3777 initial_phase: RuntimePhase,
3778 observation: supervisor.ProviderTruthObservation,
3779 expected_phase: RuntimePhase,
3780 expected_fingerprint: str | None,
3781) -> None:
3782 state = supervisor._provider_runtime_states["local"]
3783 old_plan = _local_plan()
3784 state.latest_phase = initial_phase
3785 state.latest_plan = old_plan
3786 state.desired_fingerprint = old_plan.desired_fingerprint_sha256
3787 state.retry = supervisor.ProviderRetryState(
3788 attempt_count=1,
3789 next_at=9999.0,
3790 desired_fingerprint=old_plan.desired_fingerprint_sha256,
3791 )
3792 state.next_truth_at = 0.0
3793 starts: list[int] = []
3794
3795 monkeypatch.setattr(supervisor.time, "monotonic", lambda: 100.0)
3796 monkeypatch.setattr(supervisor, "_provider_executor", lambda: _InlineExecutor())
3797 monkeypatch.setattr(
3798 supervisor,
3799 "_observe_local_provider_truth",
3800 lambda: observation,
3801 )
3802 monkeypatch.setattr(
3803 supervisor,
3804 "_provider_start_worker",
3805 lambda provider, plan_arg, fence, _cancel_event: (
3806 starts.append(fence.attempt)
3807 or supervisor.ProviderLaunchOutcome(
3808 status="launch-failed",
3809 reason_code="launch-failed",
3810 detail={},
3811 )
3812 ),
3813 )
3814
3815 asyncio.run(supervisor._reconcile_local_provider_runtime([]))
3816 asyncio.run(supervisor._reconcile_local_provider_runtime([]))
3817
3818 assert state.latest_phase == expected_phase
3819 assert state.desired_fingerprint == expected_fingerprint
3820 assert state.truth_future is None
3821
3822
3823def test_retry_cadence_exhausts_after_six_attempts(monkeypatch):
3824 now = 1000.0
3825 launches: list[int] = []
3826 plan = _local_plan()
3827
3828 def monotonic() -> float:
3829 return now
3830
3831 def observe():
3832 return supervisor.ProviderTruthObservation(
3833 provider="local",
3834 phase="starting",
3835 reason_code="launch-requested",
3836 detail={},
3837 desired_fingerprint_json=plan.desired_fingerprint_json,
3838 desired_fingerprint_sha256=plan.desired_fingerprint_sha256,
3839 plan=plan,
3840 boot_required=True,
3841 )
3842
3843 def start_worker(provider, plan_arg, fence, _cancel_event):
3844 assert provider == "local"
3845 assert plan_arg is plan
3846 launches.append(fence.attempt)
3847 return supervisor.ProviderLaunchOutcome(
3848 status="launch-failed",
3849 reason_code="launch-failed",
3850 detail={"attempt": fence.attempt},
3851 )
3852
3853 monkeypatch.setattr(supervisor, "_provider_executor", lambda: _InlineExecutor())
3854 monkeypatch.setattr(supervisor.time, "monotonic", monotonic)
3855 monkeypatch.setattr(supervisor, "_observe_local_provider_truth", observe)
3856 monkeypatch.setattr(supervisor, "_provider_start_worker", start_worker)
3857
3858 state = supervisor._provider_runtime_states["local"]
3859 delays: list[float] = []
3860
3861 for _ in range(40):
3862 asyncio.run(supervisor._reconcile_local_provider_runtime([]))
3863 if state.latest_phase == "backoff":
3864 delays.append(state.retry.next_at - now)
3865 now = state.retry.next_at
3866 if state.latest_phase == "failed":
3867 break
3868
3869 assert launches == [1, 2, 3, 4, 5, 6]
3870 assert delays == [2.0, 4.0, 8.0, 16.0, 30.0]
3871 assert state.latest_phase == "failed"
3872
3873
3874def test_startup_gate_releases_on_window_and_reconciler_keeps_retrying(monkeypatch):
3875 now = supervisor.PROVIDER_STARTUP_GATE_WINDOW_SECONDS + 1.0
3876 queue = _FakeTaskQueue()
3877 plan = _local_plan()
3878 launches: list[int] = []
3879
3880 state = supervisor._provider_runtime_states["local"]
3881 state.latest_phase = "backoff"
3882 state.latest_plan = plan
3883 state.desired_fingerprint = plan.desired_fingerprint_sha256
3884 state.retry.attempt_count = 1
3885 state.retry.next_at = now
3886 monkeypatch.setattr(supervisor, "_task_queue", queue)
3887 monkeypatch.setattr(
3888 supervisor,
3889 "_provider_startup_gate",
3890 supervisor.ProviderStartupGate(started_at=0.0, required={"local"}),
3891 )
3892 monkeypatch.setattr(supervisor.time, "monotonic", lambda: now)
3893 monkeypatch.setattr(supervisor, "_provider_executor", lambda: _InlineExecutor())
3894 monkeypatch.setattr(
3895 supervisor,
3896 "_provider_start_worker",
3897 lambda provider, plan_arg, fence, _cancel_event: (
3898 launches.append(fence.attempt)
3899 or supervisor.ProviderLaunchOutcome(
3900 status="launch-failed",
3901 reason_code="launch-failed",
3902 detail={},
3903 )
3904 ),
3905 )
3906
3907 supervisor._release_provider_startup_gate_if_ready()
3908
3909 assert queue.ready_calls == 1
3910 assert supervisor._provider_startup_gate.released is True
3911
3912 asyncio.run(supervisor._reconcile_local_provider_runtime([]))
3913 asyncio.run(supervisor._reconcile_local_provider_runtime([]))
3914
3915 assert launches == [2]
3916 assert queue.ready_calls == 1
3917
3918
3919@pytest.mark.parametrize("terminal_phase", ["ready", "host-blocked"])
3920def test_startup_gate_releases_on_terminal_provider_state(
3921 monkeypatch,
3922 terminal_phase: RuntimePhase,
3923) -> None:
3924 queue = _FakeTaskQueue()
3925 state = supervisor._provider_runtime_states["local"]
3926 monkeypatch.setattr(supervisor, "_task_queue", queue)
3927 monkeypatch.setattr(
3928 supervisor,
3929 "_provider_startup_gate",
3930 supervisor.ProviderStartupGate(started_at=0.0, required={"local"}),
3931 )
3932 monkeypatch.setattr(supervisor.time, "monotonic", lambda: 1.0)
3933
3934 supervisor._finish_provider_startup_condition(state, terminal_phase)
3935 supervisor._release_provider_startup_gate_if_ready()
3936 supervisor._release_provider_startup_gate_if_ready()
3937
3938 assert queue.ready_calls == 1
3939 assert supervisor._provider_startup_gate.released is True
3940
3941
3942def test_fenced_ready_result_publishes_port(monkeypatch):
3943 from solstone.think.providers import local_server
3944
3945 plan = _local_plan()
3946 state = supervisor._provider_runtime_states["local"]
3947 state.latest_plan = plan
3948 state.latest_phase = "starting"
3949 state.desired_fingerprint = plan.desired_fingerprint_sha256
3950 state.retry.attempt_count = 1
3951 state.generation = 3
3952 fence = supervisor._provider_fence(state, 1)
3953 managed = _FakeManaged()
3954 ports: list[tuple[str, int]] = []
3955 order: list[str] = []
3956 state.start_fence = fence
3957 state.start_future = _future_with(
3958 supervisor.ProviderLaunchOutcome(
3959 status="ready",
3960 reason_code="probe-ready",
3961 detail={"port": 45678},
3962 managed=managed,
3963 )
3964 )
3965 monkeypatch.setattr(
3966 supervisor,
3967 "write_service_port",
3968 lambda service, port: order.append("port") or ports.append((service, port)),
3969 )
3970 monkeypatch.setattr(
3971 local_server,
3972 "write_local_context_window",
3973 lambda _tokens: order.append("context"),
3974 )
3975 original_write = supervisor._write_provider_runtime
3976
3977 def write_with_order(*args, **kwargs):
3978 order.append(f"runtime:{kwargs['phase']}")
3979 return original_write(*args, **kwargs)
3980
3981 monkeypatch.setattr(supervisor, "_write_provider_runtime", write_with_order)
3982
3983 assert supervisor._handle_provider_start_result(state, []) is True
3984
3985 assert ports == [("local", 45678)]
3986 assert order[:3] == ["runtime:ready", "context", "port"]
3987 record = read_runtime_health("local")
3988 assert record["phase"] == "ready"
3989 assert record["generation"] == 3
3990 assert record["attempt"] == 1
3991 assert record["process"]["port"] == 45678
3992
3993
3994def test_ready_ownership_write_failure_does_not_publish_port_or_ready(
3995 monkeypatch,
3996) -> None:
3997 plan = _local_plan()
3998 state = supervisor._provider_runtime_states["local"]
3999 state.latest_plan = plan
4000 state.latest_phase = "starting"
4001 state.desired_fingerprint = plan.desired_fingerprint_sha256
4002 state.retry.attempt_count = 1
4003 state.generation = 1
4004 managed = _FakeManaged()
4005 state.start_fence = supervisor._provider_fence(state, 1)
4006 state.start_future = _future_with(
4007 supervisor.ProviderLaunchOutcome(
4008 status="ready",
4009 reason_code="probe-ready",
4010 detail={"port": 45678},
4011 managed=managed,
4012 )
4013 )
4014 writes: list[RuntimePhase] = []
4015 monkeypatch.setattr(
4016 supervisor,
4017 "write_service_port",
4018 lambda *_args, **_kwargs: pytest.fail("port must not be published"),
4019 )
4020
4021 def failed_ready_write(*args, **kwargs):
4022 phase = kwargs["phase"]
4023 writes.append(phase)
4024 if phase == "ready":
4025 return None
4026 return {
4027 **read_runtime_health("local"),
4028 "phase": phase,
4029 "reason_code": kwargs["reason_code"],
4030 "detail": kwargs["detail"],
4031 "desired_fingerprint_sha256": state.desired_fingerprint,
4032 "incarnation": supervisor._PROVIDER_INCARNATION,
4033 "generation": state.generation,
4034 "attempt": state.retry.attempt_count,
4035 "process": kwargs.get("process"),
4036 "updated_at": "2026-07-19T00:00:00+00:00",
4037 "owner": {"test": "failed-ready-write"},
4038 }
4039
4040 monkeypatch.setattr(supervisor, "_write_provider_runtime", failed_ready_write)
4041
4042 assert supervisor._handle_provider_start_result(state, []) is True
4043
4044 assert writes == ["ready", "state-unavailable"]
4045 assert state.latest_phase == "state-unavailable"
4046 assert supervisor.read_service_port("local") is None
4047 managed.terminate.assert_called_once()
4048 managed.cleanup.assert_called_once_with()
4049
4050
4051def test_mlx_ready_clears_stale_local_context_and_capacity_cache(
4052 monkeypatch,
4053) -> None:
4054 from solstone.think.providers import local_server
4055
4056 journal = Path(supervisor.get_journal())
4057 health = journal / "health"
4058 health.mkdir(parents=True, exist_ok=True)
4059 (health / "local.ctx").write_text("32768", encoding="utf-8")
4060 supervisor.write_service_port("local", 11111)
4061 monkeypatch.setattr(local_server, "fetch_props", lambda _port: None)
4062 local_server.reset_parallel_slots_cache()
4063 assert local_server.read_server_capacity().parallel_slots == 2
4064
4065 plan = _mlx_plan()
4066 state = supervisor._provider_runtime_states["local"]
4067 state.latest_plan = plan
4068 state.latest_phase = "starting"
4069 state.desired_fingerprint = plan.desired_fingerprint_sha256
4070 state.retry.attempt_count = 1
4071 state.generation = 1
4072 managed = _FakeManaged(supervisor.MLX_SERVER_PROCESS_NAME)
4073 state.start_fence = supervisor._provider_fence(state, 1)
4074 state.start_future = _future_with(
4075 supervisor.ProviderLaunchOutcome(
4076 status="ready",
4077 reason_code="probe-ready",
4078 detail={"port": 45678},
4079 managed=managed,
4080 )
4081 )
4082
4083 assert supervisor._handle_provider_start_result(state, []) is True
4084
4085 assert not (health / "local.ctx").exists()
4086 assert supervisor.read_service_port("local") == 45678
4087 assert local_server.read_server_capacity().parallel_slots == 1
4088 assert local_server.read_server_capacity().source == "default"
4089
4090
4091def test_local_capacity_observed_before_ready_is_reset_on_ready(
4092 monkeypatch,
4093) -> None:
4094 from solstone.think.providers import local_server
4095
4096 monkeypatch.setattr(local_server, "fetch_props", lambda _port: None)
4097 local_server.reset_parallel_slots_cache()
4098 assert local_server.read_server_capacity().parallel_slots == 1
4099
4100 plan = _cuda_plan()
4101 state = supervisor._provider_runtime_states["local"]
4102 state.latest_plan = plan
4103 state.latest_phase = "starting"
4104 state.desired_fingerprint = plan.desired_fingerprint_sha256
4105 state.retry.attempt_count = 1
4106 state.generation = 1
4107 managed = _FakeManaged()
4108 state.start_fence = supervisor._provider_fence(state, 1)
4109 state.start_future = _future_with(
4110 supervisor.ProviderLaunchOutcome(
4111 status="ready",
4112 reason_code="probe-ready",
4113 detail={"port": 45678},
4114 managed=managed,
4115 )
4116 )
4117
4118 assert supervisor._handle_provider_start_result(state, []) is True
4119
4120 assert supervisor.read_service_port("local") == 45678
4121 assert local_server.read_local_context_window() == plan.context_tokens
4122 assert local_server.read_server_capacity().parallel_slots == 2
4123 assert local_server.read_server_capacity().source == "local_ctx"
4124
4125
4126def test_fenced_parakeet_ready_result_writes_placement_after_ownership(
4127 monkeypatch,
4128) -> None:
4129 from solstone.think.providers import parakeet_server
4130
4131 plan = _parakeet_plan("vulkan")
4132 state = supervisor._provider_runtime_states["parakeet"]
4133 state.latest_plan = plan
4134 state.latest_phase = "starting"
4135 state.desired_fingerprint = plan.desired_fingerprint_sha256
4136 state.retry.attempt_count = 1
4137 state.generation = 1
4138 managed = _FakeManaged(supervisor.PARAKEET_SERVER_PROCESS_NAME)
4139 ports: list[tuple[str, int]] = []
4140 order: list[str] = []
4141 state.start_fence = supervisor._provider_fence(state, 1)
4142 state.start_future = _future_with(
4143 supervisor.ProviderLaunchOutcome(
4144 status="ready",
4145 reason_code="probe-ready",
4146 detail={"port": 45678, "placement": "gpu"},
4147 managed=managed,
4148 )
4149 )
4150 monkeypatch.setattr(
4151 supervisor,
4152 "write_service_port",
4153 lambda service, port: order.append("port") or ports.append((service, port)),
4154 )
4155 monkeypatch.setattr(
4156 parakeet_server,
4157 "write_parakeet_placement",
4158 lambda placement: order.append(f"placement:{placement}"),
4159 )
4160 original_write = supervisor._write_provider_runtime
4161
4162 def write_with_order(*args, **kwargs):
4163 order.append(f"runtime:{kwargs['phase']}")
4164 return original_write(*args, **kwargs)
4165
4166 monkeypatch.setattr(supervisor, "_write_provider_runtime", write_with_order)
4167
4168 assert supervisor._handle_provider_start_result(state, []) is True
4169
4170 assert ports == [("parakeet-cpp", 45678)]
4171 assert order[:3] == ["runtime:ready", "placement:gpu", "port"]
4172
4173
4174def test_boot_incarnation_invalidates_late_start_result(monkeypatch):
4175 plan = _local_plan()
4176 state = supervisor._provider_runtime_states["local"]
4177 state.latest_plan = plan
4178 state.latest_phase = "starting"
4179 state.desired_fingerprint = plan.desired_fingerprint_sha256
4180 state.retry.attempt_count = 1
4181 state.generation = 1
4182 stale_fence = supervisor.ProviderFence(
4183 incarnation="old-boot",
4184 generation=1,
4185 fingerprint=plan.desired_fingerprint_sha256,
4186 attempt=1,
4187 )
4188 managed = _FakeManaged()
4189 state.start_fence = stale_fence
4190 state.start_future = _future_with(
4191 supervisor.ProviderLaunchOutcome(
4192 status="ready",
4193 reason_code="probe-ready",
4194 detail={"port": 45678},
4195 managed=managed,
4196 )
4197 )
4198 published: list[tuple[str, int]] = []
4199 monkeypatch.setattr(
4200 supervisor,
4201 "write_service_port",
4202 lambda service, port: published.append((service, port)),
4203 )
4204
4205 assert supervisor._handle_provider_start_result(state, []) is True
4206
4207 assert published == []
4208 managed.terminate.assert_called_once()
4209 managed.cleanup.assert_called_once_with()
4210 assert read_runtime_health("local")["reason_code"] == "stale-result-ignored"
4211
4212
4213def test_superseded_attempt_cannot_publish_or_clear_newer_port(monkeypatch):
4214 plan = _local_plan()
4215 state = supervisor._provider_runtime_states["local"]
4216 state.latest_plan = plan
4217 state.latest_phase = "ready"
4218 state.desired_fingerprint = "fp-new"
4219 state.retry.attempt_count = 2
4220 state.generation = 2
4221 supervisor.write_service_port("local", 22222)
4222 write_runtime_health(
4223 _runtime_record(
4224 "local",
4225 phase="ready",
4226 fingerprint="fp-new",
4227 generation=2,
4228 attempt=2,
4229 process={
4230 "name": supervisor.LOCAL_SERVER_PROCESS_NAME,
4231 "pid": 12345,
4232 "ref": "ref-new",
4233 "port": 22222,
4234 },
4235 )
4236 )
4237 old_fence = supervisor.ProviderFence(
4238 incarnation=supervisor._PROVIDER_INCARNATION,
4239 generation=1,
4240 fingerprint="fp-old",
4241 attempt=1,
4242 )
4243 old_managed = _FakeManaged()
4244 state.start_fence = old_fence
4245 state.start_future = _future_with(
4246 supervisor.ProviderLaunchOutcome(
4247 status="ready",
4248 reason_code="probe-ready",
4249 detail={"port": 11111},
4250 managed=old_managed,
4251 )
4252 )
4253 published: list[tuple[str, int]] = []
4254 monkeypatch.setattr(
4255 supervisor,
4256 "write_service_port",
4257 lambda service, port: published.append((service, port)),
4258 )
4259
4260 assert supervisor._handle_provider_start_result(state, []) is True
4261
4262 assert published == []
4263 assert supervisor.read_service_port("local") == 22222
4264 old_managed.terminate.assert_called_once()
4265 old_managed.cleanup.assert_called_once_with()
4266
4267
4268@pytest.mark.parametrize(
4269 ("provider", "plan", "managed_name", "detail"),
4270 [
4271 (
4272 "local",
4273 _local_plan(),
4274 supervisor.LOCAL_SERVER_PROCESS_NAME,
4275 {"port": 11111},
4276 ),
4277 (
4278 "parakeet",
4279 _parakeet_plan("vulkan"),
4280 supervisor.PARAKEET_SERVER_PROCESS_NAME,
4281 {"port": 11111, "placement": "gpu"},
4282 ),
4283 ],
4284)
4285def test_superseded_ready_result_writes_no_context_or_placement(
4286 monkeypatch,
4287 provider: str,
4288 plan: supervisor.LocalServerLaunchPlan | supervisor.ParakeetServerLaunchPlan,
4289 managed_name: str,
4290 detail: dict[str, Any],
4291) -> None:
4292 from solstone.think.providers import local_server, parakeet_server
4293
4294 state = supervisor._provider_runtime_states[provider]
4295 state.latest_plan = plan
4296 state.latest_phase = "ready"
4297 state.desired_fingerprint = "fp-new"
4298 state.retry.attempt_count = 2
4299 state.generation = 2
4300 state.start_fence = supervisor.ProviderFence(
4301 incarnation=supervisor._PROVIDER_INCARNATION,
4302 generation=1,
4303 fingerprint="fp-old",
4304 attempt=1,
4305 )
4306 state.start_future = _future_with(
4307 supervisor.ProviderLaunchOutcome(
4308 status="ready",
4309 reason_code="probe-ready",
4310 detail=detail,
4311 managed=_FakeManaged(managed_name),
4312 )
4313 )
4314 context_writes: list[int] = []
4315 placement_writes: list[str] = []
4316 monkeypatch.setattr(
4317 local_server,
4318 "write_local_context_window",
4319 lambda tokens: context_writes.append(tokens),
4320 )
4321 monkeypatch.setattr(
4322 parakeet_server,
4323 "write_parakeet_placement",
4324 lambda placement: placement_writes.append(placement),
4325 )
4326
4327 assert supervisor._handle_provider_start_result(state, []) is True
4328
4329 assert context_writes == []
4330 assert placement_writes == []
4331
4332
4333def test_provider_reconcilers_keep_local_and_parakeet_state_independent(
4334 monkeypatch,
4335) -> None:
4336 local_plan = _local_plan()
4337 parakeet_plan = _parakeet_plan()
4338 launches: list[tuple[str, int]] = []
4339 local = supervisor._provider_runtime_states["local"]
4340 parakeet = supervisor._provider_runtime_states["parakeet"]
4341 local.latest_phase = "starting"
4342 local.latest_plan = local_plan
4343 local.desired_fingerprint = local_plan.desired_fingerprint_sha256
4344 local.retry.desired_fingerprint = local_plan.desired_fingerprint_sha256
4345 local.next_truth_at = 9999.0
4346 parakeet.latest_phase = "starting"
4347 parakeet.latest_plan = parakeet_plan
4348 parakeet.desired_fingerprint = parakeet_plan.desired_fingerprint_sha256
4349 parakeet.retry.desired_fingerprint = parakeet_plan.desired_fingerprint_sha256
4350 parakeet.next_truth_at = 9999.0
4351
4352 def start_worker(provider, _plan_arg, fence, _cancel_event):
4353 launches.append((provider, fence.attempt))
4354 return supervisor.ProviderLaunchOutcome(
4355 status="launch-failed",
4356 reason_code="launch-failed",
4357 detail={"provider": provider},
4358 )
4359
4360 monkeypatch.setattr(supervisor, "_provider_executor", lambda: _InlineExecutor())
4361 monkeypatch.setattr(supervisor.time, "monotonic", lambda: 100.0)
4362 monkeypatch.setattr(supervisor, "_provider_start_worker", start_worker)
4363
4364 asyncio.run(supervisor._reconcile_local_provider_runtime([]))
4365 asyncio.run(supervisor._reconcile_local_provider_runtime([]))
4366 asyncio.run(supervisor._reconcile_parakeet_provider_runtime([]))
4367 asyncio.run(supervisor._reconcile_parakeet_provider_runtime([]))
4368
4369 assert launches == [("local", 1), ("parakeet", 1)]
4370 assert local.latest_phase == "backoff"
4371 assert parakeet.latest_phase == "backoff"
4372 assert local.retry.desired_fingerprint == local_plan.desired_fingerprint_sha256
4373 assert parakeet.retry.desired_fingerprint == (
4374 parakeet_plan.desired_fingerprint_sha256
4375 )
4376
4377
4378def test_spawn_failure_leaves_no_port_file(monkeypatch):
4379 from solstone.think.providers import local_server
4380
4381 plan = _local_plan()
4382 monkeypatch.setattr(
4383 local_server, "write_local_context_window", lambda _tokens: None
4384 )
4385 monkeypatch.setattr(
4386 supervisor,
4387 "_launch_process",
4388 lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("spawn failed")),
4389 )
4390
4391 outcome = supervisor.start_local_server(plan, _FakeReservation(port=34567))
4392
4393 assert outcome.status == "launch-failed"
4394 assert supervisor.read_service_port("local") is None
4395
4396
4397@pytest.mark.parametrize(
4398 "backend",
4399 ["vulkan", "cuda", "mlx", "parakeet-vulkan", "parakeet-cpu"],
4400)
4401@pytest.mark.parametrize("status", ["warmup-timeout", "exited", "launch-failed"])
4402def test_non_ready_outcome_cleanup_runs_before_backoff_record(
4403 monkeypatch,
4404 backend: str,
4405 status: supervisor.LaunchOutcomeStatus,
4406) -> None:
4407 provider = "parakeet" if backend.startswith("parakeet") else "local"
4408 state = supervisor._provider_runtime_states[provider]
4409 state.latest_phase = "starting"
4410 state.desired_fingerprint = f"fp-{backend}"
4411 state.retry.attempt_count = 1
4412 state.retry.desired_fingerprint = state.desired_fingerprint
4413 state.generation = 1
4414 fence = supervisor._provider_fence(state, 1)
4415 managed_name = (
4416 supervisor.PARAKEET_SERVER_PROCESS_NAME
4417 if provider == "parakeet"
4418 else (
4419 supervisor.MLX_SERVER_PROCESS_NAME
4420 if backend == "mlx"
4421 else supervisor.LOCAL_SERVER_PROCESS_NAME
4422 )
4423 )
4424 managed = _FakeManaged(managed_name)
4425 state.start_fence = fence
4426 state.start_future = _future_with(
4427 supervisor.ProviderLaunchOutcome(
4428 status=status,
4429 reason_code=(
4430 "warmup-timeout"
4431 if status == "warmup-timeout"
4432 else ("process-exited" if status == "exited" else "launch-failed")
4433 ),
4434 detail={"backend": backend, "port": 45678},
4435 managed=managed,
4436 )
4437 )
4438 order: list[str] = []
4439
4440 monkeypatch.setattr(
4441 supervisor,
4442 "_terminate_cleanup_handle",
4443 lambda managed_arg, *, reason, state_name=None: order.append(
4444 f"cleanup:{managed_arg.name}:{reason}"
4445 ),
4446 )
4447 original_write = supervisor._write_provider_runtime
4448
4449 def write_with_order(*args, **kwargs):
4450 order.append(f"write:{kwargs['phase']}")
4451 return original_write(*args, **kwargs)
4452
4453 monkeypatch.setattr(supervisor, "_write_provider_runtime", write_with_order)
4454
4455 assert supervisor._handle_provider_start_result(state, []) is True
4456
4457 assert order[0].startswith(f"cleanup:{managed_name}:")
4458 assert order[1] == "write:backoff"
4459 assert state.latest_phase == "backoff"
4460
4461
4462def test_non_ready_cleanup_runs_before_failed_record(monkeypatch):
4463 state = supervisor._provider_runtime_states["local"]
4464 state.latest_phase = "starting"
4465 state.desired_fingerprint = "fp-local"
4466 state.retry.attempt_count = len(supervisor.PROVIDER_RETRY_SCHEDULE_SECONDS)
4467 state.generation = 1
4468 fence = supervisor._provider_fence(
4469 state, len(supervisor.PROVIDER_RETRY_SCHEDULE_SECONDS)
4470 )
4471 managed = _FakeManaged()
4472 state.start_fence = fence
4473 state.start_future = _future_with(
4474 supervisor.ProviderLaunchOutcome(
4475 status="warmup-timeout",
4476 reason_code="warmup-timeout",
4477 detail={"port": 45678},
4478 managed=managed,
4479 )
4480 )
4481 order: list[str] = []
4482 monkeypatch.setattr(
4483 supervisor,
4484 "_terminate_cleanup_handle",
4485 lambda managed_arg, *, reason, state_name=None: order.append("cleanup"),
4486 )
4487 original_write = supervisor._write_provider_runtime
4488
4489 def write_with_order(*args, **kwargs):
4490 order.append(f"write:{kwargs['phase']}")
4491 return original_write(*args, **kwargs)
4492
4493 monkeypatch.setattr(supervisor, "_write_provider_runtime", write_with_order)
4494
4495 assert supervisor._handle_provider_start_result(state, []) is True
4496
4497 assert order[:2] == ["cleanup", "write:failed"]
4498 assert state.latest_phase == "failed"
4499
4500
4501def test_launch_helper_returns_reserved_port_without_publishing(monkeypatch):
4502 from solstone.think.providers import local_server, local_vulkan
4503
4504 plan = _local_plan()
4505 managed = _FakeManaged()
4506 ports: list[tuple[str, int]] = []
4507 spawned: list[list[str]] = []
4508
4509 monkeypatch.setattr(
4510 supervisor,
4511 "write_service_port",
4512 lambda service, port: ports.append((service, port)),
4513 )
4514 monkeypatch.setattr(
4515 local_server,
4516 "write_local_context_window",
4517 lambda _tokens: None,
4518 )
4519 monkeypatch.setattr(
4520 local_server,
4521 "_probe_health",
4522 lambda _port: (local_server.STATE_READY, None),
4523 )
4524 monkeypatch.setattr(local_server, "fetch_props", lambda _port: None)
4525 monkeypatch.setattr(local_vulkan, "device_local_used_mib", lambda _index: None)
4526 monkeypatch.setattr(
4527 supervisor,
4528 "_launch_process",
4529 lambda name, cmd, **_kwargs: spawned.append(cmd) or managed,
4530 )
4531
4532 assert not hasattr(plan, "port")
4533
4534 outcome = supervisor.start_local_server(plan, _FakeReservation(port=45678))
4535
4536 assert outcome.status == "ready"
4537 assert outcome.managed is managed
4538 assert ports == []
4539 assert spawned[0][spawned[0].index("--port") + 1] == "45678"
4540 assert RUNTIME_PHASES >= {"starting", "backoff", "ready"}