solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4import asyncio
5import importlib
6import io
7import json
8import logging
9import os
10import socket
11import subprocess
12import sys
13import threading
14import time
15from datetime import date
16from pathlib import Path
17from types import SimpleNamespace
18from unittest.mock import MagicMock
19
20import psutil
21import pytest
22
23from solstone.think.maint import MaintTask, MaintTaskResult
24from solstone.think.processing import (
25 DISPLAY_POWERSAVE_UNAVAILABLE,
26 DisplayPowersaveReading,
27 DisplayPowersaveSettings,
28 GateSettings,
29 ProcessingSettings,
30 TimeWindowSettings,
31)
32from solstone.think.providers.artifact_proof import ReadinessOutcome
33from tests.helpers.module_mocks import (
34 capturing_thread_constructor,
35 module_mock,
36)
37
38
39def _mlx_readiness(
40 *,
41 model_installed: bool = True,
42 ram_sufficient: bool = True,
43 platform_supported: bool = True,
44 package_available: bool = True,
45 runtime_dir: str = "/tmp/snap",
46 model_id: str = "mlx-model",
47) -> ReadinessOutcome:
48 ready = (
49 model_installed and ram_sufficient and platform_supported and package_available
50 )
51 return ReadinessOutcome(
52 provider="local",
53 status="ready" if ready else "missing-or-mismatched",
54 reason_code="ready" if ready else "manifest_missing",
55 target={"model_id": model_id},
56 install={
57 "install_state": "idle",
58 "install_error": None,
59 "error_code": None,
60 "attempt_id": None,
61 "progress_bytes_received": None,
62 "progress_bytes_total": None,
63 "last_transition_at": None,
64 "last_progress_at": None,
65 },
66 host={
67 "platform_supported": platform_supported,
68 "package_available": package_available,
69 "ram_sufficient": ram_sufficient,
70 },
71 artifacts={
72 "model_installed": model_installed,
73 "runtime_dir": runtime_dir,
74 },
75 proof={},
76 )
77
78
79@pytest.fixture(autouse=True)
80def _default_thinking_engine_selected(monkeypatch):
81 mod = importlib.import_module("solstone.think.supervisor")
82 monkeypatch.setattr(mod, "no_thinking_engine_chosen", lambda: False)
83
84
85def test_sd_notify_no_socket_is_noop(monkeypatch):
86 from solstone.think.supervisor import _sd_notify
87
88 monkeypatch.delenv("NOTIFY_SOCKET", raising=False)
89 _sd_notify("READY=1")
90
91
92def test_sd_notify_sends_payload(tmp_path, monkeypatch):
93 from solstone.think.supervisor import _sd_notify
94
95 sock_path = tmp_path / "notify.sock"
96 with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as listener:
97 listener.bind(str(sock_path))
98 listener.settimeout(1)
99 monkeypatch.setenv("NOTIFY_SOCKET", str(sock_path))
100
101 _sd_notify("READY=1")
102
103 assert listener.recv(1024) == b"READY=1"
104
105
106def test_start_sense(tmp_path, mock_callosum, monkeypatch):
107 """Test that sense launches correctly."""
108 mod = importlib.import_module("solstone.think.supervisor")
109
110 started = []
111
112 class DummyProc:
113 def __init__(self):
114 self.stdout = io.StringIO()
115 self.stderr = io.StringIO()
116 self.pid = 12345
117
118 def terminate(self):
119 pass
120
121 def wait(self, timeout=None):
122 pass
123
124 def fake_popen(
125 cmd,
126 stdin=None,
127 stdout=None,
128 stderr=None,
129 text=False,
130 bufsize=-1,
131 process_group=None,
132 env=None,
133 **_kwargs,
134 ):
135 proc = DummyProc()
136 started.append((cmd, stdout, stderr))
137 return proc
138
139 monkeypatch.setattr(mod.subprocess, "Popen", fake_popen)
140 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
141
142 # Test start_sense()
143 sense_proc = mod.start_sense()
144 assert sense_proc is not None
145 assert any(cmd == ["journal", "sense", "-v"] for cmd, _, _ in started)
146
147 # Check that stdout and stderr capture pipes
148 for cmd, stdout, stderr in started:
149 assert stdout == subprocess.PIPE
150 assert stderr == subprocess.PIPE
151
152
153def test_launch_process_records_service_state(monkeypatch):
154 mod = importlib.import_module("solstone.think.supervisor")
155 mod._SERVICE_STATE.clear()
156
157 process = MagicMock()
158 process.pid = 12345
159 managed = mod.RunnerManagedProcess(
160 process=process,
161 name="unit",
162 log_writer=MagicMock(),
163 cmd=["journal", "sense"],
164 _threads=[],
165 ref="ref-1",
166 _start_time=100.0,
167 _callosum=None,
168 )
169
170 def fake_spawn(cmd, *, ref=None, callosum=None, day=None, env=None):
171 assert cmd == ["journal", "sense"]
172 assert ref == "ref-1"
173 assert day is None
174 return managed
175
176 monkeypatch.setattr(mod.RunnerManagedProcess, "spawn", fake_spawn)
177
178 result = mod._launch_process(
179 "unit",
180 ["journal", "sense"],
181 restart=True,
182 shutdown_timeout=7,
183 ref="ref-1",
184 )
185
186 assert result is managed
187 assert isinstance(result, mod.RunnerManagedProcess)
188 assert mod._SERVICE_STATE["unit"] == {
189 "restart": True,
190 "shutdown_timeout": 7,
191 }
192
193
194def test_parse_args_remote_flag():
195 """Test that parse_args includes --remote flag."""
196 mod = importlib.reload(importlib.import_module("solstone.think.supervisor"))
197
198 parser = mod.parse_args()
199 args = parser.parse_args(["--remote", "https://server/ingest/key"])
200
201 assert args.remote == "https://server/ingest/key"
202
203
204def test_parse_args_remote_flag_optional():
205 """Test that --remote is optional."""
206 mod = importlib.reload(importlib.import_module("solstone.think.supervisor"))
207
208 parser = mod.parse_args()
209 args = parser.parse_args([])
210
211 assert args.remote is None
212
213
214def test_parse_args_app_supervised_flag():
215 mod = importlib.reload(importlib.import_module("solstone.think.supervisor"))
216
217 parser = mod.parse_args()
218 args = parser.parse_args(
219 ["5016", "--app-supervised", "--no-daily", "--no-schedule"]
220 )
221
222 assert args.port == 5016
223 assert args.app_supervised is True
224 assert args.no_daily is True
225 assert args.no_schedule is True
226
227
228def test_parse_args_lifecycle_verb_hint(monkeypatch, capsys):
229 mod = importlib.reload(importlib.import_module("solstone.think.supervisor"))
230 monkeypatch.setattr(sys, "argv", ["sol", "supervisor", "stop"])
231
232 parser = mod.parse_args()
233 with pytest.raises(SystemExit) as exc_info:
234 parser.parse_args(["stop"])
235
236 captured = capsys.readouterr()
237 assert exc_info.value.code == 2
238 assert (
239 "journal supervisor is the server-launch command (takes a port). "
240 "For lifecycle, use: journal service <verb>. "
241 "Did you mean: journal service stop ?"
242 ) in captured.err
243
244
245def test_shutdown_stops_in_reverse_order(monkeypatch):
246 """Shutdown stops services in reverse order."""
247 mod = importlib.import_module("solstone.think.supervisor")
248 operations = []
249
250 class MockManaged:
251 def __init__(self, name):
252 self.name = name
253 self.terminate = MagicMock(
254 side_effect=lambda timeout=None: operations.append(
255 ("terminate", self.name, timeout)
256 )
257 )
258 self.cleanup = MagicMock(
259 side_effect=lambda: operations.append(("cleanup", self.name))
260 )
261
262 procs = [
263 MockManaged("convey"),
264 MockManaged("sense"),
265 MockManaged("cortex"),
266 ]
267 mod._SERVICE_STATE.clear()
268 for managed in procs:
269 mod._SERVICE_STATE[managed.name] = {
270 "restart": True,
271 "shutdown_timeout": 15,
272 }
273
274 for managed in reversed(procs):
275 mod._stop_process(managed)
276
277 assert operations == [
278 ("terminate", "cortex", 15),
279 ("cleanup", "cortex"),
280 ("terminate", "sense", 15),
281 ("cleanup", "sense"),
282 ("terminate", "convey", 15),
283 ("cleanup", "convey"),
284 ]
285
286
287def test_graceful_shutdown_calls_stop_process_for_each_managed_proc(
288 tmp_path, monkeypatch
289):
290 """The main shutdown path stops managed services in reverse startup order."""
291 mod = importlib.reload(importlib.import_module("solstone.think.supervisor"))
292 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
293 monkeypatch.delenv("SOL_SUPERVISOR_SPAWNED", raising=False)
294 monkeypatch.setattr(
295 sys,
296 "argv",
297 ["supervisor", "0", "--no-daily", "--no-schedule"],
298 )
299 monkeypatch.setattr(mod, "run_pending_tasks", lambda *a, **k: [])
300 monkeypatch.setattr(mod, "_sweep_orphaned_sol_processes", lambda *_a, **_k: 0)
301 monkeypatch.setattr(mod.time, "sleep", lambda _seconds: None)
302 monkeypatch.setattr(mod, "start_callosum_in_process", lambda: None)
303 monkeypatch.setattr(mod, "stop_callosum_in_process", lambda **_kwargs: None)
304 monkeypatch.setattr(mod, "wait_for_convey_ready", lambda _proc: True)
305
306 class FakeCallosumConnection:
307 def __init__(self, *args, **kwargs):
308 pass
309
310 def start(self, *args, **kwargs):
311 pass
312
313 def emit(self, *args, **kwargs):
314 pass
315
316 def stop(self):
317 pass
318
319 monkeypatch.setattr(mod, "CallosumConnection", FakeCallosumConnection)
320
321 procs = []
322 for name in ["convey", "sense", "cortex", "spl"]:
323 managed = _TaskManagedStub(cmd=["journal", name])
324 managed.name = name
325 procs.append(managed)
326
327 monkeypatch.setattr(
328 mod,
329 "start_convey_server",
330 lambda verbose, debug=False, port=0: (procs[0], 5015),
331 )
332 monkeypatch.setattr(mod, "start_sense", lambda: procs[1])
333 monkeypatch.setattr(mod, "start_cortex_server", lambda: procs[2])
334 monkeypatch.setattr(mod, "start_spl_service", lambda: procs[3])
335
336 stop_order = []
337 monkeypatch.setattr(
338 mod,
339 "_stop_process",
340 lambda managed, **_kwargs: stop_order.append(managed.name),
341 )
342
343 def interrupt_supervise(coro):
344 coro.close()
345 raise KeyboardInterrupt
346
347 monkeypatch.setattr(mod.asyncio, "run", interrupt_supervise)
348
349 try:
350 mod.main()
351 finally:
352 os.environ.pop("SOL_SUPERVISOR_SPAWNED", None)
353
354 assert stop_order == ["spl", "cortex", "sense", "convey"]
355
356
357@pytest.mark.parametrize(
358 ("convey_accepting", "expected_ready"),
359 [(True, True), (False, False)],
360)
361def test_supervisor_readiness_marker_requires_started_convey_accepting(
362 tmp_path, monkeypatch, convey_accepting, expected_ready
363):
364 """A started Convey process must still accept before supervisor marks ready."""
365 mod = importlib.reload(importlib.import_module("solstone.think.supervisor"))
366 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
367 monkeypatch.delenv("SOL_SUPERVISOR_SPAWNED", raising=False)
368 monkeypatch.setattr(
369 sys,
370 "argv",
371 ["supervisor", "0", "--no-daily", "--no-schedule"],
372 )
373 monkeypatch.setattr(mod, "run_pending_tasks", lambda *a, **k: [])
374 monkeypatch.setattr(mod, "_sweep_orphaned_sol_processes", lambda *_a, **_k: 0)
375 monkeypatch.setattr(mod.time, "sleep", lambda _seconds: None)
376 monkeypatch.setattr(mod, "start_callosum_in_process", lambda: None)
377 monkeypatch.setattr(mod, "stop_callosum_in_process", lambda **_kwargs: None)
378 monkeypatch.setattr(mod, "wait_for_convey_ready", lambda _proc: True)
379 monkeypatch.setattr(mod, "is_solstone_up", lambda timeout=1.0: convey_accepting)
380 monkeypatch.setattr(mod, "is_local_provider_needed", lambda: False)
381 monkeypatch.setattr(mod, "read_service_port", lambda _name: 5015)
382
383 class FakeCallosumConnection:
384 def __init__(self, *args, **kwargs):
385 pass
386
387 def start(self, *args, **kwargs):
388 pass
389
390 def emit(self, *args, **kwargs):
391 pass
392
393 def stop(self):
394 pass
395
396 monkeypatch.setattr(mod, "CallosumConnection", FakeCallosumConnection)
397
398 procs = []
399 for name in ["convey", "sense", "cortex", "spl"]:
400 managed = _TaskManagedStub(cmd=["journal", name])
401 managed.name = name
402 procs.append(managed)
403
404 monkeypatch.setattr(
405 mod,
406 "start_convey_server",
407 lambda verbose, debug=False, port=0: (procs[0], 5015),
408 )
409 monkeypatch.setattr(mod, "start_sense", lambda: procs[1])
410 monkeypatch.setattr(mod, "start_cortex_server", lambda: procs[2])
411 monkeypatch.setattr(mod, "start_spl_service", lambda: procs[3])
412 monkeypatch.setattr(mod, "_stop_process", lambda managed, **_kwargs: None)
413
414 events: list[str] = []
415 monkeypatch.setattr(mod, "signal_ready", lambda: events.append("ready"))
416
417 def interrupt_supervise(coro):
418 coro.close()
419 raise KeyboardInterrupt
420
421 monkeypatch.setattr(mod.asyncio, "run", interrupt_supervise)
422
423 try:
424 mod.main()
425 finally:
426 os.environ.pop("SOL_SUPERVISOR_SPAWNED", None)
427
428 assert ("ready" in events) is expected_ready
429
430
431def _run_supervisor_main_for_shutdown_knobs(tmp_path, monkeypatch, *, argv):
432 mod = importlib.reload(importlib.import_module("solstone.think.supervisor"))
433 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
434 monkeypatch.delenv("SOL_SUPERVISOR_SPAWNED", raising=False)
435 monkeypatch.delenv("SOLSTONE_APP_SUPERVISED", raising=False)
436 monkeypatch.setattr(sys, "argv", argv)
437 monkeypatch.setattr(mod, "run_pending_tasks", lambda *a, **k: [])
438 monkeypatch.setattr(mod, "_sweep_orphaned_sol_processes", lambda *_a, **_k: 0)
439 monkeypatch.setattr(mod.time, "sleep", lambda _seconds: None)
440 monkeypatch.setattr(mod, "start_callosum_in_process", lambda: None)
441 monkeypatch.setattr(mod, "is_local_provider_needed", lambda: False)
442
443 class FakeCallosumConnection:
444 def __init__(self, *args, **kwargs):
445 pass
446
447 def start(self, *args, **kwargs):
448 pass
449
450 def emit(self, *args, **kwargs):
451 pass
452
453 def stop(self):
454 pass
455
456 monkeypatch.setattr(mod, "CallosumConnection", FakeCallosumConnection)
457
458 managed = _TaskManagedStub(cmd=["journal", "sense"])
459 managed.name = "sense"
460 monkeypatch.setattr(mod, "start_sense", lambda: managed)
461
462 events: list[str] = []
463 captures: dict[str, list] = {
464 "task_shutdown": [],
465 "stop_process": [],
466 "callosum_join": [],
467 }
468
469 class FakeTaskQueue:
470 def __init__(self, *args, **kwargs):
471 self.caps = {}
472
473 @staticmethod
474 def get_command_name(cmd):
475 return mod._command_partition(cmd)
476
477 def set_cap(self, cmd_name, seconds):
478 self.caps[cmd_name] = seconds
479
480 def set_ready(self):
481 pass
482
483 def shutdown(self, *, timeout):
484 captures["task_shutdown"].append(timeout)
485
486 monkeypatch.setattr(mod, "TaskQueue", FakeTaskQueue)
487 monkeypatch.setattr(mod, "signal_ready", lambda: events.append("ready"))
488
489 def record_watcher_start():
490 events.append("watcher")
491 assert mod._managed_procs == [managed]
492
493 monkeypatch.setattr(mod, "start_parent_death_watcher", record_watcher_start)
494 monkeypatch.setattr(
495 mod,
496 "_stop_process",
497 lambda proc, *, timeout_cap=None: captures["stop_process"].append(
498 (proc.name, timeout_cap)
499 ),
500 )
501 monkeypatch.setattr(
502 mod,
503 "stop_callosum_in_process",
504 lambda *, join_timeout=5.0: captures["callosum_join"].append(join_timeout),
505 )
506 exit_now = MagicMock()
507 monkeypatch.setattr(mod.os, "_exit", exit_now)
508
509 def interrupt_supervise(coro):
510 events.append("run")
511 coro.close()
512 raise KeyboardInterrupt
513
514 monkeypatch.setattr(mod.asyncio, "run", interrupt_supervise)
515
516 try:
517 mod.main()
518 finally:
519 os.environ.pop("SOL_SUPERVISOR_SPAWNED", None)
520
521 return mod, captures, events, exit_now
522
523
524def test_app_supervised_main_uses_watcher_and_compressed_shutdown_knobs(
525 tmp_path, monkeypatch
526):
527 from solstone.think import install_guard, service
528
529 reconcile = MagicMock()
530 install_wrappers = MagicMock()
531 monkeypatch.setattr(service, "reconcile_installed_unit", reconcile)
532 monkeypatch.setattr(install_guard, "install_wrappers", install_wrappers)
533
534 mod, captures, events, exit_now = _run_supervisor_main_for_shutdown_knobs(
535 tmp_path,
536 monkeypatch,
537 argv=[
538 "supervisor",
539 "0",
540 "--app-supervised",
541 "--no-daily",
542 "--no-schedule",
543 "--no-convey",
544 "--no-cortex",
545 "--no-spl",
546 ],
547 )
548
549 assert events == ["ready", "watcher", "run"]
550 assert captures["task_shutdown"] == [mod.APP_SUPERVISED_TASK_DRAIN_S]
551 assert captures["stop_process"] == [("sense", mod.APP_SUPERVISED_CHILD_STOP_S)]
552 assert captures["callosum_join"] == [mod.APP_SUPERVISED_CALLOSUM_JOIN_S]
553 exit_now.assert_not_called()
554 reconcile.assert_not_called()
555 install_wrappers.assert_not_called()
556
557
558def test_default_main_uses_default_shutdown_knobs(tmp_path, monkeypatch):
559 mod, captures, events, _exit_now = _run_supervisor_main_for_shutdown_knobs(
560 tmp_path,
561 monkeypatch,
562 argv=[
563 "supervisor",
564 "0",
565 "--no-daily",
566 "--no-schedule",
567 "--no-convey",
568 "--no-cortex",
569 "--no-spl",
570 ],
571 )
572
573 assert events == ["ready", "run"]
574 assert captures["task_shutdown"] == [10]
575 assert captures["stop_process"] == [("sense", None)]
576 assert captures["callosum_join"] == [5.0]
577
578
579def test_get_command_name():
580 """Test command name extraction for queue serialization."""
581 mod = importlib.import_module("solstone.think.supervisor")
582 get = mod.TaskQueue.get_command_name
583
584 # sol X -> X
585 assert get(["journal", "indexer", "--rescan"]) == "indexer"
586 assert get(["sol", "insight", "20240101"]) == "insight"
587 assert get(["journal", "think", "--day", "20240101"]) == "daily"
588 assert get(["journal", "brain", "renew-prerequisites"]) == "brain"
589 assert get(["journal", "maintenance", "list"]) == "maintenance"
590 assert get(["journal", "maintenance", "run", "foo:bar"]) == "maintenance:foo:bar"
591 assert get(["journal", "maintenance", "run", "baz:qux"]) == "maintenance:baz:qux"
592 assert get(["journal", "maintenance", "run", "foo:bar"]) == get(
593 ["journal", "maintenance", "run", "foo:bar"]
594 )
595 assert get(["journal", "maintenance", "run", "foo:bar"]) != get(
596 ["journal", "maintenance", "run", "baz:qux"]
597 )
598
599 # Other commands -> basename
600 assert get(["/usr/bin/python", "script.py"]) == "python"
601 assert get(["custom-tool"]) == "custom-tool"
602
603 # Empty -> unknown
604 assert get([]) == "unknown"
605
606
607@pytest.mark.parametrize(
608 "cmd",
609 [
610 ["journal", "think", "--day", "20260527"],
611 ["journal", "think", "--day", "20260527", "--segment", "120000_300"],
612 [
613 "journal",
614 "think",
615 "--day",
616 "20260527",
617 "--segment",
618 "120000_300",
619 "--stream",
620 "screen",
621 ],
622 [
623 "journal",
624 "think",
625 "--day",
626 "20260527",
627 "--segment",
628 "120000_300",
629 "--flush",
630 ],
631 ["journal", "think", "--day", "20260527", "--segments"],
632 [
633 "journal",
634 "think",
635 "--activity",
636 "activity-id",
637 "--facet",
638 "work",
639 "--day",
640 "20260527",
641 ],
642 ["journal", "think", "--weekly", "-v"],
643 ["journal", "think"],
644 ["journal", "indexer", "--rescan"],
645 ["journal", "sense", "--day", "20260101"],
646 ["journal", "maintenance", "list"],
647 ["journal", "maintenance", "run", "foo:bar"],
648 ],
649)
650def test_command_partition_matches_task_queue_get_command_name(cmd):
651 mod = importlib.import_module("solstone.think.supervisor")
652 runner = importlib.import_module("solstone.think.runner")
653
654 assert runner._command_partition(cmd) == mod.TaskQueue.get_command_name(cmd)
655
656
657def _fresh_task_queue(mod, *, on_queue_change=None):
658 mod._task_queue = mod.TaskQueue(on_queue_change=on_queue_change)
659 mod._supervisor_callosum = None
660 return mod._task_queue
661
662
663def _capture_thread_starts(monkeypatch, mod):
664 spawned = []
665 monkeypatch.setattr(
666 mod,
667 "threading",
668 module_mock(
669 mod.threading,
670 Thread=capturing_thread_constructor(
671 spawned,
672 capture=lambda thread: thread._args,
673 ),
674 ),
675 )
676 return spawned
677
678
679class _CaptureTaskQueue:
680 def __init__(self):
681 self.submissions = []
682
683 def submit(self, cmd, day=None):
684 self.submissions.append({"cmd": cmd, "day": day})
685
686
687def _supervisor_processing_settings(
688 mode: str,
689 *,
690 display_powersave_enabled: bool = False,
691) -> ProcessingSettings:
692 return ProcessingSettings(
693 mode=mode,
694 gate=GateSettings(
695 time_window=TimeWindowSettings(
696 enabled=True,
697 start="02:00",
698 end="06:00",
699 ),
700 display_powersave=DisplayPowersaveSettings(
701 enabled=display_powersave_enabled
702 ),
703 ),
704 )
705
706
707def test_handle_segment_observed_live_command_marks_live(monkeypatch):
708 mod = importlib.import_module("solstone.think.supervisor")
709 capture = _CaptureTaskQueue()
710 monkeypatch.setattr(mod, "_task_queue", capture)
711
712 mod._handle_segment_observed(
713 {
714 "tract": "observe",
715 "event": "observed",
716 "day": "20260527",
717 "segment": "120000_300",
718 }
719 )
720
721 assert len(capture.submissions) == 1
722 assert capture.submissions[0]["day"] == "20260527"
723 assert "--live" in capture.submissions[0]["cmd"]
724
725
726def test_handle_segment_observed_batch_submits_nothing(monkeypatch):
727 mod = importlib.import_module("solstone.think.supervisor")
728 capture = _CaptureTaskQueue()
729 monkeypatch.setattr(mod, "_task_queue", capture)
730
731 mod._handle_segment_observed(
732 {
733 "tract": "observe",
734 "event": "observed",
735 "day": "20260527",
736 "segment": "120000_300",
737 "batch": True,
738 }
739 )
740
741 assert len(capture.submissions) == 0
742
743
744def test_handle_segment_observed_batch_leaves_flush_state_untouched(monkeypatch):
745 mod = importlib.import_module("solstone.think.supervisor")
746 capture = _CaptureTaskQueue()
747 monkeypatch.setattr(mod, "_task_queue", capture)
748
749 mod._flush_state["last_segment_ts"] = 0.0
750 mod._flush_state["day"] = None
751 mod._flush_state["segment"] = None
752 mod._flush_state["flushed"] = True
753
754 mod._handle_segment_observed(
755 {
756 "tract": "observe",
757 "event": "observed",
758 "day": "20260527",
759 "segment": "120000_300",
760 }
761 )
762
763 assert len(capture.submissions) == 1
764 live_state = {
765 "day": mod._flush_state["day"],
766 "segment": mod._flush_state["segment"],
767 "flushed": mod._flush_state["flushed"],
768 "last_segment_ts": mod._flush_state["last_segment_ts"],
769 }
770 assert live_state["day"] == "20260527"
771 assert live_state["segment"] == "120000_300"
772 assert live_state["flushed"] is False
773 assert live_state["last_segment_ts"] > 0
774
775 mod._handle_segment_observed(
776 {
777 "tract": "observe",
778 "event": "observed",
779 "day": "20260101",
780 "segment": "090000_300",
781 "batch": True,
782 }
783 )
784
785 assert len(capture.submissions) == 1
786 assert mod._flush_state["day"] == live_state["day"]
787 assert mod._flush_state["segment"] == live_state["segment"]
788 assert mod._flush_state["flushed"] == live_state["flushed"]
789 assert mod._flush_state["last_segment_ts"] == live_state["last_segment_ts"]
790
791
792def test_handle_segment_observed_live_stream_command(monkeypatch):
793 mod = importlib.import_module("solstone.think.supervisor")
794 capture = _CaptureTaskQueue()
795 monkeypatch.setattr(mod, "_task_queue", capture)
796
797 mod._handle_segment_observed(
798 {
799 "tract": "observe",
800 "event": "observed",
801 "day": "20260527",
802 "segment": "120000_300",
803 "stream": "archon",
804 }
805 )
806
807 assert len(capture.submissions) == 1
808 cmd = capture.submissions[0]["cmd"]
809 assert "--live" in cmd
810 stream_index = cmd.index("--stream")
811 assert cmd[stream_index + 1] == "archon"
812
813
814def test_handle_segment_observed_deferred_live_submits_nothing(monkeypatch):
815 mod = importlib.import_module("solstone.think.supervisor")
816 capture = _CaptureTaskQueue()
817 monkeypatch.setattr(mod, "_task_queue", capture)
818 monkeypatch.setattr(
819 mod,
820 "load_processing_settings",
821 lambda: _supervisor_processing_settings("deferred"),
822 )
823 mod._flush_state.update(
824 {
825 "last_segment_ts": 123.0,
826 "day": "20260526",
827 "segment": "110000_300",
828 "stream": "archon",
829 "flushed": True,
830 }
831 )
832 before = dict(mod._flush_state)
833
834 mod._handle_segment_observed(
835 {
836 "tract": "observe",
837 "event": "observed",
838 "day": "20260527",
839 "segment": "120000_300",
840 }
841 )
842
843 assert capture.submissions == []
844 assert dict(mod._flush_state) == before
845
846
847def test_handle_segment_observed_no_engine_submits_nothing(monkeypatch):
848 mod = importlib.import_module("solstone.think.supervisor")
849 capture = _CaptureTaskQueue()
850 monkeypatch.setattr(mod, "_task_queue", capture)
851 monkeypatch.setattr(
852 mod,
853 "load_processing_settings",
854 lambda: _supervisor_processing_settings("realtime"),
855 )
856 monkeypatch.setattr(mod, "no_thinking_engine_chosen", lambda: True)
857 mod._flush_state.update(
858 {
859 "last_segment_ts": 123.0,
860 "day": "20260526",
861 "segment": "110000_300",
862 "stream": "archon",
863 "flushed": True,
864 }
865 )
866 before = dict(mod._flush_state)
867
868 mod._handle_segment_observed(
869 {
870 "tract": "observe",
871 "event": "observed",
872 "day": "20260527",
873 "segment": "120000_300",
874 }
875 )
876
877 assert capture.submissions == []
878 assert dict(mod._flush_state) == before
879
880
881def test_handle_segment_observed_reads_processing_mode_fresh(monkeypatch):
882 mod = importlib.import_module("solstone.think.supervisor")
883 capture = _CaptureTaskQueue()
884 monkeypatch.setattr(mod, "_task_queue", capture)
885 modes = ["realtime", "deferred"]
886
887 def load_settings():
888 return _supervisor_processing_settings(modes.pop(0))
889
890 monkeypatch.setattr(mod, "load_processing_settings", load_settings)
891
892 mod._handle_segment_observed(
893 {
894 "tract": "observe",
895 "event": "observed",
896 "day": "20260527",
897 "segment": "120000_300",
898 }
899 )
900 after_realtime = dict(mod._flush_state)
901 mod._handle_segment_observed(
902 {
903 "tract": "observe",
904 "event": "observed",
905 "day": "20260527",
906 "segment": "120500_300",
907 }
908 )
909
910 assert len(capture.submissions) == 1
911 assert capture.submissions[0]["cmd"][-1] == "--live"
912 assert dict(mod._flush_state) == after_realtime
913
914
915def test_check_segment_flush_deferred_guard_submits_nothing(monkeypatch):
916 mod = importlib.import_module("solstone.think.supervisor")
917 capture = _CaptureTaskQueue()
918 monkeypatch.setattr(mod, "_task_queue", capture)
919 monkeypatch.setattr(
920 mod,
921 "load_processing_settings",
922 lambda: _supervisor_processing_settings("deferred"),
923 )
924 mod._flush_state.update(
925 {
926 "last_segment_ts": 1.0,
927 "day": "20260527",
928 "segment": "120000_300",
929 "stream": None,
930 "flushed": False,
931 }
932 )
933
934 mod._check_segment_flush()
935
936 assert capture.submissions == []
937 assert mod._flush_state["flushed"] is False
938
939
940def test_check_segment_flush_no_engine_guard_submits_nothing(monkeypatch):
941 mod = importlib.import_module("solstone.think.supervisor")
942 capture = _CaptureTaskQueue()
943 monkeypatch.setattr(mod, "_task_queue", capture)
944 monkeypatch.setattr(
945 mod,
946 "load_processing_settings",
947 lambda: _supervisor_processing_settings("realtime"),
948 )
949 monkeypatch.setattr(mod, "no_thinking_engine_chosen", lambda: True)
950 mod._flush_state.update(
951 {
952 "last_segment_ts": 1.0,
953 "day": "20260527",
954 "segment": "120000_300",
955 "stream": None,
956 "flushed": False,
957 }
958 )
959
960 mod._check_segment_flush()
961
962 assert capture.submissions == []
963 assert mod._flush_state["flushed"] is False
964
965
966def test_check_segment_flush_realtime_submits(monkeypatch):
967 mod = importlib.import_module("solstone.think.supervisor")
968 capture = _CaptureTaskQueue()
969 monkeypatch.setattr(mod, "_task_queue", capture)
970 monkeypatch.setattr(
971 mod,
972 "load_processing_settings",
973 lambda: _supervisor_processing_settings("realtime"),
974 )
975 monkeypatch.setattr(mod.time, "time", lambda: 10_000.0)
976 mod._flush_state.update(
977 {
978 "last_segment_ts": 1.0,
979 "day": "20260527",
980 "segment": "120000_300",
981 "stream": None,
982 "flushed": False,
983 }
984 )
985
986 mod._check_segment_flush()
987
988 assert len(capture.submissions) == 1
989 assert "--flush" in capture.submissions[0]["cmd"]
990 assert mod._flush_state["flushed"] is True
991
992
993def test_task_queue_daily_and_segment_run_independently(monkeypatch):
994 mod = importlib.import_module("solstone.think.supervisor")
995 queue = _fresh_task_queue(mod)
996 _capture_thread_starts(monkeypatch, mod)
997
998 queue.submit(["journal", "think", "--day", "20260527"], ref="daily-ref")
999 queue.submit(
1000 ["journal", "think", "--day", "20260527", "--segment", "120000_300"],
1001 ref="segment-ref",
1002 )
1003
1004 assert set(queue._running) == {"daily", "segment"}
1005 assert queue._queues == {}
1006
1007
1008def test_task_queue_segment_and_flush_run_independently(monkeypatch):
1009 mod = importlib.import_module("solstone.think.supervisor")
1010 queue = _fresh_task_queue(mod)
1011 _capture_thread_starts(monkeypatch, mod)
1012
1013 queue.submit(
1014 ["journal", "think", "--day", "20260527", "--segment", "120000_300"],
1015 ref="segment-ref",
1016 )
1017 queue.submit(
1018 [
1019 "journal",
1020 "think",
1021 "--day",
1022 "20260527",
1023 "--segment",
1024 "120000_300",
1025 "--flush",
1026 ],
1027 ref="flush-ref",
1028 )
1029
1030 assert set(queue._running) == {"segment", "flush"}
1031 assert queue._queues == {}
1032
1033
1034def test_task_queue_daily_and_activity_run_independently(monkeypatch):
1035 mod = importlib.import_module("solstone.think.supervisor")
1036 queue = _fresh_task_queue(mod)
1037 _capture_thread_starts(monkeypatch, mod)
1038
1039 queue.submit(["journal", "think", "--day", "20260527"], ref="daily-ref")
1040 queue.submit(
1041 [
1042 "journal",
1043 "think",
1044 "--activity",
1045 "activity-id",
1046 "--facet",
1047 "work",
1048 "--day",
1049 "20260527",
1050 ],
1051 ref="activity-ref",
1052 )
1053
1054 assert set(queue._running) == {"daily", "activity"}
1055 assert queue._queues == {}
1056
1057
1058def test_task_queue_daily_and_weekly_run_independently(monkeypatch):
1059 mod = importlib.import_module("solstone.think.supervisor")
1060 queue = _fresh_task_queue(mod)
1061 _capture_thread_starts(monkeypatch, mod)
1062
1063 queue.submit(["journal", "think", "--day", "20260527"], ref="daily-ref")
1064 queue.submit(["journal", "think", "--weekly", "-v"], ref="weekly-ref")
1065
1066 assert set(queue._running) == {"daily", "weekly"}
1067 assert queue._queues == {}
1068
1069
1070def test_task_queue_segments_plural_shares_segment_partition(monkeypatch):
1071 mod = importlib.import_module("solstone.think.supervisor")
1072 queue = _fresh_task_queue(mod)
1073 _capture_thread_starts(monkeypatch, mod)
1074
1075 queue.submit(
1076 ["journal", "think", "--day", "20260527", "--segment", "120000_300"],
1077 ref="segment-ref",
1078 )
1079 queue.submit(
1080 ["journal", "think", "--day", "20260527", "--segments"],
1081 ref="segments-ref",
1082 )
1083
1084 assert set(queue._running) == {"segment"}
1085 assert queue._queues["segment"][0]["refs"] == ["segments-ref"]
1086 assert queue._queues["segment"][0]["cmd"] == [
1087 "journal",
1088 "think",
1089 "--day",
1090 "20260527",
1091 "--segments",
1092 ]
1093
1094
1095def test_task_queue_flush_flag_precedes_segment_flag():
1096 runner = importlib.import_module("solstone.think.runner")
1097
1098 assert (
1099 runner._command_partition(
1100 ["journal", "think", "--day", "20260527", "--segment", "00", "--flush"]
1101 )
1102 == "flush"
1103 )
1104
1105
1106def test_task_queue_within_mode_serialization_segment(monkeypatch):
1107 mod = importlib.import_module("solstone.think.supervisor")
1108 queue = _fresh_task_queue(mod)
1109 spawned = _capture_thread_starts(monkeypatch, mod)
1110
1111 queue.submit(
1112 ["journal", "think", "--day", "20260527", "--segment", "120000_300"],
1113 ref="first-ref",
1114 )
1115 queue.submit(
1116 ["journal", "think", "--day", "20260527", "--segment", "120500_300"],
1117 ref="second-ref",
1118 )
1119
1120 assert set(queue._running) == {"segment"}
1121 assert queue._running["segment"]["ref"] == "first-ref"
1122 assert set(queue._queues) == {"segment"}
1123 assert queue._queues["segment"][0]["refs"] == ["second-ref"]
1124
1125 queue._process_next("segment")
1126
1127 assert queue._running["segment"]["ref"] == "second-ref"
1128 assert queue._queues["segment"] == []
1129 assert spawned[-1][0] == ["second-ref"]
1130
1131
1132def test_task_queue_dedup_within_segment_partition(monkeypatch):
1133 mod = importlib.import_module("solstone.think.supervisor")
1134 queue = _fresh_task_queue(mod)
1135 _capture_thread_starts(monkeypatch, mod)
1136 cmd = ["journal", "think", "--day", "20260527", "--segment", "120000_300"]
1137
1138 queue.submit(cmd, ref="first-ref")
1139 queue.submit(cmd, ref="second-ref")
1140 queue.submit(cmd, ref="third-ref")
1141
1142 assert queue._running["segment"]["ref"] == "first-ref"
1143 assert len(queue._queues["segment"]) == 1
1144 assert queue._queues["segment"][0]["cmd"] == cmd
1145 assert queue._queues["segment"][0]["refs"] == ["second-ref", "third-ref"]
1146
1147
1148def test_task_queue_stale_thread_reclamation_per_partition(monkeypatch):
1149 mod = importlib.import_module("solstone.think.supervisor")
1150 dead_thread = threading.Thread(target=lambda: None)
1151 dead_thread.start()
1152 dead_thread.join()
1153
1154 queue = _fresh_task_queue(mod)
1155 _capture_thread_starts(monkeypatch, mod)
1156
1157 queue.submit(["journal", "think", "--day", "20260527"], ref="old-daily-ref")
1158 queue.submit(
1159 ["journal", "think", "--day", "20260527", "--segment", "120000_300"],
1160 ref="segment-ref",
1161 )
1162
1163 queue._running["daily"]["thread"] = dead_thread
1164
1165 queue.submit(["journal", "think", "--day", "20260528"], ref="new-daily-ref")
1166
1167 assert queue._running["daily"]["ref"] == "new-daily-ref"
1168 assert queue._running["segment"]["ref"] == "segment-ref"
1169 assert set(queue._running) == {"daily", "segment"}
1170
1171
1172def test_handle_task_request_routes_to_mode_partition(monkeypatch):
1173 mod = importlib.import_module("solstone.think.supervisor")
1174 queue = _fresh_task_queue(mod)
1175 _capture_thread_starts(monkeypatch, mod)
1176
1177 queue.submit(["journal", "think", "--day", "20260527"], ref="daily-ref")
1178 mod._handle_task_request(
1179 {
1180 "tract": "supervisor",
1181 "event": "request",
1182 "cmd": ["journal", "think", "--day", "20260527", "--segment", "120000_300"],
1183 "ref": "segment-ref",
1184 }
1185 )
1186
1187 assert queue._running["daily"]["ref"] == "daily-ref"
1188 assert queue._running["segment"]["ref"] == "segment-ref"
1189 assert queue._queues == {}
1190
1191
1192def test_scheduler_weekly_cap_registers_under_weekly(monkeypatch):
1193 mod = importlib.import_module("solstone.think.supervisor")
1194 queue = mod.TaskQueue(on_queue_change=None)
1195 monkeypatch.setattr(
1196 mod.scheduler,
1197 "collect_runtime_caps",
1198 lambda: [(["journal", "think", "--weekly", "-v"], 60.0)],
1199 )
1200
1201 for cmd, seconds in mod.scheduler.collect_runtime_caps():
1202 queue.set_cap(mod.TaskQueue.get_command_name(cmd), seconds)
1203
1204 assert queue._caps == {"weekly": 60.0}
1205
1206
1207def test_reactive_task_caps_values():
1208 mod = importlib.import_module("solstone.think.supervisor")
1209
1210 assert mod.REACTIVE_TASK_CAPS == {
1211 "daily": 21600,
1212 "segment": 4500,
1213 "indexer": 7200,
1214 "importer": 3600,
1215 }
1216
1217
1218def test_register_baseline_caps_sets_explicit_caps():
1219 mod = importlib.import_module("solstone.think.supervisor")
1220 queue = mod.TaskQueue(on_queue_change=None)
1221
1222 mod.register_baseline_caps(queue)
1223
1224 backup_partition = mod.TaskQueue.get_command_name(
1225 ["journal", "maintenance", "run", "backup:run"]
1226 )
1227 expected = {
1228 "daily": 21600,
1229 "segment": 4500,
1230 "indexer": 7200,
1231 "importer": 3600,
1232 backup_partition: 49 * 60 * 60,
1233 }
1234 for name, seconds in expected.items():
1235 assert queue._effective_cap(name) == seconds
1236 assert queue._effective_cap(name) != mod.DEFAULT_TASK_MAX_RUNTIME
1237
1238
1239def test_from_scratch_reprocess_resolves_to_daily():
1240 mod = importlib.import_module("solstone.think.supervisor")
1241
1242 assert (
1243 mod.TaskQueue.get_command_name(
1244 ["journal", "think", "-v", "--day", "20260527", "--from-scratch"]
1245 )
1246 == "daily"
1247 )
1248
1249
1250def test_queue_event_carries_mode_partition_name(monkeypatch):
1251 mod = importlib.import_module("solstone.think.supervisor")
1252 events = []
1253 queue = _fresh_task_queue(
1254 mod,
1255 on_queue_change=lambda command, running, queued: events.append(
1256 (command, running, queued)
1257 ),
1258 )
1259 _capture_thread_starts(monkeypatch, mod)
1260
1261 queue.submit(
1262 ["journal", "think", "--day", "20260527", "--segment", "120000_300"],
1263 ref="first-ref",
1264 )
1265 queue.submit(
1266 ["journal", "think", "--day", "20260527", "--segment", "120500_300"],
1267 ref="second-ref",
1268 )
1269
1270 assert events[-1][0] == "segment"
1271
1272
1273def test_no_literal_think_queue_keys_in_source():
1274 mod = importlib.import_module("solstone.think.supervisor")
1275
1276 assert mod.TaskQueue.get_command_name(
1277 ["journal", "think", "--day", "20260527"]
1278 ) != ("think")
1279
1280
1281def test_task_queue_same_command_queued(monkeypatch):
1282 """Test that same command is queued when already running."""
1283 mod = importlib.import_module("solstone.think.supervisor")
1284
1285 # Create fresh task queue (no callback to avoid callosum events)
1286 mod._task_queue = mod.TaskQueue(on_queue_change=None)
1287
1288 spawned = _capture_thread_starts(monkeypatch, mod)
1289
1290 # First request - should run immediately
1291 msg1 = {
1292 "tract": "supervisor",
1293 "event": "request",
1294 "cmd": ["journal", "indexer", "--rescan"],
1295 }
1296 mod._handle_task_request(msg1)
1297
1298 assert "indexer" in mod._task_queue._running
1299 assert len(spawned) == 1
1300
1301 # Second request (different args) - should be queued
1302 msg2 = {
1303 "tract": "supervisor",
1304 "event": "request",
1305 "cmd": ["journal", "indexer", "--rescan-full"],
1306 }
1307 mod._handle_task_request(msg2)
1308
1309 assert len(spawned) == 1 # No new spawn
1310 assert "indexer" in mod._task_queue._queues
1311 assert len(mod._task_queue._queues["indexer"]) == 1
1312 # Queue entries are {refs, cmd} dicts (refs is a list for coalescing)
1313 assert mod._task_queue._queues["indexer"][0]["cmd"] == [
1314 "journal",
1315 "indexer",
1316 "--rescan-full",
1317 ]
1318 assert len(mod._task_queue._queues["indexer"][0]["refs"]) == 1
1319
1320
1321def test_task_queue_dedupe_exact_match(monkeypatch):
1322 """Test that exact same command is deduped in queue."""
1323 mod = importlib.import_module("solstone.think.supervisor")
1324
1325 # Create fresh task queue (no callback to avoid callosum events)
1326 mod._task_queue = mod.TaskQueue(on_queue_change=None)
1327
1328 _capture_thread_starts(monkeypatch, mod)
1329
1330 # First request - runs
1331 msg1 = {
1332 "tract": "supervisor",
1333 "event": "request",
1334 "cmd": ["journal", "indexer", "--rescan"],
1335 }
1336 mod._handle_task_request(msg1)
1337
1338 # Second request (same cmd) - queued
1339 msg2 = {
1340 "tract": "supervisor",
1341 "event": "request",
1342 "cmd": ["journal", "indexer", "--rescan"],
1343 }
1344 mod._handle_task_request(msg2)
1345
1346 assert len(mod._task_queue._queues["indexer"]) == 1
1347
1348 # Third request (same cmd again) - deduped, not added
1349 msg3 = {
1350 "tract": "supervisor",
1351 "event": "request",
1352 "cmd": ["journal", "indexer", "--rescan"],
1353 }
1354 mod._handle_task_request(msg3)
1355
1356 assert len(mod._task_queue._queues["indexer"]) == 1 # Still just 1
1357
1358
1359def test_task_queue_different_commands_independent(monkeypatch):
1360 """Test that different commands have independent queues."""
1361 mod = importlib.import_module("solstone.think.supervisor")
1362
1363 # Create fresh task queue (no callback to avoid callosum events)
1364 mod._task_queue = mod.TaskQueue(on_queue_change=None)
1365
1366 spawned = _capture_thread_starts(monkeypatch, mod)
1367
1368 # Indexer request - runs
1369 msg1 = {
1370 "tract": "supervisor",
1371 "event": "request",
1372 "cmd": ["journal", "indexer", "--rescan"],
1373 }
1374 mod._handle_task_request(msg1)
1375
1376 # Insight request - also runs (different command)
1377 msg2 = {
1378 "tract": "supervisor",
1379 "event": "request",
1380 "cmd": ["sol", "insight", "20240101"],
1381 }
1382 mod._handle_task_request(msg2)
1383
1384 assert len(spawned) == 2 # Both spawned
1385 assert "indexer" in mod._task_queue._running
1386 assert "insight" in mod._task_queue._running
1387
1388
1389def test_process_queue_spawns_next(monkeypatch):
1390 """Test that _process_next spawns next queued task."""
1391 mod = importlib.import_module("solstone.think.supervisor")
1392
1393 # Create task queue with pre-set state
1394 mod._task_queue = mod.TaskQueue(on_queue_change=None)
1395 mod._task_queue._running = {"indexer": {"ref": "ref123", "thread": None}}
1396 mod._task_queue._queues = {
1397 "indexer": [
1398 {"refs": ["queued-ref"], "cmd": ["journal", "indexer", "--rescan-full"]}
1399 ]
1400 }
1401
1402 spawned = _capture_thread_starts(monkeypatch, mod)
1403
1404 # Process queue
1405 mod._task_queue._process_next("indexer")
1406
1407 # Should have spawned the queued task with its refs list
1408 assert len(spawned) == 1
1409 assert spawned[0][0] == ["queued-ref"] # refs list preserved from queue
1410 assert spawned[0][1] == ["journal", "indexer", "--rescan-full"] # cmd
1411 assert spawned[0][2] == "indexer" # cmd_name
1412
1413 # Queue should be empty now
1414 assert mod._task_queue._queues["indexer"] == []
1415
1416
1417def test_process_queue_clears_running_when_empty(monkeypatch):
1418 """Test that _process_next clears running state when queue is empty."""
1419 mod = importlib.import_module("solstone.think.supervisor")
1420
1421 # Create task queue with pre-set state (no queued tasks)
1422 mod._task_queue = mod.TaskQueue(on_queue_change=None)
1423 mod._task_queue._running = {"indexer": {"ref": "ref123", "thread": None}}
1424 mod._task_queue._queues = {"indexer": []}
1425
1426 spawned = _capture_thread_starts(monkeypatch, mod)
1427
1428 # Process queue
1429 mod._task_queue._process_next("indexer")
1430
1431 # No spawn (queue was empty)
1432 assert len(spawned) == 0
1433
1434 # Running state should be cleared
1435 assert "indexer" not in mod._task_queue._running
1436
1437
1438def test_task_request_uses_caller_provided_ref(monkeypatch):
1439 """Test that caller-provided ref is used and preserved through queue."""
1440 mod = importlib.import_module("solstone.think.supervisor")
1441
1442 # Create fresh task queue (no callback to avoid callosum events)
1443 mod._task_queue = mod.TaskQueue(on_queue_change=None)
1444
1445 spawned = _capture_thread_starts(monkeypatch, mod)
1446
1447 # Request with caller-provided ref
1448 msg = {
1449 "tract": "supervisor",
1450 "event": "request",
1451 "cmd": ["journal", "indexer", "--rescan"],
1452 "ref": "my-custom-ref-123",
1453 }
1454 mod._handle_task_request(msg)
1455
1456 # Should use the provided ref
1457 assert mod._task_queue._running["indexer"]["ref"] == "my-custom-ref-123"
1458 assert spawned[0][0] == ["my-custom-ref-123"] # refs is a list
1459
1460
1461def test_task_queue_preserves_caller_ref(monkeypatch):
1462 """Test that queued requests preserve their caller-provided ref."""
1463 mod = importlib.import_module("solstone.think.supervisor")
1464
1465 # Create fresh task queue (no callback to avoid callosum events)
1466 mod._task_queue = mod.TaskQueue(on_queue_change=None)
1467
1468 _capture_thread_starts(monkeypatch, mod)
1469
1470 # First request runs immediately
1471 msg1 = {
1472 "tract": "supervisor",
1473 "event": "request",
1474 "cmd": ["journal", "indexer", "--rescan"],
1475 "ref": "first-ref",
1476 }
1477 mod._handle_task_request(msg1)
1478
1479 # Second request gets queued with its own ref
1480 msg2 = {
1481 "tract": "supervisor",
1482 "event": "request",
1483 "cmd": ["journal", "indexer", "--rescan-full"],
1484 "ref": "second-ref",
1485 }
1486 mod._handle_task_request(msg2)
1487
1488 # Verify queued entry has the caller's ref in refs list
1489 assert len(mod._task_queue._queues["indexer"]) == 1
1490 assert mod._task_queue._queues["indexer"][0]["refs"] == ["second-ref"]
1491 assert mod._task_queue._queues["indexer"][0]["cmd"] == [
1492 "journal",
1493 "indexer",
1494 "--rescan-full",
1495 ]
1496
1497
1498def test_task_queue_coalesces_refs_on_dedupe(monkeypatch):
1499 """Test that duplicate queued requests coalesce their refs."""
1500 mod = importlib.import_module("solstone.think.supervisor")
1501
1502 # Create fresh task queue (no callback to avoid callosum events)
1503 mod._task_queue = mod.TaskQueue(on_queue_change=None)
1504
1505 _capture_thread_starts(monkeypatch, mod)
1506
1507 # First request runs immediately
1508 msg1 = {
1509 "tract": "supervisor",
1510 "event": "request",
1511 "cmd": ["journal", "indexer", "--rescan"],
1512 "ref": "first-ref",
1513 }
1514 mod._handle_task_request(msg1)
1515
1516 # Second request (same cmd) gets queued
1517 msg2 = {
1518 "tract": "supervisor",
1519 "event": "request",
1520 "cmd": ["journal", "indexer", "--rescan"],
1521 "ref": "second-ref",
1522 }
1523 mod._handle_task_request(msg2)
1524
1525 # Third request (same cmd) should coalesce its ref into existing queue entry
1526 msg3 = {
1527 "tract": "supervisor",
1528 "event": "request",
1529 "cmd": ["journal", "indexer", "--rescan"],
1530 "ref": "third-ref",
1531 }
1532 mod._handle_task_request(msg3)
1533
1534 # Should still be just one queue entry
1535 assert len(mod._task_queue._queues["indexer"]) == 1
1536 # But it should have both refs
1537 assert mod._task_queue._queues["indexer"][0]["refs"] == [
1538 "second-ref",
1539 "third-ref",
1540 ]
1541
1542
1543def test_process_queue_spawns_with_multiple_refs(monkeypatch):
1544 """Test that dequeued task has all coalesced refs."""
1545 mod = importlib.import_module("solstone.think.supervisor")
1546
1547 # Create task queue with pre-set state (queued task with multiple refs)
1548 mod._task_queue = mod.TaskQueue(on_queue_change=None)
1549 mod._task_queue._running = {"indexer": {"ref": "running-ref", "thread": None}}
1550 mod._task_queue._queues = {
1551 "indexer": [
1552 {
1553 "refs": ["ref-A", "ref-B", "ref-C"],
1554 "cmd": ["journal", "indexer", "--rescan"],
1555 }
1556 ]
1557 }
1558
1559 spawned = _capture_thread_starts(monkeypatch, mod)
1560
1561 # Process queue
1562 mod._task_queue._process_next("indexer")
1563
1564 # Should spawn with all three refs
1565 assert len(spawned) == 1
1566 assert spawned[0][0] == ["ref-A", "ref-B", "ref-C"] # all refs passed
1567 assert spawned[0][1] == ["journal", "indexer", "--rescan"]
1568
1569
1570def test_stale_queue_detected_on_submit(monkeypatch):
1571 """Test that a dead task thread is detected and cleared on next submit."""
1572 import threading
1573
1574 mod = importlib.import_module("solstone.think.supervisor")
1575
1576 mod._task_queue = mod.TaskQueue(on_queue_change=None)
1577
1578 # Create a dead thread BEFORE monkeypatching Thread.start
1579 dead_thread = threading.Thread(target=lambda: None)
1580 dead_thread.start()
1581 dead_thread.join()
1582 assert not dead_thread.is_alive()
1583
1584 spawned = _capture_thread_starts(monkeypatch, mod)
1585
1586 mod._task_queue._running = {"indexer": {"ref": "stale-ref", "thread": dead_thread}}
1587 mod._task_queue._queues = {
1588 "indexer": [
1589 {"refs": ["queued-ref"], "cmd": ["journal", "indexer", "--rescan-full"]}
1590 ]
1591 }
1592
1593 # Submit a new indexer task — should detect stale state and start immediately
1594 msg = {
1595 "tract": "supervisor",
1596 "event": "request",
1597 "cmd": ["journal", "indexer", "--rescan-new"],
1598 "ref": "new-ref",
1599 }
1600 mod._handle_task_request(msg)
1601
1602 # Stale entry should have been cleared, new task started
1603 assert mod._task_queue._running["indexer"]["ref"] == "new-ref"
1604 assert len(spawned) == 1
1605
1606 # Old queued entries should still be in queue (stale clear only removes _running)
1607 assert len(mod._task_queue._queues["indexer"]) == 1
1608
1609
1610class _TaskProcessStub:
1611 def __init__(self):
1612 self.poll = MagicMock(return_value=None)
1613 self.pid = 12345
1614
1615
1616class _TaskManagedStub:
1617 def __init__(self, *, cmd, start_time=100.0):
1618 self.name = "task"
1619 self.cmd = cmd
1620 self.start_time = start_time
1621 self.process = _TaskProcessStub()
1622 self.ref = "ref-1"
1623 self.terminate = MagicMock()
1624 self.cleanup = MagicMock()
1625 self.is_running = MagicMock(return_value=True)
1626
1627
1628BENIGN_LLAMA_LOAD_LOG = (
1629 "2026-06-12T12:00:00+00:00 [llama-server:stderr] "
1630 "llama_model_loader: loading model tensors\n"
1631 "2026-06-12T12:00:02+00:00 [llama-server:stderr] "
1632 "common_init_from_params: setting dry_penalty_last_n to ctx_size = 16384\n"
1633)
1634
1635
1636def _local_artifacts(local_install, binary, gguf, mmproj, *, backend="vulkan"):
1637 return local_install.LocalArtifacts(
1638 backend=backend,
1639 backend_reason=f"test {backend}",
1640 binary_path=binary,
1641 lib_dir=None,
1642 gguf_path=gguf,
1643 mmproj_path=mmproj,
1644 )
1645
1646
1647def _cuda_local_artifacts(local_install, binary, lib_dir, gguf, mmproj):
1648 return local_install.LocalArtifacts(
1649 backend="cuda",
1650 backend_reason="test cuda",
1651 binary_path=binary,
1652 lib_dir=lib_dir,
1653 gguf_path=gguf,
1654 mmproj_path=mmproj,
1655 )
1656
1657
1658class _FakeReservation:
1659 def __init__(self, port: int = 2468):
1660 self.port = port
1661 self.released = False
1662 self.closed = False
1663
1664 def release_for_spawn(self) -> int:
1665 self.released = True
1666 self.closed = True
1667 return self.port
1668
1669 def close(self) -> None:
1670 self.closed = True
1671
1672
1673def _mlx_launch_plan(mod, runtime_dir: Path, *, model_id: str = "mlx-model"):
1674 return mod.LocalServerLaunchPlan(
1675 backend="mlx",
1676 desired_fingerprint_json='{"provider":"local"}',
1677 desired_fingerprint_sha256="fp-local",
1678 runtime_dir=runtime_dir,
1679 model_id=model_id,
1680 )
1681
1682
1683def _vulkan_launch_plan(
1684 mod,
1685 binary: Path,
1686 gguf: Path,
1687 mmproj: Path | None,
1688 *,
1689 vram_mib: int = 6390,
1690):
1691 from solstone.think.providers import local_server
1692
1693 tier = local_server.select_server_tier(vram_mib)
1694 return mod.LocalServerLaunchPlan(
1695 backend="vulkan",
1696 desired_fingerprint_json='{"provider":"local"}',
1697 desired_fingerprint_sha256="fp-local",
1698 binary_path=binary,
1699 model_path=gguf,
1700 mmproj_path=mmproj,
1701 gpu_index=1,
1702 gpu_name="NVIDIA GeForce GTX 1660 Ti",
1703 gpu_vram_mib=vram_mib,
1704 vram_before_mib=512,
1705 context_tokens=tier.context_tokens,
1706 parallel_slots=tier.parallel_slots,
1707 prompt_cache_mib=tier.prompt_cache_mib,
1708 visible_devices_env="GGML_VK_VISIBLE_DEVICES",
1709 visible_devices_value="1",
1710 env_updates={"GGML_VK_VISIBLE_DEVICES": "1"},
1711 backend_reason="test vulkan",
1712 )
1713
1714
1715def _cuda_launch_plan(
1716 mod,
1717 binary: Path,
1718 gguf: Path,
1719 mmproj: Path,
1720 lib_dir: Path,
1721 *,
1722 tiering_memory_mib: int | None = 20000,
1723 visible_device: str = "0",
1724):
1725 from solstone.think.providers import local_server
1726
1727 tier = local_server.select_server_tier(tiering_memory_mib or 0)
1728 existing_ld = os.environ.get("LD_LIBRARY_PATH", "")
1729 ld_library_path = f"{lib_dir}:{existing_ld}" if existing_ld else str(lib_dir)
1730 return mod.LocalServerLaunchPlan(
1731 backend="cuda",
1732 desired_fingerprint_json='{"provider":"local"}',
1733 desired_fingerprint_sha256="fp-local",
1734 binary_path=binary,
1735 model_path=gguf,
1736 mmproj_path=mmproj,
1737 lib_dir=lib_dir,
1738 gpu_index=int(visible_device),
1739 gpu_vram_mib=tiering_memory_mib,
1740 context_tokens=tier.context_tokens,
1741 parallel_slots=tier.parallel_slots,
1742 prompt_cache_mib=tier.prompt_cache_mib,
1743 visible_devices_env="CUDA_VISIBLE_DEVICES",
1744 visible_devices_value=visible_device,
1745 env_updates={
1746 "CUDA_VISIBLE_DEVICES": visible_device,
1747 "LD_LIBRARY_PATH": ld_library_path,
1748 },
1749 backend_reason="test cuda",
1750 )
1751
1752
1753def test_ensure_venv_bin_on_path_prepends_when_missing(monkeypatch):
1754 mod = importlib.import_module("solstone.think.supervisor")
1755 monkeypatch.setenv("PATH", "/usr/bin")
1756 monkeypatch.setattr(sys, "executable", "/fake/venv/bin/python3")
1757
1758 mod._ensure_venv_bin_on_path()
1759
1760 parts = os.environ["PATH"].split(os.pathsep)
1761 assert parts[0] == "/fake/venv/bin"
1762 assert "/usr/bin" in parts[1:]
1763
1764
1765def test_ensure_venv_bin_on_path_idempotent(monkeypatch):
1766 mod = importlib.import_module("solstone.think.supervisor")
1767 monkeypatch.setenv("PATH", "/usr/bin")
1768 monkeypatch.setattr(sys, "executable", "/fake/venv/bin/python3")
1769
1770 mod._ensure_venv_bin_on_path()
1771 mod._ensure_venv_bin_on_path()
1772
1773 parts = os.environ["PATH"].split(os.pathsep)
1774 assert parts.count("/fake/venv/bin") == 1
1775
1776
1777def test_taskqueue_set_cap_records_cap():
1778 mod = importlib.import_module("solstone.think.supervisor")
1779 queue = mod.TaskQueue(on_queue_change=None)
1780
1781 queue.set_cap("import", 1800)
1782
1783 assert queue._caps["import"] == 1800
1784
1785
1786def test_task_queue_history_records_completion(tmp_path, monkeypatch):
1787 mod = importlib.import_module("solstone.think.supervisor")
1788 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
1789 (tmp_path / "health").mkdir(parents=True, exist_ok=True)
1790
1791 queue = mod.TaskQueue(on_queue_change=None)
1792 callosum = MagicMock()
1793
1794 class FakeCallosum:
1795 def start(self, callback=None):
1796 return None
1797
1798 def emit(self, *args, **kwargs):
1799 return callosum.emit(*args, **kwargs)
1800
1801 def stop(self):
1802 return None
1803
1804 managed = MagicMock()
1805 managed.pid = 12345
1806 managed.wait.return_value = 0
1807 managed.cleanup = MagicMock()
1808
1809 def fake_spawn(cmd, *, ref=None, callosum=None, day=None, env=None):
1810 managed.cmd = cmd
1811 managed.ref = ref
1812 return managed
1813
1814 monkeypatch.setattr(mod, "CallosumConnection", FakeCallosum)
1815 monkeypatch.setattr(mod.RunnerManagedProcess, "spawn", fake_spawn)
1816
1817 queue._run_task(
1818 ["ref-1"],
1819 ["journal", "heartbeat"],
1820 "heartbeat",
1821 None,
1822 "heartbeat",
1823 )
1824
1825 assert list(queue._history) == [
1826 {
1827 "name": "heartbeat",
1828 "cmd": ["journal", "heartbeat"],
1829 "ref": "ref-1",
1830 "ended_at": queue._history[0]["ended_at"],
1831 "exit_status": "ok",
1832 "scheduler_name": "heartbeat",
1833 }
1834 ]
1835
1836
1837def test_scheduler_completion_updates_scheduler_json(tmp_path, monkeypatch):
1838 mod = importlib.import_module("solstone.think.supervisor")
1839 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
1840 health_dir = tmp_path / "health"
1841 health_dir.mkdir(parents=True, exist_ok=True)
1842 state_path = health_dir / "scheduler.json"
1843 state_path.write_text(
1844 '{"heartbeat": {"custom": "kept"}, "other": {"last_run": 1}}',
1845 encoding="utf-8",
1846 )
1847
1848 mod._record_scheduler_completion(
1849 "heartbeat",
1850 ended_at=123.0,
1851 exit_status="ok",
1852 ref="ref-1",
1853 cmd=["journal", "heartbeat"],
1854 )
1855
1856 data = json.loads(state_path.read_text(encoding="utf-8"))
1857 assert data["heartbeat"] == {
1858 "custom": "kept",
1859 "last_run": 123.0,
1860 "last_status": "ok",
1861 "last_ref": "ref-1",
1862 }
1863 assert data["other"] == {"last_run": 1}
1864
1865
1866def test_exit_status_for_code_maps_empty_sentinel():
1867 from solstone.think.supervisor import _exit_status_for_code
1868 from solstone.think.utils import EXIT_EMPTY
1869
1870 assert _exit_status_for_code(0) == "ok"
1871 assert _exit_status_for_code(EXIT_EMPTY) == "empty"
1872 assert _exit_status_for_code(1) == "error"
1873 assert _exit_status_for_code(75) == "error"
1874
1875
1876def test_run_task_completes_when_scheduler_writeback_fails(monkeypatch):
1877 mod = importlib.import_module("solstone.think.supervisor")
1878 queue = mod.TaskQueue(on_queue_change=None)
1879 callosum = MagicMock()
1880
1881 class FakeCallosum:
1882 def start(self, callback=None):
1883 return None
1884
1885 def emit(self, *args, **kwargs):
1886 return callosum.emit(*args, **kwargs)
1887
1888 def stop(self):
1889 return callosum.stop()
1890
1891 managed = MagicMock()
1892 managed.pid = 12345
1893 managed.wait.return_value = 0
1894 managed.cleanup = MagicMock()
1895
1896 def fake_spawn(cmd, *, ref=None, callosum=None, day=None, env=None):
1897 managed.cmd = cmd
1898 managed.ref = ref
1899 return managed
1900
1901 monkeypatch.setattr(mod, "CallosumConnection", FakeCallosum)
1902 monkeypatch.setattr(mod.RunnerManagedProcess, "spawn", fake_spawn)
1903 monkeypatch.setattr(
1904 mod,
1905 "_record_scheduler_completion",
1906 MagicMock(side_effect=OSError("disk full")),
1907 )
1908 process_next = MagicMock()
1909 monkeypatch.setattr(queue, "_process_next", process_next)
1910
1911 queue._run_task(
1912 ["ref-1"],
1913 ["journal", "heartbeat"],
1914 "heartbeat",
1915 None,
1916 "heartbeat",
1917 )
1918
1919 callosum.stop.assert_called_once()
1920 process_next.assert_called_once_with("heartbeat")
1921
1922
1923def test_run_task_records_attempt_and_outcome_on_spawn(monkeypatch):
1924 mod = importlib.import_module("solstone.think.supervisor")
1925 queue = mod.TaskQueue(on_queue_change=None)
1926 callosum = MagicMock()
1927 record_attempt = MagicMock()
1928 record_outcome = MagicMock(return_value=SimpleNamespace(entered_backoff=False))
1929
1930 class FakeCallosum:
1931 def start(self, callback=None):
1932 return None
1933
1934 def emit(self, *args, **kwargs):
1935 return callosum.emit(*args, **kwargs)
1936
1937 def stop(self):
1938 return callosum.stop()
1939
1940 managed = MagicMock()
1941 managed.pid = 12345
1942 managed.wait.return_value = 0
1943 managed.cleanup = MagicMock()
1944
1945 def fake_spawn(cmd, *, ref=None, callosum=None, day=None, env=None):
1946 managed.cmd = cmd
1947 managed.ref = ref
1948 return managed
1949
1950 monkeypatch.setattr(mod, "CallosumConnection", FakeCallosum)
1951 monkeypatch.setattr(mod.RunnerManagedProcess, "spawn", fake_spawn)
1952 monkeypatch.setattr(mod, "record_attempt", record_attempt)
1953 monkeypatch.setattr(mod, "record_outcome", record_outcome)
1954 monkeypatch.setattr(queue, "_process_next", MagicMock())
1955
1956 cmd = ["journal", "think", "-v", "--day", "20250101"]
1957 queue._run_task(["ref-1"], cmd, "daily", "20250101")
1958
1959 record_attempt.assert_called_once()
1960 assert record_attempt.call_args.args == (cmd, "20250101", "ref-1")
1961 assert isinstance(record_attempt.call_args.kwargs["started_at"], float)
1962 record_outcome.assert_called_once()
1963 assert record_outcome.call_args.args == (cmd, "20250101", "ref-1")
1964 assert record_outcome.call_args.kwargs["exit_status"] == "ok"
1965 assert isinstance(record_outcome.call_args.kwargs["ended_at"], float)
1966
1967
1968def test_run_task_spawn_failure_does_not_record_attempt_or_outcome(monkeypatch):
1969 mod = importlib.import_module("solstone.think.supervisor")
1970 queue = mod.TaskQueue(on_queue_change=None)
1971 callosum = MagicMock()
1972 record_attempt = MagicMock()
1973 record_outcome = MagicMock()
1974 process_next = MagicMock()
1975
1976 class FakeCallosum:
1977 def start(self, callback=None):
1978 return None
1979
1980 def emit(self, *args, **kwargs):
1981 return callosum.emit(*args, **kwargs)
1982
1983 def stop(self):
1984 return callosum.stop()
1985
1986 def fake_spawn(cmd, *, ref=None, callosum=None, day=None, env=None):
1987 raise RuntimeError("spawn failed")
1988
1989 monkeypatch.setattr(mod, "CallosumConnection", FakeCallosum)
1990 monkeypatch.setattr(mod.RunnerManagedProcess, "spawn", fake_spawn)
1991 monkeypatch.setattr(mod, "record_attempt", record_attempt)
1992 monkeypatch.setattr(mod, "record_outcome", record_outcome)
1993 monkeypatch.setattr(queue, "_process_next", process_next)
1994
1995 cmd = ["journal", "think", "-v", "--day", "20250101"]
1996 queue._run_task(["ref-1"], cmd, "daily", "20250101")
1997
1998 record_attempt.assert_not_called()
1999 record_outcome.assert_not_called()
2000 callosum.stop.assert_called_once()
2001 process_next.assert_called_once_with("daily")
2002
2003
2004def test_submit_coalesce_does_not_record_attempt(monkeypatch):
2005 mod = importlib.import_module("solstone.think.supervisor")
2006 queue = mod.TaskQueue(on_queue_change=None)
2007 monkeypatch.setattr(mod, "record_attempt", MagicMock())
2008 _capture_thread_starts(monkeypatch, mod)
2009 cmd = ["journal", "think", "-v", "--day", "20250101"]
2010
2011 queue.submit(cmd, ref="running", day="20250101")
2012 queue.submit(cmd, ref="queued", day="20250101")
2013 queue.submit(cmd, ref="coalesced", day="20250101")
2014
2015 mod.record_attempt.assert_not_called()
2016 assert queue._queues["daily"][0]["refs"] == ["queued", "coalesced"]
2017
2018
2019def test_handle_task_request_skip_does_not_record(monkeypatch):
2020 mod = importlib.import_module("solstone.think.supervisor")
2021 queue = mod.TaskQueue(on_queue_change=None)
2022 managed = _TaskManagedStub(
2023 cmd=["journal", "importer", "--sync", "plaud"], start_time=100.0
2024 )
2025 queue._active["active-ref"] = managed
2026 queue.set_cap("importer", 50)
2027
2028 monkeypatch.setattr(mod, "_task_queue", queue)
2029 monkeypatch.setattr(mod, "_supervisor_callosum", MagicMock())
2030 monkeypatch.setattr(mod, "record_attempt", MagicMock())
2031 monkeypatch.setattr(mod.time, "time", lambda: 150.0)
2032
2033 mod._handle_task_request(
2034 {
2035 "tract": "supervisor",
2036 "event": "request",
2037 "cmd": ["journal", "importer", "--sync", "plaud"],
2038 "ref": "requested-ref",
2039 }
2040 )
2041
2042 mod.record_attempt.assert_not_called()
2043
2044
2045def test_run_task_emits_backoff_notification_once(monkeypatch):
2046 mod = importlib.import_module("solstone.think.supervisor")
2047 queue = mod.TaskQueue(on_queue_change=None)
2048 callosum = MagicMock()
2049
2050 class FakeCallosum:
2051 def start(self, callback=None):
2052 return None
2053
2054 def emit(self, *args, **kwargs):
2055 return callosum.emit(*args, **kwargs)
2056
2057 def stop(self):
2058 return callosum.stop()
2059
2060 managed = MagicMock()
2061 managed.pid = 12345
2062 managed.wait.return_value = 0
2063 managed.cleanup = MagicMock()
2064
2065 def fake_spawn(cmd, *, ref=None, callosum=None, day=None, env=None):
2066 managed.cmd = cmd
2067 managed.ref = ref
2068 return managed
2069
2070 monkeypatch.setattr(mod, "CallosumConnection", FakeCallosum)
2071 monkeypatch.setattr(mod.RunnerManagedProcess, "spawn", fake_spawn)
2072 monkeypatch.setattr(mod, "record_attempt", MagicMock())
2073 record_outcome = MagicMock(
2074 return_value=SimpleNamespace(
2075 entered_backoff=True,
2076 day="20250101",
2077 attempts=3,
2078 consecutive_non_completion=3,
2079 last_outcome="timeout",
2080 )
2081 )
2082 monkeypatch.setattr(mod, "record_outcome", record_outcome)
2083 monkeypatch.setattr(queue, "_process_next", MagicMock())
2084 cmd = ["journal", "think", "-v", "--day", "20250101"]
2085
2086 queue._run_task(["ref-1"], cmd, "daily", "20250101")
2087
2088 emitted = [call_args.args[:2] for call_args in callosum.emit.call_args_list]
2089 assert ("storage", "warning") in emitted
2090 assert ("notification", "show") in emitted
2091
2092 callosum.emit.reset_mock()
2093 record_outcome.return_value = SimpleNamespace(entered_backoff=False)
2094 queue._run_task(["ref-2"], cmd, "daily", "20250101")
2095
2096 emitted = [call_args.args[:2] for call_args in callosum.emit.call_args_list]
2097 assert ("storage", "warning") not in emitted
2098 assert ("notification", "show") not in emitted
2099
2100
2101def test_run_task_nudges_after_uncompleted_daily_only(monkeypatch):
2102 mod = importlib.import_module("solstone.think.supervisor")
2103 queue = mod.TaskQueue(on_queue_change=None)
2104 callosum = MagicMock()
2105 nudge = MagicMock()
2106
2107 class FakeCallosum:
2108 def start(self, callback=None):
2109 return None
2110
2111 def emit(self, *args, **kwargs):
2112 return callosum.emit(*args, **kwargs)
2113
2114 def stop(self):
2115 return callosum.stop()
2116
2117 managed = MagicMock()
2118 managed.pid = 12345
2119 managed.wait.return_value = 0
2120 managed.cleanup = MagicMock()
2121
2122 def fake_spawn(cmd, *, ref=None, callosum=None, day=None, env=None):
2123 managed.cmd = cmd
2124 managed.ref = ref
2125 return managed
2126
2127 monkeypatch.setattr(mod, "CallosumConnection", FakeCallosum)
2128 monkeypatch.setattr(mod.RunnerManagedProcess, "spawn", fake_spawn)
2129 monkeypatch.setattr(mod, "record_attempt", MagicMock())
2130 record_outcome = MagicMock()
2131 monkeypatch.setattr(mod, "record_outcome", record_outcome)
2132 monkeypatch.setattr(mod, "_nudge_catchup_drain", nudge)
2133 monkeypatch.setattr(queue, "_process_next", MagicMock())
2134
2135 daily_cmd = ["journal", "think", "-v", "--day", "20250101"]
2136 record_outcome.return_value = SimpleNamespace(
2137 recorded=True,
2138 command_kind=mod.KIND_DAILY_CATCHUP,
2139 completed=False,
2140 entered_backoff=False,
2141 )
2142 queue._cap_terminated.add("ref-timeout")
2143 queue._run_task(["ref-timeout"], daily_cmd, "daily", "20250101")
2144
2145 assert record_outcome.call_args.kwargs["exit_status"] == "timeout"
2146 nudge.assert_called_once_with(exclude_today=True)
2147
2148 nudge.reset_mock()
2149 record_outcome.return_value = SimpleNamespace(
2150 recorded=True,
2151 command_kind=mod.KIND_DAILY_CATCHUP,
2152 completed=True,
2153 entered_backoff=False,
2154 )
2155 queue._run_task(["ref-complete"], daily_cmd, "daily", "20250101")
2156 nudge.assert_not_called()
2157
2158 nudge.reset_mock()
2159 record_outcome.return_value = SimpleNamespace(
2160 recorded=True,
2161 command_kind="segment",
2162 completed=False,
2163 entered_backoff=False,
2164 )
2165 segment_cmd = [
2166 "journal",
2167 "think",
2168 "-v",
2169 "--day",
2170 "20250101",
2171 "--segment",
2172 "120000_300",
2173 ]
2174 queue._run_task(["ref-segment"], segment_cmd, "segment", "20250101")
2175 nudge.assert_not_called()
2176
2177
2178def test_handle_supervisor_drain_routes_exclude_today(monkeypatch):
2179 mod = importlib.import_module("solstone.think.supervisor")
2180 calls = []
2181
2182 class FakeDateTime:
2183 @staticmethod
2184 def now():
2185 return SimpleNamespace(date=lambda: date(2025, 1, 2))
2186
2187 def fake_drain(*args, **kwargs):
2188 calls.append((args, kwargs))
2189
2190 monkeypatch.setattr(mod, "datetime", FakeDateTime)
2191 monkeypatch.setattr(mod, "run_catchup_drain", fake_drain)
2192
2193 mod._handle_supervisor_drain(
2194 {
2195 "tract": "supervisor",
2196 "event": "drain",
2197 "day": "20250101",
2198 "exclude_today": True,
2199 }
2200 )
2201 assert calls == [((), {"force_days": {"20250101"}})]
2202
2203 calls.clear()
2204 mod._handle_supervisor_drain(
2205 {"tract": "supervisor", "event": "drain", "exclude_today": True}
2206 )
2207 assert calls == [((), {"exclude": {"20250102"}})]
2208
2209 calls.clear()
2210 mod._handle_supervisor_drain({"tract": "supervisor", "event": "drain"})
2211 assert calls == [((), {})]
2212
2213
2214def test_startup_catchup_drain_reconciles_before_drain(monkeypatch):
2215 mod = importlib.import_module("solstone.think.supervisor")
2216 order = []
2217
2218 def reconcile():
2219 order.append("reconcile")
2220 return []
2221
2222 def drain():
2223 order.append("drain")
2224
2225 monkeypatch.setattr(mod, "reconcile_interrupted_attempts", reconcile)
2226 monkeypatch.setattr(mod, "run_catchup_drain", drain)
2227
2228 mod._startup_catchup_drain()
2229
2230 assert order == ["reconcile", "drain"]
2231
2232
2233def test_run_catchup_drain_no_engine_submits_nothing(monkeypatch):
2234 mod = importlib.import_module("solstone.think.supervisor")
2235 capture = _CaptureTaskQueue()
2236 monkeypatch.setattr(mod, "_task_queue", capture)
2237 monkeypatch.setattr(mod, "no_thinking_engine_chosen", lambda: True)
2238 monkeypatch.setattr(
2239 mod,
2240 "updated_days",
2241 lambda **_kwargs: pytest.fail("drain should not enumerate days"),
2242 )
2243
2244 assert mod.run_catchup_drain() == []
2245 assert capture.submissions == []
2246
2247
2248def test_run_gate_tick_deferred_open_runs_catchup_drain(monkeypatch):
2249 mod = importlib.import_module("solstone.think.supervisor")
2250 drain = MagicMock()
2251 monkeypatch.setattr(
2252 mod,
2253 "load_processing_settings",
2254 lambda: _supervisor_processing_settings("deferred"),
2255 )
2256 monkeypatch.setattr(
2257 mod,
2258 "evaluate_drain_gate",
2259 lambda settings, now, reading: SimpleNamespace(open=True),
2260 )
2261 monkeypatch.setattr(mod, "run_catchup_drain", drain)
2262 mod._last_gate_tick = 0.0
2263
2264 mod._run_gate_tick(60.0)
2265
2266 drain.assert_called_once_with()
2267
2268
2269def test_run_gate_tick_deferred_closed_skips_catchup_drain(monkeypatch):
2270 mod = importlib.import_module("solstone.think.supervisor")
2271 drain = MagicMock()
2272 monkeypatch.setattr(
2273 mod,
2274 "load_processing_settings",
2275 lambda: _supervisor_processing_settings("deferred"),
2276 )
2277 monkeypatch.setattr(
2278 mod,
2279 "evaluate_drain_gate",
2280 lambda settings, now, reading: SimpleNamespace(open=False),
2281 )
2282 monkeypatch.setattr(mod, "run_catchup_drain", drain)
2283 mod._last_gate_tick = 0.0
2284
2285 mod._run_gate_tick(60.0)
2286
2287 drain.assert_not_called()
2288
2289
2290def _write_catchup_state(journal: Path, records: dict[str, dict]) -> None:
2291 """Write catchup-state.json entries keyed '<day>:<kind>'."""
2292 health = journal / "health"
2293 health.mkdir(parents=True, exist_ok=True)
2294 (health / "catchup-state.json").write_text(
2295 json.dumps({"version": 1, "entries": records}), encoding="utf-8"
2296 )
2297
2298
2299def _catchup_record(day: str, kind: str, **overrides) -> dict:
2300 record = {
2301 "day": day,
2302 "command_kind": kind,
2303 "attempts": 1,
2304 "consecutive_non_completion": 1,
2305 "last_attempt_at": 0,
2306 "last_outcome": "timeout",
2307 "next_retry_at": 0,
2308 "entered_backoff_at": None,
2309 "notified_at": None,
2310 "fingerprint": "fp",
2311 "active": None,
2312 "reason_code": None,
2313 "timeout_seconds": None,
2314 "bounded": None,
2315 }
2316 record.update(overrides)
2317 return record
2318
2319
2320def _arm_catchup_retry_tick(mod, monkeypatch, *, mode="realtime", remote=False):
2321 """Put the tick past its cadence gate with a seeded watermark."""
2322 monkeypatch.setattr(
2323 mod, "load_processing_settings", lambda: _supervisor_processing_settings(mode)
2324 )
2325 monkeypatch.setattr(mod, "_is_remote_mode", remote)
2326 mod._last_catchup_retry_tick = 0.0
2327
2328
2329def test_read_catchup_retry_expiries_uses_later_of_both_kinds(tmp_path, monkeypatch):
2330 mod = importlib.import_module("solstone.think.supervisor")
2331 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
2332 _write_catchup_state(
2333 tmp_path,
2334 {
2335 "20260101:daily-catchup": _catchup_record(
2336 "20260101", "daily-catchup", next_retry_at=100.0
2337 ),
2338 "20260101:segment-repair": _catchup_record(
2339 "20260101", "segment-repair", next_retry_at=250.0
2340 ),
2341 },
2342 )
2343
2344 assert mod._read_catchup_retry_expiries() == [250.0]
2345
2346
2347def test_read_catchup_retry_expiries_skips_active_and_unretried(tmp_path, monkeypatch):
2348 mod = importlib.import_module("solstone.think.supervisor")
2349 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
2350 _write_catchup_state(
2351 tmp_path,
2352 {
2353 # Active on one kind: cannot become eligible by expiry alone.
2354 "20260101:daily-catchup": _catchup_record(
2355 "20260101", "daily-catchup", next_retry_at=100.0, active="pid-1"
2356 ),
2357 "20260101:segment-repair": _catchup_record(
2358 "20260101", "segment-repair", next_retry_at=250.0
2359 ),
2360 # No backoff: already eligible, so no future crossing to detect.
2361 "20260102:daily-catchup": _catchup_record("20260102", "daily-catchup"),
2362 # KIND_SEGMENT never carries a meaningful retry and is not gated on.
2363 "20260103:segment": _catchup_record(
2364 "20260103", "segment", next_retry_at=400.0
2365 ),
2366 },
2367 )
2368
2369 assert mod._read_catchup_retry_expiries() == []
2370
2371
2372def test_catchup_retry_tick_seeds_watermark_without_draining(tmp_path, monkeypatch):
2373 mod = importlib.import_module("solstone.think.supervisor")
2374 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
2375 drain = MagicMock()
2376 monkeypatch.setattr(mod, "run_catchup_drain", drain)
2377 _arm_catchup_retry_tick(mod, monkeypatch)
2378 mod._catchup_retry_watermark = 0.0
2379
2380 now = time.time()
2381 _write_catchup_state(
2382 tmp_path,
2383 {
2384 "20260101:daily-catchup": _catchup_record(
2385 "20260101", "daily-catchup", next_retry_at=now - 10
2386 )
2387 },
2388 )
2389
2390 mod._run_catchup_retry_tick(now)
2391
2392 # _startup_catchup_drain() already covered days expired before boot.
2393 drain.assert_not_called()
2394 assert mod._catchup_retry_watermark == now
2395
2396
2397def test_catchup_retry_tick_drains_excluding_today(tmp_path, monkeypatch):
2398 mod = importlib.import_module("solstone.think.supervisor")
2399 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
2400 drain = MagicMock()
2401 monkeypatch.setattr(mod, "run_catchup_drain", drain)
2402 _arm_catchup_retry_tick(mod, monkeypatch)
2403
2404 now = time.time()
2405 mod._catchup_retry_watermark = now - 60
2406 _write_catchup_state(
2407 tmp_path,
2408 {
2409 "20260101:daily-catchup": _catchup_record(
2410 "20260101", "daily-catchup", next_retry_at=now - 10
2411 )
2412 },
2413 )
2414
2415 mod._run_catchup_retry_tick(now)
2416
2417 today_str = date.today().strftime("%Y%m%d")
2418 drain.assert_called_once_with(exclude={today_str})
2419
2420
2421def test_catchup_retry_tick_fires_once_per_expiry(tmp_path, monkeypatch):
2422 """A day that re-enters backoff waits for its new expiry, not the next tick."""
2423 mod = importlib.import_module("solstone.think.supervisor")
2424 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
2425 drain = MagicMock()
2426 monkeypatch.setattr(mod, "run_catchup_drain", drain)
2427 _arm_catchup_retry_tick(mod, monkeypatch)
2428
2429 now = time.time()
2430 mod._catchup_retry_watermark = now - 60
2431 _write_catchup_state(
2432 tmp_path,
2433 {
2434 "20260101:daily-catchup": _catchup_record(
2435 "20260101", "daily-catchup", next_retry_at=now - 10
2436 )
2437 },
2438 )
2439
2440 mod._run_catchup_retry_tick(now)
2441 mod._last_catchup_retry_tick = 0.0
2442 mod._run_catchup_retry_tick(now + 60)
2443
2444 assert drain.call_count == 1
2445
2446
2447@pytest.mark.parametrize(
2448 "mode,remote", [("deferred", False), ("realtime", True), ("deferred", True)]
2449)
2450def test_catchup_retry_tick_skips_deferred_and_remote(
2451 tmp_path, monkeypatch, mode, remote
2452):
2453 mod = importlib.import_module("solstone.think.supervisor")
2454 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
2455 drain = MagicMock()
2456 monkeypatch.setattr(mod, "run_catchup_drain", drain)
2457 _arm_catchup_retry_tick(mod, monkeypatch, mode=mode, remote=remote)
2458
2459 now = time.time()
2460 mod._catchup_retry_watermark = now - 60
2461 _write_catchup_state(
2462 tmp_path,
2463 {
2464 "20260101:daily-catchup": _catchup_record(
2465 "20260101", "daily-catchup", next_retry_at=now - 10
2466 )
2467 },
2468 )
2469
2470 mod._run_catchup_retry_tick(now)
2471
2472 drain.assert_not_called()
2473
2474
2475def test_catchup_retry_tick_without_expiry_does_not_fingerprint(tmp_path, monkeypatch):
2476 mod = importlib.import_module("solstone.think.supervisor")
2477 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
2478 drain = MagicMock()
2479 monkeypatch.setattr(mod, "run_catchup_drain", drain)
2480 monkeypatch.setattr(
2481 "solstone.think.catchup_state.read_raw_input_fingerprint",
2482 lambda day: pytest.fail("expiry gate must not hash raw input"),
2483 )
2484 _arm_catchup_retry_tick(mod, monkeypatch)
2485
2486 now = time.time()
2487 mod._catchup_retry_watermark = now - 60
2488 _write_catchup_state(
2489 tmp_path,
2490 {
2491 "20260101:daily-catchup": _catchup_record(
2492 "20260101", "daily-catchup", next_retry_at=now + 3600
2493 )
2494 },
2495 )
2496
2497 mod._run_catchup_retry_tick(now)
2498
2499 drain.assert_not_called()
2500 assert mod._catchup_retry_watermark == now
2501
2502
2503def test_catchup_retry_tick_submits_expired_day_through_drain(tmp_path, monkeypatch):
2504 """End to end: the real drain submits the expired day and skips the backed-off one."""
2505 mod = importlib.import_module("solstone.think.supervisor")
2506 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
2507 capture = _CaptureTaskQueue()
2508 monkeypatch.setattr(mod, "_task_queue", capture)
2509 monkeypatch.setattr(mod, "no_thinking_engine_chosen", lambda: False)
2510 monkeypatch.setattr(mod, "updated_days", lambda **_kw: ["20260101", "20260102"])
2511 monkeypatch.setattr(
2512 "solstone.think.catchup_state.read_raw_input_fingerprint", lambda day: "fp"
2513 )
2514 _arm_catchup_retry_tick(mod, monkeypatch)
2515
2516 now = time.time()
2517 mod._catchup_retry_watermark = now - 60
2518 _write_catchup_state(
2519 tmp_path,
2520 {
2521 "20260101:daily-catchup": _catchup_record(
2522 "20260101", "daily-catchup", next_retry_at=now - 10
2523 ),
2524 # Still inside its backoff window, fingerprint unchanged.
2525 "20260102:daily-catchup": _catchup_record(
2526 "20260102", "daily-catchup", next_retry_at=now + 3600
2527 ),
2528 },
2529 )
2530
2531 mod._run_catchup_retry_tick(now)
2532
2533 assert capture.submissions == [
2534 {"cmd": ["journal", "think", "-v", "--day", "20260101"], "day": "20260101"}
2535 ]
2536
2537
2538def test_catchup_retry_tick_respects_drain_engine_gate(tmp_path, monkeypatch):
2539 """The re-fire routes through run_catchup_drain, so its gates still apply."""
2540 mod = importlib.import_module("solstone.think.supervisor")
2541 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
2542 capture = _CaptureTaskQueue()
2543 monkeypatch.setattr(mod, "_task_queue", capture)
2544 monkeypatch.setattr(mod, "no_thinking_engine_chosen", lambda: True)
2545 monkeypatch.setattr(
2546 mod,
2547 "updated_days",
2548 lambda **_kw: pytest.fail("drain should not enumerate days"),
2549 )
2550 _arm_catchup_retry_tick(mod, monkeypatch)
2551
2552 now = time.time()
2553 mod._catchup_retry_watermark = now - 60
2554 _write_catchup_state(
2555 tmp_path,
2556 {
2557 "20260101:daily-catchup": _catchup_record(
2558 "20260101", "daily-catchup", next_retry_at=now - 10
2559 )
2560 },
2561 )
2562
2563 mod._run_catchup_retry_tick(now)
2564
2565 assert capture.submissions == []
2566
2567
2568def test_catchup_retry_tick_throttles_to_cadence(tmp_path, monkeypatch):
2569 mod = importlib.import_module("solstone.think.supervisor")
2570 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
2571 settings = MagicMock(
2572 side_effect=AssertionError("cadence gate should short-circuit")
2573 )
2574 monkeypatch.setattr(mod, "load_processing_settings", settings)
2575 mod._last_catchup_retry_tick = 100.0
2576
2577 mod._run_catchup_retry_tick(100.0 + mod.CATCHUP_RETRY_TICK_INTERVAL_S - 1)
2578
2579 settings.assert_not_called()
2580
2581
2582def test_run_gate_tick_realtime_skips_catchup_drain(monkeypatch):
2583 mod = importlib.import_module("solstone.think.supervisor")
2584 drain = MagicMock()
2585 evaluate = MagicMock(return_value=SimpleNamespace(open=True))
2586 monkeypatch.setattr(
2587 mod,
2588 "load_processing_settings",
2589 lambda: _supervisor_processing_settings("realtime"),
2590 )
2591 monkeypatch.setattr(mod, "evaluate_drain_gate", evaluate)
2592 monkeypatch.setattr(mod, "run_catchup_drain", drain)
2593 mod._last_gate_tick = 0.0
2594
2595 mod._run_gate_tick(60.0)
2596
2597 evaluate.assert_not_called()
2598 drain.assert_not_called()
2599
2600
2601def test_run_gate_tick_throttles_catchup_drain(monkeypatch):
2602 mod = importlib.import_module("solstone.think.supervisor")
2603 drain = MagicMock()
2604 monkeypatch.setattr(
2605 mod,
2606 "load_processing_settings",
2607 lambda: _supervisor_processing_settings("deferred"),
2608 )
2609 monkeypatch.setattr(
2610 mod,
2611 "evaluate_drain_gate",
2612 lambda settings, now, reading: SimpleNamespace(open=True),
2613 )
2614 monkeypatch.setattr(mod, "run_catchup_drain", drain)
2615 mod._last_gate_tick = 0.0
2616
2617 mod._run_gate_tick(60.0)
2618 mod._run_gate_tick(119.0)
2619 mod._run_gate_tick(121.0)
2620
2621 assert drain.call_count == 2
2622
2623
2624def test_run_gate_tick_disabled_display_powersave_does_not_poll(monkeypatch):
2625 mod = importlib.import_module("solstone.think.supervisor")
2626 drain = MagicMock()
2627 poll = MagicMock(side_effect=AssertionError("poll should not be called"))
2628
2629 def evaluate(settings, now, reading):
2630 assert reading == DISPLAY_POWERSAVE_UNAVAILABLE
2631 return SimpleNamespace(open=True)
2632
2633 monkeypatch.setattr(mod, "_is_remote_mode", False)
2634 monkeypatch.setattr(
2635 mod,
2636 "load_processing_settings",
2637 lambda: _supervisor_processing_settings("deferred"),
2638 )
2639 monkeypatch.setattr(mod, "poll_display_powersave", poll)
2640 monkeypatch.setattr(mod, "evaluate_drain_gate", evaluate)
2641 monkeypatch.setattr(mod, "run_catchup_drain", drain)
2642 mod._last_gate_tick = 0.0
2643
2644 mod._run_gate_tick(60.0)
2645
2646 poll.assert_not_called()
2647 drain.assert_called_once_with()
2648
2649
2650def test_run_gate_tick_enabled_display_powersave_polls(monkeypatch):
2651 mod = importlib.import_module("solstone.think.supervisor")
2652 reading = DisplayPowersaveReading(available=True, asleep=True, debounced=True)
2653 poll = MagicMock(return_value=reading)
2654 captured = {}
2655
2656 def evaluate(settings, now, display_reading):
2657 captured["reading"] = display_reading
2658 return SimpleNamespace(open=False)
2659
2660 monkeypatch.setattr(mod, "_is_remote_mode", False)
2661 monkeypatch.setattr(
2662 mod,
2663 "load_processing_settings",
2664 lambda: _supervisor_processing_settings(
2665 "deferred",
2666 display_powersave_enabled=True,
2667 ),
2668 )
2669 monkeypatch.setattr(mod, "poll_display_powersave", poll)
2670 monkeypatch.setattr(mod, "evaluate_drain_gate", evaluate)
2671 monkeypatch.setattr(mod, "run_catchup_drain", MagicMock())
2672 mod._last_gate_tick = 0.0
2673
2674 mod._run_gate_tick(60.0)
2675
2676 poll.assert_called_once()
2677 assert isinstance(poll.call_args.args[0], float)
2678 assert captured["reading"] == reading
2679
2680
2681def test_supervise_resets_display_powersave_monitor_on_entry(monkeypatch):
2682 mod = importlib.import_module("solstone.think.supervisor")
2683 reset = MagicMock()
2684 monkeypatch.setattr(mod, "reset_display_powersave_monitor", reset)
2685 monkeypatch.setattr(mod, "shutdown_requested", True)
2686
2687 asyncio.run(mod.supervise(daily=False, schedule=False, procs=[]))
2688
2689 reset.assert_called_once_with()
2690
2691
2692def test_supervise_logs_tick_step_failure_and_continues(caplog, monkeypatch):
2693 mod = importlib.import_module("solstone.think.supervisor")
2694 monkeypatch.setattr(mod, "shutdown_requested", False)
2695 monkeypatch.setattr(mod, "_task_queue", None)
2696 monkeypatch.setattr(mod, "_supervisor_callosum", None)
2697 monkeypatch.setattr(mod, "_last_tick_step_failure", None)
2698 monkeypatch.setattr(mod, "reset_display_powersave_monitor", lambda: None)
2699 monkeypatch.setattr(mod, "_run_sync_tick", lambda _now: True)
2700
2701 flush_calls = []
2702 monkeypatch.setattr(
2703 mod, "_check_segment_flush", lambda: flush_calls.append("flush")
2704 )
2705
2706 scheduler_calls = 0
2707
2708 def check_schedule():
2709 nonlocal scheduler_calls
2710 scheduler_calls += 1
2711 if scheduler_calls == 1:
2712 raise Exception("schedule boom")
2713
2714 monkeypatch.setattr(mod.scheduler, "check", check_schedule)
2715
2716 async def stop_after_two_ticks(_seconds):
2717 if scheduler_calls >= 2:
2718 mod.shutdown_requested = True
2719
2720 monkeypatch.setattr(mod.asyncio, "sleep", stop_after_two_ticks)
2721
2722 with caplog.at_level(logging.DEBUG):
2723 asyncio.run(mod.supervise(daily=False, schedule=True, procs=[]))
2724
2725 errors = [
2726 record
2727 for record in caplog.records
2728 if record.levelno == logging.ERROR and "scheduler_check" in record.message
2729 ]
2730 assert len(errors) == 1
2731 assert errors[0].exc_info is not None
2732 assert scheduler_calls == 2
2733 assert len(flush_calls) == 2
2734
2735
2736def test_supervise_propagates_cancelled_error_from_guarded_step(monkeypatch):
2737 mod = importlib.import_module("solstone.think.supervisor")
2738 monkeypatch.setattr(mod, "shutdown_requested", False)
2739 monkeypatch.setattr(mod, "_task_queue", None)
2740 monkeypatch.setattr(mod, "_supervisor_callosum", None)
2741 monkeypatch.setattr(mod, "reset_display_powersave_monitor", lambda: None)
2742 monkeypatch.setattr(mod, "_check_segment_flush", lambda: None)
2743 monkeypatch.setattr(mod, "_run_sync_tick", lambda _now: True)
2744 monkeypatch.setattr(
2745 mod.scheduler,
2746 "check",
2747 lambda: (_ for _ in ()).throw(asyncio.CancelledError()),
2748 )
2749
2750 with pytest.raises(asyncio.CancelledError):
2751 asyncio.run(mod.supervise(daily=False, schedule=True, procs=[]))
2752
2753
2754def test_record_scheduler_completion_serializes_concurrent_writes(
2755 tmp_path, monkeypatch
2756):
2757 mod = importlib.import_module("solstone.think.supervisor")
2758 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
2759
2760 threads = [
2761 threading.Thread(
2762 target=mod._record_scheduler_completion,
2763 args=(name,),
2764 kwargs={
2765 "ended_at": ended_at,
2766 "exit_status": "ok",
2767 "ref": f"ref-{name}",
2768 "cmd": ["sol", name],
2769 },
2770 )
2771 for name, ended_at in [("first", 101.0), ("second", 202.0)]
2772 ]
2773
2774 for thread in threads:
2775 thread.start()
2776 for thread in threads:
2777 thread.join()
2778
2779 state_path = tmp_path / "health" / "scheduler.json"
2780 data = json.loads(state_path.read_text(encoding="utf-8"))
2781 assert data["first"]["last_run"] == 101.0
2782 assert data["second"]["last_run"] == 202.0
2783
2784
2785def test_task_history_records_cap_kill_as_timeout(monkeypatch):
2786 mod = importlib.import_module("solstone.think.supervisor")
2787 queue = mod.TaskQueue(on_queue_change=None)
2788 queue.set_cap("import", 50)
2789 callosum = MagicMock()
2790
2791 class FakeCallosum:
2792 def start(self, callback=None):
2793 return None
2794
2795 def emit(self, *args, **kwargs):
2796 return callosum.emit(*args, **kwargs)
2797
2798 def stop(self):
2799 return None
2800
2801 managed = MagicMock()
2802 managed.pid = 12345
2803 managed.cmd = ["sol", "import"]
2804 managed.ref = "ref-1"
2805 managed.start_time = 100.0
2806 managed.cleanup = MagicMock()
2807
2808 def wait():
2809 queue.enforce_deadlines(200.0)
2810 return -15
2811
2812 managed.wait.side_effect = wait
2813
2814 def fake_spawn(cmd, *, ref=None, callosum=None, day=None, env=None):
2815 return managed
2816
2817 monkeypatch.setattr(mod, "CallosumConnection", FakeCallosum)
2818 monkeypatch.setattr(mod.RunnerManagedProcess, "spawn", fake_spawn)
2819 monkeypatch.setattr(mod, "_start_termination_thread", MagicMock())
2820
2821 queue._run_task(["ref-1"], ["sol", "import"], "import")
2822
2823 assert queue._history[0]["exit_status"] == "timeout"
2824
2825
2826def test_handle_task_request_skips_still_running(monkeypatch):
2827 mod = importlib.import_module("solstone.think.supervisor")
2828 queue = mod.TaskQueue(on_queue_change=None)
2829 managed = _TaskManagedStub(
2830 cmd=["journal", "importer", "--sync", "plaud"], start_time=100.0
2831 )
2832 queue._active["active-ref"] = managed
2833 queue.set_cap("importer", 50)
2834 callosum = MagicMock()
2835
2836 monkeypatch.setattr(mod, "_task_queue", queue)
2837 monkeypatch.setattr(mod, "_supervisor_callosum", callosum)
2838 monkeypatch.setattr(mod.time, "time", lambda: 150.0)
2839
2840 mod._handle_task_request(
2841 {
2842 "tract": "supervisor",
2843 "event": "request",
2844 "cmd": ["journal", "importer", "--sync", "plaud"],
2845 "ref": "requested-ref",
2846 "scheduler_name": "sync-plaud",
2847 }
2848 )
2849
2850 callosum.emit.assert_called_once_with(
2851 "supervisor",
2852 "skipped",
2853 reason="still_running",
2854 ref="requested-ref",
2855 active_ref="active-ref",
2856 cmd=["journal", "importer", "--sync", "plaud"],
2857 scheduler_name="sync-plaud",
2858 )
2859 assert queue._queues == {}
2860
2861
2862def test_handle_task_request_skips_wedged(monkeypatch):
2863 mod = importlib.import_module("solstone.think.supervisor")
2864 queue = mod.TaskQueue(on_queue_change=None)
2865 managed = _TaskManagedStub(
2866 cmd=["journal", "importer", "--sync", "plaud"], start_time=100.0
2867 )
2868 queue._active["active-ref"] = managed
2869 queue.set_cap("importer", 50)
2870 callosum = MagicMock()
2871
2872 monkeypatch.setattr(mod, "_task_queue", queue)
2873 monkeypatch.setattr(mod, "_supervisor_callosum", callosum)
2874 monkeypatch.setattr(mod.time, "time", lambda: 201.0)
2875
2876 mod._handle_task_request(
2877 {
2878 "tract": "supervisor",
2879 "event": "request",
2880 "cmd": ["journal", "importer", "--sync", "plaud"],
2881 "ref": "requested-ref",
2882 }
2883 )
2884
2885 assert callosum.emit.call_args.kwargs["reason"] == "wedged"
2886 assert callosum.emit.call_args.kwargs["active_ref"] == "active-ref"
2887
2888
2889def test_task_queue_shutdown_terminates_active_tasks():
2890 mod = importlib.import_module("solstone.think.supervisor")
2891 queue = mod.TaskQueue(on_queue_change=None)
2892 first = _TaskManagedStub(cmd=["sol", "import"])
2893 second = _TaskManagedStub(cmd=["journal", "indexer"])
2894 queue._active = {"first": first, "second": second}
2895
2896 assert queue.shutdown() == 2
2897
2898 first.terminate.assert_called_once_with(timeout=10.0)
2899 second.terminate.assert_called_once_with(timeout=10.0)
2900
2901
2902def test_task_queue_shutdown_empty_is_noop():
2903 mod = importlib.import_module("solstone.think.supervisor")
2904 queue = mod.TaskQueue(on_queue_change=None)
2905
2906 assert queue.shutdown() == 0
2907
2908
2909def test_task_queue_shutdown_continues_after_timeout():
2910 mod = importlib.import_module("solstone.think.supervisor")
2911 queue = mod.TaskQueue(on_queue_change=None)
2912 first = _TaskManagedStub(cmd=["sol", "import"])
2913 second = _TaskManagedStub(cmd=["journal", "indexer"])
2914 first.terminate.side_effect = subprocess.TimeoutExpired(
2915 cmd=["sol", "import"], timeout=10
2916 )
2917 queue._active = {"first": first, "second": second}
2918
2919 assert queue.shutdown() == 2
2920
2921 first.terminate.assert_called_once_with(timeout=10.0)
2922 second.terminate.assert_called_once_with(timeout=10.0)
2923
2924
2925def test_enforce_deadlines_terminates_when_elapsed_exceeds_cap(caplog, monkeypatch):
2926 mod = importlib.import_module("solstone.think.supervisor")
2927 queue = mod.TaskQueue(on_queue_change=None)
2928 managed = _TaskManagedStub(
2929 cmd=["journal", "importer", "--sync", "plaud", "--save"],
2930 start_time=100.0,
2931 )
2932 queue._active["ref-1"] = managed
2933 queue.set_cap("importer", 50)
2934
2935 def terminate_now(key, managed_arg, timeout, reason):
2936 assert key == "ref-1"
2937 assert managed_arg is managed
2938 assert timeout == 2.0
2939 assert reason == "cap"
2940 managed_arg.terminate(timeout=timeout)
2941
2942 monkeypatch.setattr(mod, "_start_termination_thread", terminate_now)
2943 caplog.set_level("WARNING")
2944 queue.enforce_deadlines(200.0)
2945
2946 managed.terminate.assert_called_once_with(timeout=2.0)
2947 assert (
2948 "Task importer (cmd=journal importer --sync plaud --save, ref=ref-1) exceeded "
2949 "max_runtime of 50s (elapsed=100s); terminating"
2950 ) in caplog.text
2951
2952
2953def test_collect_task_status_reports_default_cap(monkeypatch):
2954 mod = importlib.import_module("solstone.think.supervisor")
2955 queue = mod.TaskQueue(on_queue_change=None)
2956 managed = _TaskManagedStub(cmd=["journal", "providers"], start_time=100.0)
2957 queue._active["ref-1"] = managed
2958 monkeypatch.setattr(mod.time, "time", lambda: 112.0)
2959
2960 assert queue.collect_task_status() == [
2961 {
2962 "ref": "ref-1",
2963 "name": "providers",
2964 "duration_seconds": 12,
2965 "max_runtime_seconds": mod.DEFAULT_TASK_MAX_RUNTIME,
2966 "slow": False,
2967 "stuck": False,
2968 }
2969 ]
2970
2971
2972def test_collect_task_status_under_cap(monkeypatch):
2973 mod = importlib.import_module("solstone.think.supervisor")
2974 queue = mod.TaskQueue(on_queue_change=None)
2975 managed = _TaskManagedStub(cmd=["journal", "providers"], start_time=100.0)
2976 queue._active["ref-1"] = managed
2977 queue.set_cap("providers", 300)
2978 monkeypatch.setattr(mod.time, "time", lambda: 112.0)
2979
2980 status = queue.collect_task_status()
2981
2982 assert status[0]["max_runtime_seconds"] == 300
2983 assert status[0]["slow"] is False
2984 assert status[0]["stuck"] is False
2985
2986
2987def test_collect_task_status_slow_under_cap(monkeypatch):
2988 mod = importlib.import_module("solstone.think.supervisor")
2989 queue = mod.TaskQueue(on_queue_change=None)
2990 managed = _TaskManagedStub(cmd=["journal", "providers"], start_time=100.0)
2991 queue._active["ref-1"] = managed
2992 queue.set_cap("providers", 15)
2993 monkeypatch.setattr(mod.time, "time", lambda: 112.0)
2994
2995 status = queue.collect_task_status()
2996
2997 assert status[0]["max_runtime_seconds"] == 15
2998 assert status[0]["slow"] is True
2999 assert status[0]["stuck"] is False
3000
3001
3002def test_collect_task_status_over_cap(monkeypatch):
3003 mod = importlib.import_module("solstone.think.supervisor")
3004 queue = mod.TaskQueue(on_queue_change=None)
3005 managed = _TaskManagedStub(cmd=["journal", "providers"], start_time=100.0)
3006 queue._active["ref-1"] = managed
3007 queue.set_cap("providers", 5)
3008 monkeypatch.setattr(mod.time, "time", lambda: 112.0)
3009
3010 status = queue.collect_task_status()
3011
3012 assert status[0]["max_runtime_seconds"] == 5
3013 assert status[0]["slow"] is True
3014 assert status[0]["stuck"] is True
3015
3016
3017def test_collect_task_status_default_cap_stuck(monkeypatch):
3018 mod = importlib.import_module("solstone.think.supervisor")
3019 queue = mod.TaskQueue(on_queue_change=None)
3020 managed = _TaskManagedStub(cmd=["journal", "providers"], start_time=100.0)
3021 queue._active["ref-1"] = managed
3022 monkeypatch.setattr(
3023 mod.time, "time", lambda: 100.0 + mod.DEFAULT_TASK_MAX_RUNTIME + 5
3024 )
3025
3026 status = queue.collect_task_status()
3027
3028 assert status[0]["max_runtime_seconds"] == mod.DEFAULT_TASK_MAX_RUNTIME
3029 assert status[0]["stuck"] is True
3030
3031
3032def test_collect_task_status_snapshots_active_under_lock():
3033 mod = importlib.import_module("solstone.think.supervisor")
3034 queue = mod.TaskQueue(on_queue_change=None)
3035
3036 for index in range(25):
3037 managed = _TaskManagedStub(cmd=["journal", "providers"], start_time=100.0)
3038 managed.is_running = MagicMock(side_effect=lambda: time.sleep(0.0001) or True)
3039 queue._active[f"ref-{index}"] = managed
3040
3041 stop = threading.Event()
3042 thread_errors = []
3043
3044 def mutate_active():
3045 index = 0
3046 try:
3047 while not stop.is_set():
3048 ref = f"bg-{index}"
3049 managed = _TaskManagedStub(cmd=["journal", "providers"])
3050 with queue._lock:
3051 queue._active[ref] = managed
3052 time.sleep(0)
3053 with queue._lock:
3054 queue._active.pop(ref, None)
3055 index += 1
3056 except BaseException as exc:
3057 thread_errors.append(exc)
3058
3059 thread = threading.Thread(target=mutate_active)
3060 thread.start()
3061 try:
3062 for _ in range(100):
3063 queue.collect_task_status()
3064 finally:
3065 stop.set()
3066 thread.join(timeout=2)
3067
3068 assert not thread.is_alive()
3069 assert thread_errors == []
3070
3071
3072def test_enforce_deadlines_terminates_stopped_task(caplog, monkeypatch):
3073 mod = importlib.import_module("solstone.think.supervisor")
3074
3075 class StoppedProcess:
3076 def __init__(self, pid):
3077 self.pid = pid
3078
3079 def status(self):
3080 return mod.psutil.STATUS_STOPPED
3081
3082 monkeypatch.setattr(mod.psutil, "Process", StoppedProcess)
3083 terminate = MagicMock()
3084 monkeypatch.setattr(mod, "_start_termination_thread", terminate)
3085 queue = mod.TaskQueue(on_queue_change=None)
3086 managed = _TaskManagedStub(cmd=["sol", "import"], start_time=100.0)
3087 queue.set_cap("import", 300)
3088 queue._active["ref-1"] = managed
3089 caplog.set_level(logging.WARNING)
3090
3091 queue.enforce_deadlines(110.0)
3092 terminate.assert_not_called()
3093
3094 queue.enforce_deadlines(110.0)
3095
3096 terminate.assert_called_once_with("ref-1", managed, timeout=2.0, reason="stopped")
3097 assert "stopped" in caplog.text
3098
3099
3100def test_enforce_deadlines_does_not_probe_status_under_lock(monkeypatch):
3101 mod = importlib.import_module("solstone.think.supervisor")
3102 queue = mod.TaskQueue(on_queue_change=None)
3103 managed = _TaskManagedStub(cmd=["sol", "import"], start_time=100.0)
3104 managed.process.pid = 123
3105 queue._active["ref-1"] = managed
3106 queue.set_cap("import", 300)
3107
3108 class ProbeProcess:
3109 def __init__(self, pid):
3110 assert pid == 123
3111
3112 def status(self):
3113 acquired = queue._lock.acquire(blocking=False)
3114 assert acquired
3115 queue._lock.release()
3116 return mod.psutil.STATUS_RUNNING
3117
3118 monkeypatch.setattr(mod.psutil, "Process", ProbeProcess)
3119 monkeypatch.setattr(mod, "_start_termination_thread", MagicMock())
3120
3121 queue.enforce_deadlines(110.0)
3122
3123
3124def test_enforce_deadlines_skips_stopped_probe_for_new_cap_kill(monkeypatch):
3125 mod = importlib.import_module("solstone.think.supervisor")
3126 queue = mod.TaskQueue(on_queue_change=None)
3127 managed = _TaskManagedStub(cmd=["sol", "import"], start_time=100.0)
3128 queue._active["ref-1"] = managed
3129 queue.set_cap("import", 50)
3130 probe = MagicMock()
3131 terminate = MagicMock()
3132 monkeypatch.setattr(mod.psutil, "Process", probe)
3133 monkeypatch.setattr(mod, "_start_termination_thread", terminate)
3134
3135 queue.enforce_deadlines(200.0)
3136
3137 probe.assert_not_called()
3138 terminate.assert_called_once_with("ref-1", managed, timeout=2.0, reason="cap")
3139
3140
3141def test_terminate_managed_logs_timeout(caplog):
3142 mod = importlib.import_module("solstone.think.supervisor")
3143 managed = _TaskManagedStub(cmd=["sol", "import"], start_time=100.0)
3144 managed.terminate.side_effect = subprocess.TimeoutExpired(
3145 cmd=managed.cmd, timeout=3
3146 )
3147
3148 caplog.set_level("WARNING")
3149 mod._terminate_managed(managed, 3, reason="test")
3150
3151 managed.terminate.assert_called_once_with(timeout=3)
3152 assert "task did not terminate within 3.0s for test" in caplog.text
3153
3154
3155def test_enforce_deadlines_terminates_uncapped_at_default(monkeypatch):
3156 mod = importlib.import_module("solstone.think.supervisor")
3157 queue = mod.TaskQueue(on_queue_change=None)
3158 managed = _TaskManagedStub(cmd=["sol", "import"], start_time=100.0)
3159 queue._active["ref-1"] = managed
3160
3161 def terminate_now(key, managed_arg, timeout, reason):
3162 assert key == "ref-1"
3163 assert managed_arg is managed
3164 assert timeout == 2.0
3165 assert reason == "cap"
3166 managed_arg.terminate(timeout=timeout)
3167
3168 monkeypatch.setattr(mod, "_start_termination_thread", terminate_now)
3169 queue.enforce_deadlines(100.0 + mod.DEFAULT_TASK_MAX_RUNTIME + 1)
3170
3171 managed.terminate.assert_called_once_with(timeout=2.0)
3172
3173
3174def test_restart_service_uses_single_termination_path(monkeypatch):
3175 mod = importlib.import_module("solstone.think.supervisor")
3176 managed = _TaskManagedStub(cmd=["journal", "sense"], start_time=100.0)
3177 managed.name = "sense"
3178 managed.ref = "ref-sense"
3179 mod._managed_procs = [managed]
3180 mod._SERVICE_STATE.clear()
3181 mod._SERVICE_STATE["sense"] = {
3182 "restart": False,
3183 "shutdown_timeout": 7,
3184 }
3185
3186 def terminate_now(key, managed_arg, timeout, reason):
3187 assert key == "sense"
3188 assert managed_arg is managed
3189 assert timeout == 7
3190 assert reason == "restart"
3191 managed_arg.terminate(timeout=timeout)
3192
3193 monkeypatch.setattr(mod, "_start_termination_thread", terminate_now)
3194
3195 assert mod._restart_service("sense") is True
3196 managed.terminate.assert_called_once_with(timeout=7)
3197 assert mod._SERVICE_STATE["sense"]["restart"] is True
3198
3199
3200def test_stop_process_uses_service_shutdown_timeout():
3201 mod = importlib.import_module("solstone.think.supervisor")
3202 managed = _TaskManagedStub(cmd=["journal", "spl"], start_time=100.0)
3203 managed.name = "spl"
3204 mod._SERVICE_STATE.clear()
3205 mod._SERVICE_STATE["spl"] = {
3206 "restart": True,
3207 "shutdown_timeout": 9,
3208 }
3209
3210 mod._stop_process(managed)
3211
3212 managed.terminate.assert_called_once_with(timeout=9)
3213 managed.cleanup.assert_called_once_with()
3214
3215
3216def test_start_local_server_launches_mlx_server_on_darwin(
3217 tmp_path, monkeypatch, capsys
3218):
3219 mod = importlib.import_module("solstone.think.supervisor")
3220 from solstone.think.providers import local_server, local_vulkan
3221
3222 monkeypatch.setattr(sys, "platform", "darwin")
3223 gpu_gate = MagicMock(side_effect=AssertionError("darwin must not probe Vulkan"))
3224 monkeypatch.setattr(local_vulkan, "detect_gpus", gpu_gate)
3225 mod._SERVICE_STATE.clear()
3226 runtime_dir = tmp_path / "gemma4" / "variant-1120"
3227 written_ports = []
3228 spawned = []
3229 spawned_envs = []
3230 managed = _TaskManagedStub(cmd=[])
3231 managed.name = "mlx-vlm-server"
3232 managed.process.returncode = None
3233
3234 plan = _mlx_launch_plan(
3235 mod,
3236 runtime_dir,
3237 model_id="gemma-4-26b-a4b-it-mlx-4bit",
3238 )
3239 monkeypatch.setattr(
3240 mod,
3241 "write_service_port",
3242 lambda service, port: written_ports.append((service, port)),
3243 )
3244 monkeypatch.setattr(local_server, "_probe_health", lambda port: ("ready", None))
3245
3246 def fake_spawn(cmd, *, ref=None, callosum=None, day=None, env=None):
3247 spawned.append(cmd)
3248 spawned_envs.append(env)
3249 managed.cmd = cmd
3250 managed.ref = ref
3251 return managed
3252
3253 monkeypatch.setattr(mod.RunnerManagedProcess, "spawn", fake_spawn)
3254
3255 result = mod.start_local_server(plan, _FakeReservation())
3256
3257 assert result.status == "ready"
3258 assert result.reason_code == "probe-ready"
3259 assert result.managed is managed
3260 assert written_ports == []
3261 assert spawned == [
3262 [
3263 str(Path(sys.executable).with_name("mlx-vlm-server")),
3264 "--host",
3265 "127.0.0.1",
3266 "--port",
3267 "2468",
3268 "--model",
3269 str(runtime_dir),
3270 ]
3271 ]
3272 assert "0.0.0.0" not in spawned[0]
3273 assert "--n-gpu-layers" not in spawned[0]
3274 assert "-c" not in spawned[0]
3275 assert "--device" not in spawned[0]
3276 assert "Vulkan0" not in spawned[0]
3277 assert spawned_envs == [None]
3278 assert "--draft-model" not in spawned[0]
3279 assert "--draft-kind" not in spawned[0]
3280 assert mod._SERVICE_STATE["mlx-vlm-server"]["restart"] is False
3281 assert mod.LOCAL_MODEL_WARMING_UP_COPY in capsys.readouterr().out
3282 gpu_gate.assert_not_called()
3283
3284
3285def test_start_local_server_skips_when_mlx_not_installed_on_darwin(monkeypatch):
3286 mod = importlib.import_module("solstone.think.supervisor")
3287 from solstone.think.providers import mlx_install
3288
3289 monkeypatch.setattr(sys, "platform", "darwin")
3290 monkeypatch.setattr(
3291 mlx_install,
3292 "target_fingerprint",
3293 lambda: {"provider": "local", "backend": "mlx"},
3294 )
3295 monkeypatch.setattr(
3296 mlx_install,
3297 "inspect_readiness",
3298 lambda: _mlx_readiness(model_installed=False),
3299 )
3300 launch = MagicMock()
3301 monkeypatch.setattr(mod, "_launch_process", launch)
3302
3303 observation = mod._observe_mlx_local_provider_truth()
3304
3305 assert observation.phase == "artifact-not-ready"
3306 assert observation.plan is None
3307 launch.assert_not_called()
3308
3309
3310def test_start_local_server_skips_when_mlx_memory_blocked_on_darwin(monkeypatch):
3311 mod = importlib.import_module("solstone.think.supervisor")
3312 from solstone.think.providers import mlx_install
3313
3314 monkeypatch.setattr(sys, "platform", "darwin")
3315 monkeypatch.setattr(
3316 mlx_install,
3317 "target_fingerprint",
3318 lambda: {"provider": "local", "backend": "mlx"},
3319 )
3320 monkeypatch.setattr(
3321 mlx_install,
3322 "inspect_readiness",
3323 lambda: _mlx_readiness(ram_sufficient=False),
3324 )
3325 launch = MagicMock()
3326 monkeypatch.setattr(mod, "_launch_process", launch)
3327
3328 observation = mod._observe_mlx_local_provider_truth()
3329
3330 assert observation.phase == "artifact-not-ready"
3331 assert observation.plan is None
3332 launch.assert_not_called()
3333
3334
3335@pytest.mark.parametrize(
3336 ("vram_mib", "expected_context_tokens", "expected_parallel", "expected_cache_ram"),
3337 [
3338 (6390, None, "1", "0"),
3339 (16000, 32768, "2", "2048"),
3340 ],
3341)
3342def test_start_local_server_launches_llama_server_key_and_cmd(
3343 tmp_path,
3344 monkeypatch,
3345 capsys,
3346 vram_mib: int,
3347 expected_context_tokens: int | None,
3348 expected_parallel: str,
3349 expected_cache_ram: str,
3350):
3351 mod = importlib.import_module("solstone.think.supervisor")
3352 from solstone.think.providers import local_install, local_server, local_vulkan
3353
3354 expected_context_tokens = (
3355 local_server.LOCAL_MIN_CONTEXT_TOKENS
3356 if expected_context_tokens is None
3357 else expected_context_tokens
3358 )
3359 expected_launched_context_tokens = expected_context_tokens * int(expected_parallel)
3360 mod._SERVICE_STATE.clear()
3361 binary = tmp_path / "llama-server"
3362 # ensure_artifacts_installed always resolves artifacts under the selected
3363 # model's directory; the spawn guard rejects anything else, so the stub must
3364 # return realistic in-model-dir paths.
3365 model_artifact_dir = local_install.model_dir(mod.LOCAL_MODEL)
3366 gguf = model_artifact_dir / "model.gguf"
3367 mmproj = model_artifact_dir / "mmproj.gguf"
3368 written_ports = []
3369 written_context_windows = []
3370 spawned = []
3371 spawned_envs = []
3372 managed = _TaskManagedStub(cmd=[])
3373 managed.name = "llama-server"
3374 managed.process.returncode = None
3375 log_path = tmp_path / "llama-server.log"
3376 log_path.write_text(BENIGN_LLAMA_LOAD_LOG, encoding="utf-8")
3377 managed.log_writer = type("LogWriter", (), {"path": log_path})()
3378
3379 plan = _vulkan_launch_plan(mod, binary, gguf, mmproj, vram_mib=vram_mib)
3380 monkeypatch.setattr(local_vulkan, "device_local_used_mib", lambda index: 512)
3381 monkeypatch.setattr(
3382 mod,
3383 "write_service_port",
3384 lambda service, port: written_ports.append((service, port)),
3385 )
3386 monkeypatch.setattr(
3387 local_server,
3388 "write_local_context_window",
3389 lambda tokens: written_context_windows.append(tokens),
3390 )
3391 monkeypatch.setattr(local_server, "_probe_health", lambda port: ("ready", None))
3392 monkeypatch.setattr(local_server, "fetch_props", lambda port: None)
3393
3394 def fake_spawn(cmd, *, ref=None, callosum=None, day=None, env=None):
3395 spawned.append(cmd)
3396 spawned_envs.append(env)
3397 managed.cmd = cmd
3398 managed.ref = ref
3399 return managed
3400
3401 monkeypatch.setattr(mod.RunnerManagedProcess, "spawn", fake_spawn)
3402
3403 result = mod.start_local_server(plan, _FakeReservation())
3404
3405 assert result.status == "ready"
3406 assert result.reason_code == "probe-ready"
3407 assert result.managed is managed
3408 assert written_ports == []
3409 assert written_context_windows == []
3410 assert spawned == [
3411 [
3412 str(binary),
3413 "-m",
3414 str(gguf),
3415 "--alias",
3416 mod.LOCAL_MODEL,
3417 "--host",
3418 "127.0.0.1",
3419 "--port",
3420 "2468",
3421 "--jinja",
3422 "--n-gpu-layers",
3423 "999",
3424 "-c",
3425 str(expected_launched_context_tokens),
3426 "--parallel",
3427 expected_parallel,
3428 "--kv-unified",
3429 "--cache-ram",
3430 expected_cache_ram,
3431 "--no-context-shift",
3432 "--device",
3433 "Vulkan0",
3434 "--mmproj",
3435 str(mmproj),
3436 ]
3437 ]
3438 assert spawned_envs[0]["GGML_VK_VISIBLE_DEVICES"] == "1"
3439 assert "0.0.0.0" not in spawned[0]
3440 assert mod._SERVICE_STATE["llama-server"]["restart"] is False
3441 assert mod.LOCAL_MODEL_WARMING_UP_COPY in capsys.readouterr().out
3442
3443
3444def test_log_context_assertion(caplog):
3445 mod = importlib.import_module("solstone.think.supervisor")
3446 from solstone.think.providers import local_server
3447
3448 def plan_for(tier):
3449 return mod.LocalServerLaunchPlan(
3450 backend="vulkan",
3451 desired_fingerprint_json='{"provider":"local"}',
3452 desired_fingerprint_sha256="fp-local",
3453 context_tokens=tier.context_tokens,
3454 parallel_slots=tier.parallel_slots,
3455 )
3456
3457 floor = plan_for(local_server._FLOOR_TIER)
3458 capable = plan_for(local_server._CAPABLE_TIER)
3459 capable_n_ctx = capable.context_tokens * capable.parallel_slots
3460
3461 with caplog.at_level(logging.INFO):
3462 mod._log_context_assertion(floor, 16384, 1)
3463 assert not any(record.levelno >= logging.WARNING for record in caplog.records)
3464
3465 caplog.clear()
3466 with caplog.at_level(logging.INFO):
3467 mod._log_context_assertion(capable, capable_n_ctx, 2)
3468 assert not any(record.levelno >= logging.WARNING for record in caplog.records)
3469
3470 caplog.clear()
3471 with caplog.at_level(logging.WARNING):
3472 mod._log_context_assertion(capable, capable_n_ctx, 1)
3473 assert any("context MISMATCH" in record.message for record in caplog.records)
3474 assert any("slots MISMATCH" in record.message for record in caplog.records)
3475
3476 caplog.clear()
3477 with caplog.at_level(logging.INFO):
3478 mod._log_context_assertion(capable, capable_n_ctx, None)
3479 assert not any(record.levelno >= logging.WARNING for record in caplog.records)
3480 assert any("context OK" in record.message for record in caplog.records)
3481 assert any("slot count not reported" in record.message for record in caplog.records)
3482
3483 caplog.clear()
3484 with caplog.at_level(logging.WARNING):
3485 mod._log_context_assertion(capable, 12345, 2)
3486 assert any(record.levelno == logging.WARNING for record in caplog.records)
3487
3488 caplog.clear()
3489 with caplog.at_level(logging.INFO):
3490 mod._log_context_assertion(capable, None, None)
3491 assert not any(record.levelno >= logging.WARNING for record in caplog.records)
3492 assert any(
3493 "context assertion skipped" in record.message for record in caplog.records
3494 )
3495
3496
3497def test_start_local_server_skips_missing_artifacts(monkeypatch):
3498 mod = importlib.import_module("solstone.think.supervisor")
3499
3500 readiness = ReadinessOutcome(
3501 provider="local",
3502 status="missing-or-mismatched",
3503 reason_code="manifest_missing",
3504 target={},
3505 install={"install_state": "idle"},
3506 host={},
3507 artifacts={},
3508 proof={},
3509 )
3510 launch = MagicMock()
3511 monkeypatch.setattr(mod, "_launch_process", launch)
3512
3513 observation = mod._readiness_block_observation(
3514 provider="local",
3515 readiness=readiness,
3516 fingerprint_json='{"provider":"local"}',
3517 fingerprint_sha256_value="fp-local",
3518 boot_required=True,
3519 )
3520
3521 assert observation is not None
3522 assert observation.phase == "artifact-not-ready"
3523 assert observation.reason_code == "artifact-missing"
3524 launch.assert_not_called()
3525
3526
3527def _configure_linux_llama_start(
3528 mod,
3529 tmp_path,
3530 monkeypatch,
3531 *,
3532 log_text: str,
3533 poll_return=None,
3534):
3535 from solstone.think.providers import local_install, local_server, local_vulkan
3536
3537 mod._SERVICE_STATE.clear()
3538 binary = tmp_path / "llama-server"
3539 model_artifact_dir = local_install.model_dir(mod.LOCAL_MODEL)
3540 gguf = model_artifact_dir / "model.gguf"
3541 log_path = tmp_path / "llama-server.log"
3542 log_path.write_text(log_text, encoding="utf-8")
3543 managed = _TaskManagedStub(cmd=[])
3544 managed.name = "llama-server"
3545 managed.process.returncode = poll_return
3546 managed.process.poll = MagicMock(return_value=poll_return)
3547 managed.log_writer = type("LogWriter", (), {"path": log_path})()
3548 spawned: list[list[str]] = []
3549 spawned_envs: list[dict[str, str] | None] = []
3550 plan = _vulkan_launch_plan(mod, binary, gguf, None, vram_mib=6390)
3551
3552 monkeypatch.setattr(local_vulkan, "device_local_used_mib", lambda index: 512)
3553 monkeypatch.setattr(mod, "write_service_port", lambda _service, _port: None)
3554 monkeypatch.setattr(
3555 local_server, "write_local_context_window", lambda _tokens: None
3556 )
3557 monkeypatch.setattr(local_server, "_probe_health", lambda _port: ("ready", None))
3558 monkeypatch.setattr(local_server, "fetch_props", lambda _port: None)
3559
3560 def fake_spawn(cmd, *, ref=None, callosum=None, day=None, env=None):
3561 spawned.append(cmd)
3562 spawned_envs.append(env)
3563 managed.cmd = cmd
3564 managed.ref = ref
3565 return managed
3566
3567 monkeypatch.setattr(mod.RunnerManagedProcess, "spawn", fake_spawn)
3568 return managed, spawned, spawned_envs, plan
3569
3570
3571def test_start_local_server_skips_without_hardware_gpu(tmp_path, monkeypatch):
3572 mod = importlib.import_module("solstone.think.supervisor")
3573 from solstone.think.providers import local_vulkan
3574
3575 launch = MagicMock()
3576 monkeypatch.setattr(mod, "_launch_process", launch)
3577
3578 observation = mod.ProviderTruthObservation(
3579 provider="local",
3580 phase="host-blocked",
3581 reason_code="gpu-unavailable",
3582 desired_fingerprint_json='{"provider":"local"}',
3583 desired_fingerprint_sha256="fp-local",
3584 boot_required=True,
3585 detail={
3586 "readiness_status": "host-ineligible",
3587 "readiness_reason_code": "gpu_unavailable",
3588 "devices": mod._format_vulkan_devices([], local_vulkan),
3589 },
3590 )
3591
3592 assert observation.phase == "host-blocked"
3593 assert observation.reason_code == "gpu-unavailable"
3594 launch.assert_not_called()
3595
3596
3597def test_start_local_server_benign_load_log_returns_ready(tmp_path, monkeypatch):
3598 mod = importlib.import_module("solstone.think.supervisor")
3599 managed, _spawned, spawned_envs, plan = _configure_linux_llama_start(
3600 mod, tmp_path, monkeypatch, log_text=BENIGN_LLAMA_LOAD_LOG
3601 )
3602
3603 result = mod.start_local_server(plan, _FakeReservation())
3604
3605 assert result.status == "ready"
3606 assert result.managed is managed
3607 assert spawned_envs[0]["GGML_VK_VISIBLE_DEVICES"] == "1"
3608 managed.terminate.assert_not_called()
3609
3610
3611def test_start_local_server_process_exit_during_warmup_fails_closed(
3612 tmp_path, monkeypatch
3613):
3614 mod = importlib.import_module("solstone.think.supervisor")
3615 managed, _spawned, _envs, plan = _configure_linux_llama_start(
3616 mod,
3617 tmp_path,
3618 monkeypatch,
3619 log_text="2026-06-12T12:00:00+00:00 [llama-server:stderr] loading\n",
3620 poll_return=1,
3621 )
3622
3623 result = mod.start_local_server(plan, _FakeReservation())
3624
3625 assert result.status == "exited"
3626 assert result.reason_code == "process-exited"
3627 assert result.managed is managed
3628 managed.terminate.assert_not_called()
3629 managed.cleanup.assert_not_called()
3630
3631
3632def test_start_local_server_deadline_with_live_process_returns_managed(
3633 tmp_path, monkeypatch
3634):
3635 mod = importlib.import_module("solstone.think.supervisor")
3636 monkeypatch.setattr(mod, "LOCAL_SERVER_READY_TIMEOUT_S", 0.0)
3637 managed, _spawned, _envs, plan = _configure_linux_llama_start(
3638 mod,
3639 tmp_path,
3640 monkeypatch,
3641 log_text="2026-06-12T12:00:00+00:00 [llama-server:stderr] loading\n",
3642 )
3643
3644 result = mod.start_local_server(plan, _FakeReservation())
3645
3646 assert result.status == "warmup-timeout"
3647 assert result.reason_code == "warmup-timeout"
3648 assert result.managed is managed
3649 managed.terminate.assert_not_called()
3650 managed.cleanup.assert_not_called()
3651
3652
3653def _configure_cuda_llama_start(
3654 mod,
3655 tmp_path,
3656 monkeypatch,
3657 *,
3658 probe,
3659 ready: bool = True,
3660):
3661 from solstone.think.providers import local_install, local_server
3662
3663 mod._SERVICE_STATE.clear()
3664 binary = tmp_path / "llama-server"
3665 model_artifact_dir = local_install.model_dir(mod.LOCAL_MODEL)
3666 gguf = model_artifact_dir / "model.gguf"
3667 mmproj = model_artifact_dir / "mmproj.gguf"
3668 lib_dir = tmp_path / "cuda-lib"
3669 lib_dir.mkdir()
3670 managed = _TaskManagedStub(cmd=[])
3671 managed.name = "llama-server"
3672 managed.process.returncode = None
3673 written_ports = []
3674 written_context_windows = []
3675 spawned: list[list[str]] = []
3676 spawned_envs: list[dict[str, str] | None] = []
3677 plan = _cuda_launch_plan(
3678 mod,
3679 binary,
3680 gguf,
3681 mmproj,
3682 lib_dir,
3683 tiering_memory_mib=probe.tiering_memory_mib,
3684 visible_device=str(probe.index if probe.index is not None else 0),
3685 )
3686
3687 monkeypatch.setattr(
3688 mod,
3689 "write_service_port",
3690 lambda service, port: written_ports.append((service, port)),
3691 )
3692 monkeypatch.setattr(
3693 local_server,
3694 "write_local_context_window",
3695 lambda tokens: written_context_windows.append(tokens),
3696 )
3697 monkeypatch.setattr(
3698 local_server,
3699 "_probe_health",
3700 lambda _port: (
3701 (local_server.STATE_READY, None)
3702 if ready
3703 else (local_server.STATE_FAILED, "loading")
3704 ),
3705 )
3706 monkeypatch.setattr(local_server, "fetch_props", lambda _port: None)
3707
3708 def fake_spawn(cmd, *, ref=None, callosum=None, day=None, env=None):
3709 spawned.append(cmd)
3710 spawned_envs.append(env)
3711 managed.cmd = cmd
3712 managed.ref = ref
3713 return managed
3714
3715 monkeypatch.setattr(mod.RunnerManagedProcess, "spawn", fake_spawn)
3716 return (
3717 managed,
3718 spawned,
3719 spawned_envs,
3720 written_ports,
3721 written_context_windows,
3722 lib_dir,
3723 plan,
3724 )
3725
3726
3727def test_start_local_server_cuda_launches_llama_server_cmd_and_env(
3728 tmp_path, monkeypatch
3729):
3730 mod = importlib.import_module("solstone.think.supervisor")
3731 from solstone.think.providers import local_cuda
3732
3733 monkeypatch.setenv("LD_LIBRARY_PATH", "/existing/lib")
3734 probe = local_cuda.NvidiaProbe(
3735 index=0,
3736 compute_cap="sm_121",
3737 driver_cuda_version=13,
3738 vram_mib=20000,
3739 tiering_memory_mib=20000,
3740 memory_source=local_cuda.MEMORY_SOURCE_NVIDIA_VRAM,
3741 detected=True,
3742 )
3743 (
3744 managed,
3745 spawned,
3746 spawned_envs,
3747 written_ports,
3748 written_context_windows,
3749 lib_dir,
3750 plan,
3751 ) = _configure_cuda_llama_start(mod, tmp_path, monkeypatch, probe=probe)
3752
3753 result = mod.start_local_server(plan, _FakeReservation())
3754
3755 assert result.status == "ready"
3756 assert result.managed is managed
3757 assert written_ports == []
3758 assert written_context_windows == []
3759 assert len(spawned) == 1
3760 cmd = spawned[0]
3761 assert cmd[cmd.index("--device") + 1] == "CUDA0"
3762 assert cmd[cmd.index("-c") + 1] == str(plan.context_tokens * plan.parallel_slots)
3763 assert cmd[cmd.index("--host") + 1] == "127.0.0.1"
3764 assert cmd[cmd.index("--n-gpu-layers") + 1] == "999"
3765 assert "--mmproj" in cmd
3766 assert "0.0.0.0" not in cmd
3767 assert spawned_envs[0]["CUDA_VISIBLE_DEVICES"] == "0"
3768 assert spawned_envs[0]["LD_LIBRARY_PATH"] == f"{lib_dir}:/existing/lib"
3769
3770
3771def test_start_local_server_cuda_missing_vram_uses_floor_tier(
3772 tmp_path, monkeypatch, caplog
3773):
3774 mod = importlib.import_module("solstone.think.supervisor")
3775 from solstone.think.providers import local_cuda, local_server
3776
3777 probe = local_cuda.NvidiaProbe(
3778 index=0,
3779 compute_cap="sm_121",
3780 driver_cuda_version=13,
3781 vram_mib=None,
3782 tiering_memory_mib=None,
3783 memory_source=local_cuda.MEMORY_SOURCE_UNAVAILABLE,
3784 detected=True,
3785 )
3786 (
3787 managed,
3788 spawned,
3789 _spawned_envs,
3790 _ports,
3791 written_context_windows,
3792 _lib_dir,
3793 plan,
3794 ) = _configure_cuda_llama_start(mod, tmp_path, monkeypatch, probe=probe)
3795
3796 with caplog.at_level(logging.INFO):
3797 result = mod.start_local_server(plan, _FakeReservation())
3798
3799 assert result.status == "ready"
3800 assert result.managed is managed
3801 assert written_context_windows == []
3802 assert spawned[0][spawned[0].index("-c") + 1] == str(
3803 local_server.LOCAL_MIN_CONTEXT_TOKENS
3804 )
3805 assert any(
3806 "local server backend=cuda" in record.message for record in caplog.records
3807 )
3808
3809
3810def test_start_local_server_cuda_unified_memory_uses_capable_tier(
3811 tmp_path, monkeypatch, caplog
3812):
3813 mod = importlib.import_module("solstone.think.supervisor")
3814 from solstone.think.providers import local_cuda
3815
3816 probe = local_cuda.NvidiaProbe(
3817 index=0,
3818 compute_cap="sm_121",
3819 driver_cuda_version=13,
3820 vram_mib=None,
3821 tiering_memory_mib=26724,
3822 memory_source=local_cuda.MEMORY_SOURCE_SYSTEM_AVAILABLE,
3823 detected=True,
3824 )
3825 (
3826 managed,
3827 spawned,
3828 _spawned_envs,
3829 _ports,
3830 written_context_windows,
3831 _lib_dir,
3832 plan,
3833 ) = _configure_cuda_llama_start(mod, tmp_path, monkeypatch, probe=probe)
3834
3835 with caplog.at_level(logging.INFO):
3836 result = mod.start_local_server(plan, _FakeReservation())
3837
3838 assert result.status == "ready"
3839 assert result.managed is managed
3840 assert written_context_windows == []
3841 assert spawned[0][spawned[0].index("-c") + 1] == str(
3842 plan.context_tokens * plan.parallel_slots
3843 )
3844 assert any(
3845 "local server backend=cuda" in record.message for record in caplog.records
3846 )
3847
3848
3849def test_start_local_server_cuda_low_unified_memory_uses_floor_tier(
3850 tmp_path, monkeypatch
3851):
3852 mod = importlib.import_module("solstone.think.supervisor")
3853 from solstone.think.providers import local_cuda, local_server
3854
3855 probe = local_cuda.NvidiaProbe(
3856 index=0,
3857 compute_cap="sm_121",
3858 driver_cuda_version=13,
3859 vram_mib=None,
3860 tiering_memory_mib=8000,
3861 memory_source=local_cuda.MEMORY_SOURCE_SYSTEM_AVAILABLE,
3862 detected=True,
3863 )
3864 (
3865 _managed,
3866 spawned,
3867 _spawned_envs,
3868 _ports,
3869 written_context_windows,
3870 _lib_dir,
3871 plan,
3872 ) = _configure_cuda_llama_start(mod, tmp_path, monkeypatch, probe=probe)
3873
3874 mod.start_local_server(plan, _FakeReservation())
3875
3876 assert written_context_windows == []
3877 assert spawned[0][spawned[0].index("-c") + 1] == str(
3878 local_server.LOCAL_MIN_CONTEXT_TOKENS
3879 )
3880
3881
3882def test_start_local_server_cuda_warmup_timeout_fails_closed(tmp_path, monkeypatch):
3883 mod = importlib.import_module("solstone.think.supervisor")
3884 from solstone.think.providers import local_cuda
3885
3886 monkeypatch.setattr(mod, "LOCAL_SERVER_READY_TIMEOUT_S", 0.0)
3887 probe = local_cuda.NvidiaProbe(
3888 index=0,
3889 compute_cap="sm_121",
3890 driver_cuda_version=13,
3891 vram_mib=20000,
3892 tiering_memory_mib=20000,
3893 memory_source=local_cuda.MEMORY_SOURCE_NVIDIA_VRAM,
3894 detected=True,
3895 )
3896 (
3897 managed,
3898 _spawned,
3899 _spawned_envs,
3900 _ports,
3901 _contexts,
3902 _lib_dir,
3903 plan,
3904 ) = _configure_cuda_llama_start(
3905 mod, tmp_path, monkeypatch, probe=probe, ready=False
3906 )
3907 terminated = []
3908
3909 def terminate_managed(managed_arg, timeout, *, reason):
3910 terminated.append((managed_arg, timeout, reason))
3911
3912 monkeypatch.setattr(mod, "_terminate_managed", terminate_managed)
3913
3914 result = mod.start_local_server(plan, _FakeReservation())
3915
3916 assert result.status == "warmup-timeout"
3917 assert result.reason_code == "warmup-timeout"
3918 assert result.managed is managed
3919 assert terminated == []
3920 managed.cleanup.assert_not_called()
3921
3922
3923def test_supervisor_provider_start_lifecycle_events_are_ignored():
3924 mod = importlib.import_module("solstone.think.supervisor")
3925
3926 mod._handle_callosum_message({"tract": "supervisor", "event": "start_local"})
3927 mod._handle_callosum_message({"tract": "supervisor", "event": "start_parakeet"})
3928
3929 assert not hasattr(mod, "_request_provider_runtime_retry")
3930
3931
3932def test_supervisor_provider_start_lifecycle_handlers_are_absent():
3933 mod = importlib.import_module("solstone.think.supervisor")
3934
3935 assert not hasattr(mod, "_handle_supervisor_start_local")
3936 assert not hasattr(mod, "_handle_supervisor_start_parakeet")
3937 assert not hasattr(mod, "_request_local_server_start")
3938 assert not hasattr(mod, "_request_parakeet_server_start")
3939
3940
3941def test_handle_runner_exits_reports_llama_server_to_reconciler(monkeypatch):
3942 mod = importlib.import_module("solstone.think.supervisor")
3943 mod._SERVICE_STATE.clear()
3944 mod._RESTART_POLICIES.clear()
3945 monkeypatch.setattr(mod.time, "time", lambda: 100.0)
3946 monkeypatch.setattr(mod, "shutdown_requested", False)
3947
3948 managed = _TaskManagedStub(cmd=["/tmp/llama-server", "-m", "/tmp/model.gguf"])
3949 managed.name = "llama-server"
3950 managed.process.poll.return_value = 1
3951 managed.process.returncode = 1
3952 state = mod.ProviderRuntimeState("local")
3953 state.generation = 2
3954 state.desired_fingerprint = "fp-local"
3955 state.latest_phase = "ready"
3956 state.latest_plan = _vulkan_launch_plan(
3957 mod,
3958 Path("/tmp/llama-server"),
3959 Path("/tmp/model.gguf"),
3960 None,
3961 vram_mib=8000,
3962 )
3963 writes = []
3964
3965 mod._SERVICE_STATE["llama-server"] = {
3966 "restart": True,
3967 "shutdown_timeout": 12,
3968 }
3969 monkeypatch.setattr(
3970 mod,
3971 "_provider_runtime_states",
3972 {
3973 "local": state,
3974 "parakeet": mod.ProviderRuntimeState("parakeet"),
3975 },
3976 )
3977 monkeypatch.setattr(
3978 mod,
3979 "_recovery_state",
3980 {
3981 "local": mod.ProviderRecoveryState(),
3982 "parakeet": mod.ProviderRecoveryState(),
3983 },
3984 )
3985 monkeypatch.setattr(
3986 mod,
3987 "_write_provider_runtime",
3988 lambda state_arg, **kwargs: writes.append((state_arg, kwargs)),
3989 )
3990
3991 def fake_launch(name, cmd, *, restart=False, shutdown_timeout=15, ref=None):
3992 raise AssertionError("provider-owned exits must not use generic restart")
3993
3994 monkeypatch.setattr(mod, "_launch_process", fake_launch)
3995 monkeypatch.setattr(mod, "_supervisor_callosum", None)
3996
3997 procs = [managed]
3998 asyncio.run(mod.handle_runner_exits(procs))
3999
4000 assert procs == []
4001 assert state.generation == 3
4002 assert state.latest_phase == "stopped"
4003 assert state.latest_plan is None
4004 assert state.next_truth_at == 0.0
4005 assert mod._recovery_state["local"].down_generation == 3
4006 assert mod._SERVICE_STATE["llama-server"] == {
4007 "restart": True,
4008 "shutdown_timeout": 12,
4009 }
4010 assert writes[-1][1]["phase"] == "stopped"
4011 assert writes[-1][1]["reason_code"] == "process-exited"
4012
4013
4014def test_supervisor_singleton_lock_acquired(tmp_path, monkeypatch):
4015 mod = importlib.reload(importlib.import_module("solstone.think.supervisor"))
4016
4017 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
4018 (tmp_path / "health").mkdir(parents=True, exist_ok=True)
4019 monkeypatch.setattr(sys, "argv", ["supervisor"])
4020
4021 def stop_after_lock():
4022 raise SystemExit(0)
4023
4024 # Skip maint discovery/subprocess runs — unrelated to lock acquisition and
4025 # slow enough on a fresh tmp_path to blow the 5s pytest-timeout under load.
4026 monkeypatch.setattr(mod, "run_pending_tasks", lambda *a, **k: [])
4027 monkeypatch.setattr(mod, "_sweep_orphaned_sol_processes", lambda *_a, **_k: 0)
4028 monkeypatch.setattr(mod.time, "sleep", lambda _seconds: None)
4029 monkeypatch.setattr(mod, "start_callosum_in_process", stop_after_lock)
4030
4031 with pytest.raises(SystemExit) as exc:
4032 mod.main()
4033
4034 assert exc.value.code == 0
4035 assert (tmp_path / "health" / "supervisor.lock").exists()
4036 assert (tmp_path / "health" / "supervisor.pid").read_text().strip() == str(
4037 os.getpid()
4038 )
4039 start_time = float(
4040 (tmp_path / "health" / "supervisor.start_time").read_text().strip()
4041 )
4042 assert start_time == psutil.Process(os.getpid()).create_time()
4043 assert mod.is_supervisor_up() is True
4044
4045
4046def test_supervisor_blocks_before_callosum_on_blocking_maint_failure(
4047 tmp_path, monkeypatch, capsys
4048):
4049 mod = importlib.reload(importlib.import_module("solstone.think.supervisor"))
4050 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
4051 (tmp_path / "health").mkdir(parents=True, exist_ok=True)
4052 monkeypatch.setattr(sys, "argv", ["supervisor"])
4053 task = MaintTask(
4054 app="thinking",
4055 name="001_migrate_provider_install_state",
4056 script_path=Path("/dummy.py"),
4057 retry_on_next_start=True,
4058 blocks_supervisor_start=True,
4059 )
4060 state_file = tmp_path / "maint" / "thinking" / f"{task.name}.jsonl"
4061 result = MaintTaskResult(
4062 task=task,
4063 success=False,
4064 exit_code=7,
4065 state_file=state_file,
4066 )
4067 monkeypatch.setattr(mod, "run_pending_tasks", lambda *a, **k: [result])
4068 monkeypatch.setattr(mod, "_sweep_orphaned_sol_processes", lambda *_a, **_k: 0)
4069 start_mock = MagicMock()
4070 monkeypatch.setattr(mod, "start_callosum_in_process", start_mock)
4071
4072 with pytest.raises(SystemExit) as exc:
4073 mod.main()
4074
4075 assert exc.value.code == 1
4076 start_mock.assert_not_called()
4077 captured = capsys.readouterr()
4078 assert "thinking:001_migrate_provider_install_state" in captured.err
4079 assert str(state_file) in captured.err
4080 assert "retry-on-next-start" in captured.err
4081
4082
4083def test_supervisor_continues_after_successful_blocking_maint_task(
4084 tmp_path, monkeypatch
4085):
4086 mod = importlib.reload(importlib.import_module("solstone.think.supervisor"))
4087 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
4088 (tmp_path / "health").mkdir(parents=True, exist_ok=True)
4089 monkeypatch.setattr(sys, "argv", ["supervisor"])
4090 task = MaintTask(
4091 app="thinking",
4092 name="001_migrate_provider_install_state",
4093 script_path=Path("/dummy.py"),
4094 retry_on_next_start=True,
4095 blocks_supervisor_start=True,
4096 )
4097 result = MaintTaskResult(
4098 task=task,
4099 success=True,
4100 exit_code=0,
4101 state_file=tmp_path / "maint" / "thinking" / f"{task.name}.jsonl",
4102 )
4103 monkeypatch.setattr(mod, "run_pending_tasks", lambda *a, **k: [result])
4104 monkeypatch.setattr(mod, "_sweep_orphaned_sol_processes", lambda *_a, **_k: 0)
4105
4106 def stop_after_callosum():
4107 raise SystemExit(0)
4108
4109 start_mock = MagicMock(side_effect=stop_after_callosum)
4110 monkeypatch.setattr(mod, "start_callosum_in_process", start_mock)
4111
4112 with pytest.raises(SystemExit) as exc:
4113 mod.main()
4114
4115 assert exc.value.code == 0
4116 start_mock.assert_called_once()
4117
4118
4119def test_supervisor_singleton_lock_blocked(tmp_path, monkeypatch, capsys):
4120 import fcntl
4121
4122 mod = importlib.reload(importlib.import_module("solstone.think.supervisor"))
4123
4124 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
4125 monkeypatch.delenv("INVOCATION_ID", raising=False)
4126 health_dir = tmp_path / "health"
4127 health_dir.mkdir(parents=True, exist_ok=True)
4128 lock_file = open(health_dir / "supervisor.lock", "w")
4129 fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
4130 (health_dir / "supervisor.pid").write_text("12345")
4131 monkeypatch.setattr(sys, "argv", ["supervisor"])
4132
4133 start_mock = MagicMock()
4134 monkeypatch.setattr(mod, "start_callosum_in_process", start_mock)
4135
4136 try:
4137 with pytest.raises(SystemExit) as exc:
4138 mod.main()
4139 finally:
4140 lock_file.close()
4141
4142 assert exc.value.code == 1
4143 output = capsys.readouterr().out
4144 assert "Supervisor already running" in output
4145 assert "PID 12345" in output
4146 start_mock.assert_not_called()
4147
4148
4149def test_supervisor_singleton_lock_blocked_under_systemd_exits_cleanly(
4150 tmp_path, monkeypatch, capsys
4151):
4152 import fcntl
4153
4154 mod = importlib.reload(importlib.import_module("solstone.think.supervisor"))
4155
4156 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
4157 monkeypatch.setenv("INVOCATION_ID", "test-invocation-uuid")
4158 health_dir = tmp_path / "health"
4159 health_dir.mkdir(parents=True, exist_ok=True)
4160 lock_file = open(health_dir / "supervisor.lock", "w")
4161 fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
4162 (health_dir / "supervisor.pid").write_text("12345")
4163 monkeypatch.setattr(sys, "argv", ["supervisor"])
4164
4165 start_mock = MagicMock()
4166 monkeypatch.setattr(mod, "start_callosum_in_process", start_mock)
4167
4168 try:
4169 with pytest.raises(SystemExit) as exc:
4170 mod.main()
4171 finally:
4172 lock_file.close()
4173
4174 assert exc.value.code == 0
4175 output = capsys.readouterr().out
4176 assert (
4177 "Supervisor already running (PID 12345) - exiting cleanly under "
4178 "systemd activation"
4179 ) in output
4180 start_mock.assert_not_called()
4181
4182
4183def test_supervisor_singleton_lock_blocked_with_health(tmp_path, monkeypatch, capsys):
4184 import fcntl
4185
4186 mod = importlib.reload(importlib.import_module("solstone.think.supervisor"))
4187
4188 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
4189 monkeypatch.delenv("INVOCATION_ID", raising=False)
4190 health_dir = tmp_path / "health"
4191 health_dir.mkdir(parents=True, exist_ok=True)
4192 lock_file = open(health_dir / "supervisor.lock", "w")
4193 fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
4194 (health_dir / "supervisor.pid").write_text("12345")
4195 (health_dir / "callosum.sock").touch()
4196 monkeypatch.setattr(sys, "argv", ["supervisor"])
4197
4198 start_mock = MagicMock()
4199 health_mock = MagicMock(return_value=0)
4200 monkeypatch.setattr(mod, "start_callosum_in_process", start_mock)
4201 monkeypatch.setattr("solstone.think.health_cli.health_check", health_mock)
4202
4203 try:
4204 with pytest.raises(SystemExit) as exc:
4205 mod.main()
4206 finally:
4207 lock_file.close()
4208
4209 assert exc.value.code == 1
4210 output = capsys.readouterr().out
4211 assert "Supervisor already running" in output
4212 assert "PID 12345" in output
4213 health_mock.assert_called_once_with()
4214 start_mock.assert_not_called()
4215
4216
4217def test_supervisor_singleton_lock_blocked_with_health_under_systemd_skips_health_check(
4218 tmp_path, monkeypatch, capsys
4219):
4220 import fcntl
4221
4222 mod = importlib.reload(importlib.import_module("solstone.think.supervisor"))
4223
4224 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
4225 monkeypatch.setenv("INVOCATION_ID", "test-invocation-uuid")
4226 health_dir = tmp_path / "health"
4227 health_dir.mkdir(parents=True, exist_ok=True)
4228 lock_file = open(health_dir / "supervisor.lock", "w")
4229 fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
4230 (health_dir / "supervisor.pid").write_text("12345")
4231 (health_dir / "callosum.sock").touch()
4232 monkeypatch.setattr(sys, "argv", ["supervisor"])
4233
4234 start_mock = MagicMock()
4235 health_mock = MagicMock(return_value=0)
4236 monkeypatch.setattr(mod, "start_callosum_in_process", start_mock)
4237 monkeypatch.setattr("solstone.think.health_cli.health_check", health_mock)
4238
4239 try:
4240 with pytest.raises(SystemExit) as exc:
4241 mod.main()
4242 finally:
4243 lock_file.close()
4244
4245 assert exc.value.code == 0
4246 output = capsys.readouterr().out
4247 assert (
4248 "Supervisor already running (PID 12345) - exiting cleanly under "
4249 "systemd activation"
4250 ) in output
4251 health_mock.assert_not_called()
4252 start_mock.assert_not_called()
4253
4254
4255def test_is_supervisor_up_without_pid_file(tmp_path, monkeypatch):
4256 mod = importlib.reload(importlib.import_module("solstone.think.supervisor"))
4257
4258 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
4259 (tmp_path / "health").mkdir(parents=True, exist_ok=True)
4260
4261 assert mod.is_supervisor_up() is False
4262
4263
4264def test_is_supervisor_up_with_dead_pid(tmp_path, monkeypatch):
4265 mod = importlib.reload(importlib.import_module("solstone.think.supervisor"))
4266
4267 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
4268 health_dir = tmp_path / "health"
4269 health_dir.mkdir(parents=True, exist_ok=True)
4270
4271 proc = subprocess.Popen(["true"])
4272 proc.wait()
4273 (health_dir / "supervisor.pid").write_text(str(proc.pid))
4274
4275 assert mod.is_supervisor_up() is False
4276
4277
4278def test_is_supervisor_up_with_live_pid_missing_start_time(tmp_path, monkeypatch):
4279 mod = importlib.reload(importlib.import_module("solstone.think.supervisor"))
4280
4281 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
4282 health_dir = tmp_path / "health"
4283 health_dir.mkdir(parents=True, exist_ok=True)
4284 (health_dir / "supervisor.pid").write_text(str(os.getpid()))
4285
4286 assert mod.is_supervisor_up() is False
4287
4288
4289def test_is_supervisor_up_with_live_pid_mismatched_start_time(tmp_path, monkeypatch):
4290 mod = importlib.reload(importlib.import_module("solstone.think.supervisor"))
4291
4292 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
4293 health_dir = tmp_path / "health"
4294 health_dir.mkdir(parents=True, exist_ok=True)
4295 (health_dir / "supervisor.pid").write_text(str(os.getpid()))
4296 create_time = psutil.Process(os.getpid()).create_time()
4297 (health_dir / "supervisor.start_time").write_text(str(create_time + 60))
4298
4299 assert mod.is_supervisor_up() is False
4300
4301
4302def test_is_supervisor_up_with_matching_process_identity(tmp_path, monkeypatch):
4303 mod = importlib.reload(importlib.import_module("solstone.think.supervisor"))
4304
4305 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
4306 health_dir = tmp_path / "health"
4307 health_dir.mkdir(parents=True, exist_ok=True)
4308 (health_dir / "supervisor.pid").write_text(str(os.getpid()))
4309 (health_dir / "supervisor.start_time").write_text(
4310 str(psutil.Process(os.getpid()).create_time())
4311 )
4312
4313 assert mod.is_supervisor_up() is True
4314
4315
4316def test_is_supervisor_up_boundary_at_shared_tolerance(tmp_path, monkeypatch):
4317 mod = importlib.reload(importlib.import_module("solstone.think.supervisor"))
4318
4319 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
4320 health_dir = tmp_path / "health"
4321 health_dir.mkdir(parents=True, exist_ok=True)
4322 (health_dir / "supervisor.pid").write_text(str(os.getpid()))
4323 create_time = psutil.Process(os.getpid()).create_time()
4324 within = create_time + mod.START_TIME_TOLERANCE_S - 0.1
4325 beyond = create_time + mod.START_TIME_TOLERANCE_S + 0.1
4326
4327 (health_dir / "supervisor.start_time").write_text(str(within))
4328 assert mod.is_supervisor_up() is True
4329
4330 (health_dir / "supervisor.start_time").write_text(str(beyond))
4331 assert mod.is_supervisor_up() is False