solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4from __future__ import annotations
5
6import argparse
7import asyncio
8import concurrent.futures
9import fcntl
10import getpass
11import json
12import logging
13import os
14import signal
15import socket
16import stat
17import subprocess
18import sys
19import tempfile
20import threading
21import time
22from collections import OrderedDict, deque
23from datetime import datetime, timezone
24from logging.handlers import RotatingFileHandler
25from pathlib import Path
26from typing import Any, Callable, Iterable, NoReturn
27
28import psutil
29
30from solstone.think import maintenance, scheduler
31from solstone.think.app_supervised import FLAG, is_app_supervised, resolve_parent_fd
32from solstone.think.backup.engine import BACKUP_MAX_RUNTIME, BACKUP_RUN_CMD
33from solstone.think.callosum import CallosumConnection, CallosumServer
34from solstone.think.catchup_state import (
35 KIND_DAILY_CATCHUP,
36 STUCK_THRESHOLD,
37 day_eligible_to_drain,
38 reconcile_interrupted_attempts,
39 record_attempt,
40 record_outcome,
41)
42from solstone.think.display_powersave import (
43 poll_display_powersave,
44 reset_display_powersave_monitor,
45)
46from solstone.think.maint import run_pending_tasks
47from solstone.think.models import LOCAL_MODEL, is_local_provider_needed
48from solstone.think.processing import (
49 DISPLAY_POWERSAVE_UNAVAILABLE,
50 evaluate_drain_gate,
51 load_processing_settings,
52)
53from solstone.think.providers.mlx_server import MLX_SERVER_PROCESS_NAME
54from solstone.think.readiness import START_TIME_TOLERANCE_S, clear_ready, signal_ready
55from solstone.think.runner import ManagedProcess as RunnerManagedProcess
56from solstone.think.runner import _command_partition
57from solstone.think.sync_check import (
58 DEFAULT_INTERVAL_SECONDS,
59 SyncCheckSnapshot,
60 check_journal_sync,
61 clear_self_heartbeat,
62 format_conflict_message,
63 write_self_heartbeat,
64)
65from solstone.think.utils import (
66 EXIT_EMPTY,
67 EXIT_TEMPFAIL,
68 SOFT_RUNTIME_FRACTION,
69 day_path,
70 find_available_port,
71 get_journal,
72 get_journal_info,
73 get_rev,
74 is_solstone_up,
75 now_ms,
76 parse_duration_seconds,
77 read_service_port,
78 setup_cli,
79 updated_days,
80 write_service_port,
81)
82
83DEFAULT_TASK_MAX_RUNTIME = (
84 1800 # 30m wall-clock default for any uncapped TaskQueue partition
85)
86REACTIVE_TASK_CAPS = {
87 "daily": 21600, # 6h daily/from-scratch reprocess
88 "segment": 4500, # 75m per-segment think
89 "indexer": 7200, # 2h indexer, above the code's own 1h rescan allotment
90 "importer": 3600, # 1h importer
91}
92DEFAULT_THRESHOLD = 60
93CHECK_INTERVAL = 30
94GATE_TICK_INTERVAL_S = 60
95MAX_UPDATED_CATCHUP = 4
96TEMPFAIL_DELAY = 15 # seconds to wait before retrying a tempfail exit
97CONVEY_READY_WINDOW_SECONDS = 60.0
98REALISTIC_COLD_BIND_SECONDS = 40.0
99HANDLE_SHUTDOWN_REAP_S = 3.0
100APP_SUPERVISED_SHUTDOWN_CEILING_S = 10.0
101APP_SUPERVISED_TASK_DRAIN_S = 2.0
102APP_SUPERVISED_CHILD_STOP_S = 2.0
103APP_SUPERVISED_CALLOSUM_JOIN_S = 2.0
104PARENT_DEATH_POLL_INTERVAL_S = 1.0
105STOPPED_TICKS_THRESHOLD = 2
106LOCAL_SERVER_PROCESS_NAME = "llama-server"
107LOCAL_WEDGE_THRESHOLD = 3
108LOCAL_WEDGE_RECYCLE_GRACE_S = 120.0
109LOCAL_WEDGE_PROVIDER_MAP_CAP = 512
110LOCAL_SERVER_READY_TIMEOUT_S = 300.0
111LOCAL_SERVER_HEALTH_POLL_INTERVAL_S = 1.0
112LOCAL_MODEL_WARMING_UP_COPY = "Local model is warming up..."
113# supervisor.log is size-rotated with a bounded on-disk footprint.
114# Hard ceiling = SUPERVISOR_LOG_MAX_BYTES * (SUPERVISOR_LOG_BACKUP_COUNT + 1)
115# = 16 MiB * 6 = 96 MiB. Older lines drop; the most-recent tail is kept.
116SUPERVISOR_LOG_MAX_BYTES = 16 * 1024 * 1024
117SUPERVISOR_LOG_BACKUP_COUNT = 5
118logger = logging.getLogger(__name__)
119
120
121def _compact_log_if_oversized(log_path: Path, max_bytes: int) -> None:
122 try:
123 size = log_path.stat().st_size
124 except FileNotFoundError:
125 return
126 except OSError as error:
127 logger.warning("Could not stat supervisor log before compaction: %s", error)
128 return
129
130 if size <= max_bytes:
131 return
132
133 compact_path = log_path.with_name(log_path.name + ".compact")
134 try:
135 with log_path.open("rb") as handle:
136 handle.seek(-max_bytes, os.SEEK_END)
137 tail = handle.read(max_bytes)
138
139 first_newline = tail.find(b"\n")
140 kept = tail[first_newline + 1 :] if first_newline != -1 else b""
141
142 with compact_path.open("wb") as handle:
143 handle.write(kept)
144 compact_path.rename(log_path)
145 except OSError as error:
146 logger.warning("Could not compact oversized supervisor log: %s", error)
147 try:
148 compact_path.unlink(missing_ok=True)
149 except OSError:
150 pass
151
152
153def _configure_supervisor_logging(
154 log_path: Path,
155 level: int,
156 max_bytes: int = SUPERVISOR_LOG_MAX_BYTES,
157 backup_count: int = SUPERVISOR_LOG_BACKUP_COUNT,
158) -> None:
159 logging.getLogger().handlers = []
160 _compact_log_if_oversized(log_path, max_bytes)
161 logging.basicConfig(
162 level=level,
163 handlers=[
164 RotatingFileHandler(
165 log_path,
166 maxBytes=max_bytes,
167 backupCount=backup_count,
168 encoding="utf-8",
169 )
170 ],
171 format="%(asctime)s [supervisor:log] %(levelname)s %(message)s",
172 datefmt="%Y-%m-%dT%H:%M:%S",
173 )
174
175
176_SERVICE_LIFECYCLE_VERBS = {
177 "start",
178 "stop",
179 "restart",
180 "status",
181 "install",
182 "uninstall",
183 "logs",
184}
185
186# Global shutdown flag
187shutdown_requested = False
188_last_sync_tick: float = 0.0
189_last_gate_tick: float = 0.0
190_last_sync_snapshot: "SyncCheckSnapshot | None" = None
191_sync_conflict_shutdown: bool = False
192# Supervisor identity (set in main() once ref is assigned)
193_supervisor_ref: str | None = None
194_supervisor_start: float | None = None
195_parent_death_sigterm_sent = threading.Event()
196
197
198def app_supervised_graceful_budget_s() -> float:
199 """Sum of configured app-supervised graceful shutdown step budgets.
200
201 After the parent-death watcher self-SIGTERMs, the graceful path runs these
202 configured caps in finally-block order: handle_shutdown's managed-child
203 reap, task-queue drain, managed child-stop, and Callosum join.
204
205 The assertion that this budget stays below APP_SUPERVISED_SHUTDOWN_CEILING_S
206 guards that configured step budgets leave room under the hard parent-death
207 backstop. It is not a guarantee that wall time can never exceed this sum.
208 In the common non-D-state case, bounded terminate calls (timeout plus
209 KILL_REAP_GRACE_S) keep the graceful path well under the ceiling. For
210 pathological slow-to-reap children, a step may exceed its nominal cap; the
211 parent-death backstop remains the hard guarantee by sleeping to the ceiling,
212 SIGKILLing every child's process group, then calling os._exit(1).
213 """
214 return (
215 HANDLE_SHUTDOWN_REAP_S
216 + APP_SUPERVISED_TASK_DRAIN_S
217 + APP_SUPERVISED_CHILD_STOP_S
218 + APP_SUPERVISED_CALLOSUM_JOIN_S
219 )
220
221
222def _sd_notify(state: str) -> None:
223 addr = os.environ.get("NOTIFY_SOCKET")
224 if not addr:
225 return
226 if addr.startswith("@"):
227 addr = "\0" + addr[1:]
228 try:
229 with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as s:
230 s.sendto(state.encode(), addr)
231 except OSError as exc:
232 logging.warning("sd_notify failed: %s", exc)
233
234
235def _parent_fd_is_usable(fd: int) -> bool:
236 try:
237 mode = os.fstat(fd).st_mode
238 flags = fcntl.fcntl(fd, fcntl.F_GETFL)
239 except OSError:
240 return False
241 access_mode = flags & os.O_ACCMODE
242 return stat.S_ISFIFO(mode) and access_mode in (os.O_RDONLY, os.O_RDWR)
243
244
245def wait_until_parent_gone(
246 parent_fd: int, *, poll_interval: float = PARENT_DEATH_POLL_INTERVAL_S
247) -> str:
248 if _parent_fd_is_usable(parent_fd):
249 while True:
250 try:
251 data = os.read(parent_fd, 4096)
252 except OSError:
253 return "fd-error"
254 if data == b"":
255 return "eof"
256
257 while True:
258 if os.getppid() == 1:
259 return "orphaned"
260 time.sleep(poll_interval)
261
262
263def enforce_parent_death_shutdown_deadline(
264 reason: str,
265 *,
266 ceiling: float = APP_SUPERVISED_SHUTDOWN_CEILING_S,
267 managed_procs: Iterable[RunnerManagedProcess] | None = None,
268 task_procs: Iterable[RunnerManagedProcess] | None = None,
269 sent_event: "threading.Event | None" = None,
270 kill: Callable[[int, int], None] | None = None,
271 killpg: Callable[[int, int], None] | None = None,
272 getpgid: Callable[[int], int] | None = None,
273 exit_now: Callable[[int], NoReturn] | None = None,
274 sleep: Callable[[float], None] | None = None,
275) -> None:
276 del reason
277 own_pid = os.getpid()
278 own_pgid = os.getpgrp()
279 kill_fn = kill or os.kill
280 killpg_fn = killpg or os.killpg
281 getpgid_fn = getpgid or os.getpgid
282 exit_now_fn = exit_now or os._exit
283 sleep_fn = sleep or time.sleep
284 sigterm_sent = sent_event if sent_event is not None else _parent_death_sigterm_sent
285
286 if not sigterm_sent.is_set():
287 sigterm_sent.set()
288 kill_fn(own_pid, signal.SIGTERM)
289
290 sleep_fn(ceiling)
291
292 procs = managed_procs if managed_procs is not None else _managed_procs
293 if task_procs is not None:
294 task_snapshot = task_procs
295 elif _task_queue is not None:
296 with _task_queue._lock:
297 task_snapshot = list(_task_queue._active.values())
298 else:
299 task_snapshot = []
300
301 def _kill_group(managed: RunnerManagedProcess) -> None:
302 if not managed.is_running():
303 return
304 try:
305 pgid = getpgid_fn(managed.process.pid)
306 except (ProcessLookupError, OSError):
307 logger.exception(
308 "parent-death backstop: could not resolve pgid for %s",
309 managed.name,
310 )
311 return
312
313 if pgid == own_pgid or pgid == own_pid:
314 logger.warning(
315 "parent-death backstop: refusing to signal supervisor's own "
316 "group (pgid=%s) for %s",
317 pgid,
318 managed.name,
319 )
320 return
321
322 try:
323 killpg_fn(pgid, signal.SIGKILL)
324 except Exception:
325 logger.exception(
326 "parent-death backstop: SIGKILL failed for %s", managed.name
327 )
328
329 for managed in procs:
330 try:
331 _kill_group(managed)
332 except Exception:
333 logger.exception(
334 "parent-death backstop: unexpected failure for %s",
335 getattr(managed, "name", managed),
336 )
337 for managed in task_snapshot:
338 try:
339 _kill_group(managed)
340 except Exception:
341 logger.exception(
342 "parent-death backstop: unexpected failure for %s",
343 getattr(managed, "name", managed),
344 )
345
346 exit_now_fn(1)
347
348
349def _parent_death_watcher_main(
350 parent_fd: int,
351 *,
352 poll_interval: float = PARENT_DEATH_POLL_INTERVAL_S,
353 ceiling: float = APP_SUPERVISED_SHUTDOWN_CEILING_S,
354) -> None:
355 reason = wait_until_parent_gone(parent_fd, poll_interval=poll_interval)
356 logger.warning(
357 "parent-death detected (%s); converging to graceful shutdown", reason
358 )
359 enforce_parent_death_shutdown_deadline(reason, ceiling=ceiling)
360
361
362def start_parent_death_watcher(
363 parent_fd: int | None = None,
364 *,
365 poll_interval: float = PARENT_DEATH_POLL_INTERVAL_S,
366 ceiling: float = APP_SUPERVISED_SHUTDOWN_CEILING_S,
367) -> threading.Thread:
368 fd = parent_fd if parent_fd is not None else resolve_parent_fd()
369 thread = threading.Thread(
370 target=_parent_death_watcher_main,
371 args=(fd,),
372 kwargs={"poll_interval": poll_interval, "ceiling": ceiling},
373 name="parent-death-watcher",
374 daemon=True,
375 )
376 thread.start()
377 return thread
378
379
380def _candidate_journal(proc: "psutil.Process") -> Path | None:
381 """Return the resolved SOLSTONE_JOURNAL of ``proc``, or None on any failure.
382
383 Used by the orphan sweep to skip candidates we cannot positively classify
384 as belonging to the caller's journal. Conservative on unknown: any failure
385 to read or parse the env value returns None so the candidate is skipped.
386 """
387 try:
388 env = proc.environ()
389 except (psutil.AccessDenied, psutil.NoSuchProcess, OSError):
390 return None
391 raw = env.get("SOLSTONE_JOURNAL")
392 if not raw:
393 return None
394 try:
395 return Path(raw).resolve()
396 except (OSError, RuntimeError, ValueError):
397 return None
398
399
400# The long-lived journal proctitles set by setproctitle at sol_cli.py
401# (f"{binary}:{cmd}"). setproctitle is in-process and persists until the
402# process exits, so an orphaned service or task child still reports its title
403# via proc.name() after the supervisor dies, which is what lets the sweep find
404# it. The supervisor-owned `llama-server` reports its own bare binary name (no
405# colon prefix) and is included here so the sweep reaps it too.
406# The mlx-vlm server is a Python process, but our launcher sets the same
407# managed proctitle so proc.name() is stable for orphan sweeping.
408_LOCAL_SERVER_PROCTITLES = frozenset(
409 {
410 LOCAL_SERVER_PROCESS_NAME,
411 MLX_SERVER_PROCESS_NAME,
412 }
413)
414
415
416def _is_sweepable_orphan_name(name: str) -> bool:
417 """True if proc.name() identifies a sweepable orphan of this install.
418
419 Any `journal:*` proctitle - managed service or task-queue child - plus the
420 bare local-server binary names. A PPID-1, same-journal `journal:*` process
421 is by definition an orphan of a dead supervisor. `solstone:*`/`sol:*` and a
422 bare `journal` (no colon) are deliberately not matched because they cannot
423 be positively classified as a sub-command of this install.
424 """
425 return name.startswith("journal:") or name in _LOCAL_SERVER_PROCTITLES
426
427
428def _sweep_orphaned_sol_processes(journal: Path, grace: float = 5.0) -> int:
429 journal = journal.resolve()
430 current_user = getpass.getuser()
431 own_pid = os.getpid()
432 targets: list[int] = []
433 for proc in psutil.process_iter(["name", "ppid", "username", "pid"]):
434 try:
435 if not _is_sweepable_orphan_name(proc.name()):
436 continue
437 if proc.ppid() != 1:
438 continue
439 if proc.username() != current_user:
440 continue
441 if proc.pid == own_pid:
442 continue
443 candidate_journal = _candidate_journal(proc)
444 if candidate_journal != journal:
445 continue
446 targets.append(proc.pid)
447 except (psutil.NoSuchProcess, psutil.AccessDenied):
448 continue
449
450 if not targets:
451 return 0
452
453 logger.info(
454 "orphan sweep: terminating %d sol process(es) in journal %s",
455 len(targets),
456 journal,
457 )
458 for pid in targets:
459 logger.debug("orphan sweep: SIGTERM pid=%d", pid)
460 try:
461 os.kill(pid, signal.SIGTERM)
462 except ProcessLookupError:
463 pass
464
465 deadline = time.time() + grace
466 while time.time() < deadline:
467 if not any(psutil.pid_exists(pid) for pid in targets):
468 break
469 time.sleep(0.1)
470
471 survivors = [pid for pid in targets if psutil.pid_exists(pid)]
472 for pid in survivors:
473 logger.debug("orphan sweep: SIGKILL pid=%d", pid)
474 try:
475 os.kill(pid, signal.SIGKILL)
476 except ProcessLookupError:
477 pass
478 return len(targets)
479
480
481class CallosumLogHandler(logging.Handler):
482 """Logging handler that emits log records as callosum ``logs`` tract events.
483
484 Silently drops events on any error — callosum mirroring is best-effort.
485 """
486
487 def __init__(self, conn: CallosumConnection, ref: str):
488 super().__init__()
489 self._conn = conn
490 self._ref = ref
491 self._pid = os.getpid()
492 self._emitting = False
493
494 def emit(self, record: logging.LogRecord) -> None:
495 if self._emitting:
496 return
497 self._emitting = True
498 try:
499 self._conn.emit(
500 "logs",
501 "line",
502 ref=self._ref,
503 name="supervisor",
504 pid=self._pid,
505 stream="log",
506 line=self.format(record),
507 )
508 except Exception:
509 pass
510 finally:
511 self._emitting = False
512
513
514class SupervisorArgumentParser(argparse.ArgumentParser):
515 def error(self, message: str) -> None:
516 mistaken = next(
517 (arg for arg in sys.argv[1:] if arg in _SERVICE_LIFECYCLE_VERBS),
518 None,
519 )
520 if mistaken:
521 self.exit(
522 2,
523 "journal supervisor is the server-launch command (takes a port). "
524 "For lifecycle, use: journal service <verb>. "
525 f"Did you mean: journal service {mistaken} ?\n",
526 )
527 super().error(message)
528
529
530class TaskQueue:
531 """Manages on-demand task execution with per-command serialization.
532
533 Tasks are serialized by command name - only one task per command runs at a time.
534 Additional requests for the same command are queued (deduped by exact cmd match).
535 Multiple callers requesting the same work have their refs coalesced so all get
536 notified when the task completes.
537
538 The lock only protects state mutations, never held during I/O operations.
539 """
540
541 def __init__(self, on_queue_change: callable = None, ready: bool = True):
542 """Initialize task queue.
543
544 Args:
545 on_queue_change: Optional callback(cmd_name, running_ref, queue_entries)
546 called after queue state changes. Called outside lock.
547 """
548 self._running: dict[
549 str, dict
550 ] = {} # command_name -> {"ref": str, "thread": Thread}
551 self._queues: dict[str, list] = {} # command_name -> list of {refs, cmd} dicts
552 self._active: dict[str, RunnerManagedProcess] = {} # ref -> process
553 self._history: deque[dict[str, Any]] = deque(maxlen=100)
554 self._cap_terminated: set[str] = set()
555 self._stopped_ticks: dict[str, int] = {}
556 self._caps: dict[str, int] = {}
557 self._default_cap = DEFAULT_TASK_MAX_RUNTIME
558 self._pending: list[dict] = []
559 self._ready = ready
560 self._lock = threading.Lock()
561 self._on_queue_change = on_queue_change
562
563 @staticmethod
564 def get_command_name(cmd: list[str]) -> str:
565 """Return the canonical queue/log partition for a command."""
566 return _command_partition(cmd)
567
568 def _notify_queue_change(self, cmd_name: str) -> None:
569 """Notify listener of queue state change (called outside lock)."""
570 if not self._on_queue_change:
571 return
572
573 with self._lock:
574 if cmd_name == "pending":
575 queue = list(self._pending)
576 running_ref = None
577 else:
578 queue = list(self._queues.get(cmd_name, []))
579 entry = self._running.get(cmd_name)
580 running_ref = entry["ref"] if entry else None
581
582 self._on_queue_change(cmd_name, running_ref, queue)
583
584 def submit(
585 self,
586 cmd: list[str],
587 ref: str | None = None,
588 day: str | None = None,
589 scheduler_name: str | None = None,
590 ) -> str | None:
591 """Submit a task for execution.
592
593 If no task of this command type is running, starts immediately.
594 Otherwise queues (deduped by exact cmd match, refs coalesced).
595
596 Args:
597 cmd: Command to execute
598 ref: Optional caller-provided ref for tracking
599 day: Optional day override (YYYYMMDD) for log placement
600
601 Returns:
602 ref if task was started/queued, None if already tracked (no change)
603 """
604 ref = ref or str(now_ms())
605 cmd_name = self.get_command_name(cmd)
606
607 with self._lock:
608 if not self._ready:
609 self._pending.append(
610 {
611 "refs": [ref],
612 "cmd": cmd,
613 "day": day,
614 "scheduler_name": scheduler_name,
615 }
616 )
617 should_notify_pending = True
618 else:
619 should_notify_pending = False
620
621 if should_notify_pending:
622 self._notify_queue_change("pending")
623 return ref
624
625 should_notify = False
626 should_start = False
627
628 with self._lock:
629 # Detect stale running state (task thread died without clearing queue)
630 if cmd_name in self._running:
631 stale = self._running[cmd_name]
632 if stale["thread"] is not None and not stale["thread"].is_alive():
633 logging.warning(
634 f"Clearing stale {cmd_name} queue "
635 f"(thread dead, ref={stale['ref']})"
636 )
637 self._running.pop(cmd_name)
638
639 if cmd_name in self._running:
640 # Command already running - queue or coalesce
641 queue = self._queues.setdefault(cmd_name, [])
642 existing = next((q for q in queue if q["cmd"] == cmd), None)
643 if existing:
644 if ref not in existing["refs"]:
645 existing["refs"].append(ref)
646 logging.info(
647 f"Added ref {ref} to queued task {cmd_name} "
648 f"(refs: {len(existing['refs'])})"
649 )
650 should_notify = True
651 else:
652 logging.debug(f"Ref already tracked for queued task: {ref}")
653 return None
654 else:
655 queue.append(
656 {
657 "refs": [ref],
658 "cmd": cmd,
659 "day": day,
660 "scheduler_name": scheduler_name,
661 }
662 )
663 logging.info(
664 f"Queued task {cmd_name}: {' '.join(cmd)} ref={ref} "
665 f"(queue: {len(queue)})"
666 )
667 should_notify = True
668 else:
669 # Not running - mark as running and start
670 # Thread is set to None here; _run_task registers it on entry
671 self._running[cmd_name] = {
672 "ref": ref,
673 "thread": None,
674 "scheduler_name": scheduler_name,
675 }
676 should_start = True
677
678 # Notify outside lock
679 if should_notify:
680 self._notify_queue_change(cmd_name)
681 return ref
682
683 # Start task outside lock
684 if should_start:
685 threading.Thread(
686 target=self._run_task,
687 args=([ref], cmd, cmd_name, day, scheduler_name),
688 daemon=True,
689 ).start()
690 return ref
691
692 return None
693
694 def set_cap(self, cmd_name: str, seconds: int) -> None:
695 """Set a max runtime cap in seconds for a queued command name."""
696 with self._lock:
697 self._caps[cmd_name] = seconds
698
699 def _effective_cap(self, cmd_name: str) -> int:
700 """Resolve the wall-clock cap for a partition: explicit override or default.
701
702 Lock-free: reads only `self._caps` (atomic dict get) and the immutable
703 `self._default_cap`, so it is safe whether or not the caller holds
704 `self._lock`.
705 """
706 return self._caps.get(cmd_name) or self._default_cap
707
708 def get_active_by_cmd_name(self, name: str) -> str | None:
709 """Return the first active ref matching a command name."""
710 with self._lock:
711 for ref, managed in self._active.items():
712 if self.get_command_name(managed.cmd) == name:
713 return ref
714 return None
715
716 def enforce_deadlines(self, now: float) -> None:
717 """Enforce configured task runtime caps without blocking the supervisor tick."""
718 with self._lock:
719 for ref, managed in list(self._active.items()):
720 cmd_name = self.get_command_name(managed.cmd)
721 cap = self._effective_cap(cmd_name)
722
723 elapsed = now - managed.start_time
724 if elapsed <= cap:
725 continue
726
727 elapsed_seconds = int(elapsed)
728 logging.warning(
729 "Task %s (cmd=%s, ref=%s) exceeded max_runtime of %ds "
730 "(elapsed=%ds); terminating",
731 cmd_name,
732 " ".join(managed.cmd),
733 ref,
734 cap,
735 elapsed_seconds,
736 )
737 self._cap_terminated.add(ref)
738 _start_termination_thread(ref, managed, timeout=2.0, reason="cap")
739
740 for ref, managed in list(self._active.items()):
741 if ref in self._cap_terminated:
742 continue
743
744 try:
745 state = psutil.Process(managed.process.pid).status()
746 except (psutil.NoSuchProcess, psutil.AccessDenied):
747 self._stopped_ticks.pop(ref, None)
748 continue
749
750 if state in (psutil.STATUS_STOPPED, psutil.STATUS_TRACING_STOP):
751 ticks = self._stopped_ticks.get(ref, 0) + 1
752 self._stopped_ticks[ref] = ticks
753 if ticks >= STOPPED_TICKS_THRESHOLD:
754 cmd_name = self.get_command_name(managed.cmd)
755 logging.warning(
756 "Task %s (cmd=%s, ref=%s) was stopped (state=%s) "
757 "for %d consecutive ticks; terminating",
758 cmd_name,
759 " ".join(managed.cmd),
760 ref,
761 state,
762 ticks,
763 )
764 self._cap_terminated.add(ref)
765 _start_termination_thread(
766 ref, managed, timeout=2.0, reason="stopped"
767 )
768 self._stopped_ticks.pop(ref, None)
769 else:
770 self._stopped_ticks.pop(ref, None)
771
772 def set_ready(self) -> None:
773 """Allow buffered tasks to start dispatching through the normal queue path."""
774 with self._lock:
775 if self._ready:
776 return
777 self._ready = True
778 pending = list(self._pending)
779 self._pending.clear()
780
781 if pending:
782 self._notify_queue_change("pending")
783 for entry in pending:
784 self.submit(
785 entry["cmd"],
786 ref=entry["refs"][0],
787 day=entry.get("day"),
788 scheduler_name=entry.get("scheduler_name"),
789 )
790
791 def _run_task(
792 self,
793 refs: list[str],
794 cmd: list[str],
795 cmd_name: str,
796 day: str | None = None,
797 scheduler_name: str | None = None,
798 ) -> None:
799 """Execute a task and handle completion.
800
801 Args:
802 refs: List of refs to notify on completion
803 cmd: Command to execute
804 cmd_name: Command name for queue management
805 day: Optional day override (YYYYMMDD) for log placement
806 """
807 # Register this thread for stale-queue detection
808 with self._lock:
809 if cmd_name in self._running and self._running[cmd_name]["ref"] == refs[0]:
810 self._running[cmd_name]["thread"] = threading.current_thread()
811
812 callosum = CallosumConnection()
813 managed = None
814 primary_ref = refs[0]
815 service = cmd_name
816 exit_status = "error"
817 attempt_recorded = False
818
819 try:
820 callosum.start()
821 logging.info(f"Starting task {primary_ref}: {' '.join(cmd)}")
822
823 managed = RunnerManagedProcess.spawn(
824 cmd, ref=primary_ref, callosum=callosum, day=day
825 )
826 with self._lock:
827 self._active[primary_ref] = managed
828 started_at = time.time()
829 record_attempt(cmd, day, primary_ref, started_at=started_at)
830 attempt_recorded = True
831
832 callosum.emit(
833 "supervisor",
834 "started",
835 service=service,
836 pid=managed.pid,
837 ref=primary_ref,
838 )
839
840 exit_code = managed.wait()
841 exit_status = _exit_status_for_code(exit_code)
842
843 for ref in refs:
844 callosum.emit(
845 "supervisor",
846 "stopped",
847 service=service,
848 pid=managed.pid,
849 ref=ref,
850 exit_code=exit_code,
851 )
852
853 if exit_code == 0:
854 logging.info(f"Task {cmd_name} ({primary_ref}) finished successfully")
855 else:
856 logging.warning(
857 f"Task {cmd_name} ({primary_ref}) failed with exit code {exit_code}"
858 )
859
860 except Exception as e:
861 if isinstance(e, subprocess.TimeoutExpired):
862 exit_status = "timeout"
863 logging.exception(
864 f"Task {cmd_name} ({primary_ref}) encountered exception: {e}"
865 )
866 for ref in refs:
867 callosum.emit(
868 "supervisor",
869 "stopped",
870 service=service,
871 pid=managed.pid if managed else 0,
872 ref=ref,
873 exit_code=-1,
874 )
875 finally:
876 try:
877 if managed:
878 managed.cleanup()
879 except Exception:
880 logging.exception(f"Task {cmd_name} ({primary_ref}): cleanup failed")
881 with self._lock:
882 self._active.pop(primary_ref, None)
883 if primary_ref in self._cap_terminated:
884 exit_status = "timeout"
885 self._cap_terminated.discard(primary_ref)
886 self._stopped_ticks.pop(primary_ref, None)
887 ended_at = time.time()
888 self._history.append(
889 {
890 "name": cmd_name,
891 "cmd": list(cmd),
892 "ref": primary_ref,
893 "ended_at": ended_at,
894 "exit_status": exit_status,
895 "scheduler_name": scheduler_name,
896 }
897 )
898 if scheduler_name:
899 try:
900 _record_scheduler_completion(
901 scheduler_name,
902 ended_at=ended_at,
903 exit_status=exit_status,
904 ref=primary_ref,
905 cmd=cmd,
906 )
907 except Exception as exc:
908 logger.warning("scheduler completion writeback failed: %s", exc)
909 if attempt_recorded:
910 try:
911 outcome_result = record_outcome(
912 cmd,
913 day,
914 primary_ref,
915 exit_status=exit_status,
916 ended_at=ended_at,
917 )
918 if outcome_result.entered_backoff:
919 _emit_catchup_backoff(
920 callosum,
921 day=outcome_result.day,
922 attempts=outcome_result.attempts,
923 consecutive=outcome_result.consecutive_non_completion,
924 last_outcome=outcome_result.last_outcome,
925 )
926 except Exception:
927 logging.warning("catchup outcome writeback failed", exc_info=True)
928 try:
929 callosum.stop()
930 except Exception:
931 logging.exception(
932 f"Task {cmd_name} ({primary_ref}): callosum stop failed"
933 )
934 self._process_next(cmd_name)
935
936 def _process_next(self, cmd_name: str) -> None:
937 """Process next queued task after completion."""
938 next_cmd = None
939 refs = None
940 day = None
941 scheduler_name = None
942
943 with self._lock:
944 queue = self._queues.get(cmd_name, [])
945 if queue:
946 entry = queue.pop(0)
947 refs = entry["refs"]
948 next_cmd = entry["cmd"]
949 day = entry.get("day")
950 scheduler_name = entry.get("scheduler_name")
951 # Thread is set to None here; _run_task registers it on entry
952 self._running[cmd_name] = {
953 "ref": refs[0],
954 "thread": None,
955 "scheduler_name": scheduler_name,
956 }
957 logging.info(
958 f"Dequeued task {cmd_name}: {' '.join(next_cmd)} refs={refs} "
959 f"(remaining: {len(queue)})"
960 )
961 else:
962 self._running.pop(cmd_name, None)
963
964 # Notify and spawn outside lock
965 self._notify_queue_change(cmd_name)
966 if next_cmd:
967 threading.Thread(
968 target=self._run_task,
969 args=(refs, next_cmd, cmd_name, day, scheduler_name),
970 daemon=True,
971 ).start()
972
973 def cancel(self, ref: str) -> bool:
974 """Cancel a running task.
975
976 Returns:
977 True if task was found and terminated, False otherwise
978 """
979 if ref not in self._active:
980 logging.warning(f"Cannot cancel task {ref}: not found")
981 return False
982
983 managed = self._active[ref]
984 if not managed.is_running():
985 logging.debug(f"Task {ref} already finished")
986 return False
987
988 logging.info(f"Cancelling task {ref}...")
989 managed.terminate()
990 return True
991
992 def shutdown(self, timeout: float = 10.0) -> int:
993 with self._lock:
994 active = list(self._active.items())
995 if not active:
996 return 0
997
998 def _terminate(item: tuple[str, RunnerManagedProcess]) -> None:
999 ref, managed = item
1000 try:
1001 managed.terminate(timeout=timeout)
1002 except subprocess.TimeoutExpired:
1003 logger.warning(
1004 "task %s did not exit within %ss; KILL sent", ref, timeout
1005 )
1006 except OSError as exc:
1007 logger.warning("task %s terminate raised: %s", ref, exc)
1008
1009 with concurrent.futures.ThreadPoolExecutor(max_workers=len(active)) as executor:
1010 list(executor.map(_terminate, active))
1011 return len(active)
1012
1013 def get_status(self, ref: str) -> dict:
1014 """Get status of a task."""
1015 if ref not in self._active:
1016 return {"status": "not_found"}
1017
1018 managed = self._active[ref]
1019 return {
1020 "status": "running" if managed.is_running() else "finished",
1021 "pid": managed.pid,
1022 "returncode": managed.returncode,
1023 "log_path": str(managed.log_writer.path),
1024 "cmd": managed.cmd,
1025 }
1026
1027 def collect_task_status(self) -> list[dict]:
1028 """Collect status of all running tasks for supervisor status."""
1029 now = time.time()
1030 tasks = []
1031 for ref, managed in self._active.items():
1032 if managed.is_running():
1033 duration = int(now - managed.start_time)
1034 cmd_name = TaskQueue.get_command_name(managed.cmd)
1035 cap = self._effective_cap(cmd_name)
1036 tasks.append(
1037 {
1038 "ref": ref,
1039 "name": cmd_name,
1040 "duration_seconds": duration,
1041 "max_runtime_seconds": cap,
1042 "slow": duration >= cap * SOFT_RUNTIME_FRACTION,
1043 "stuck": duration > cap,
1044 }
1045 )
1046 return tasks
1047
1048 def collect_queue_counts(self) -> dict[str, int]:
1049 """Snapshot per-command queue depths for status reporting."""
1050 with self._lock:
1051 counts = {
1052 cmd_name: len(queue)
1053 for cmd_name, queue in self._queues.items()
1054 if queue
1055 }
1056 if self._pending:
1057 counts["pending"] = len(self._pending)
1058 return counts
1059
1060
1061# Global task queue instance (initialized in main())
1062_task_queue: TaskQueue | None = None
1063
1064# Global supervisor callosum connection for event emissions
1065_supervisor_callosum: CallosumConnection | None = None
1066
1067# Global reference to managed processes for restart control
1068_managed_procs: list[RunnerManagedProcess] = []
1069_SERVICE_STATE: dict[str, dict[str, Any]] = {}
1070_termination_threads: dict[str, threading.Thread] = {}
1071_termination_threads_lock = threading.Lock()
1072_SCHEDULER_JSON_LOCK = threading.Lock()
1073
1074# Global reference to in-process Callosum server
1075_callosum_server: CallosumServer | None = None
1076_callosum_thread: threading.Thread | None = None
1077
1078# Track whether running in remote mode (upload-only, no local processing)
1079_is_remote_mode: bool = False
1080
1081# State for daily processing (tracks day boundary for midnight think trigger)
1082_daily_state = {
1083 "last_day": None, # Track which day we last processed
1084}
1085
1086# State for local provider recovery nudges
1087_recovery_state = {
1088 "local_server_down": False,
1089}
1090
1091# State for local provider wedge detection
1092_wedge_state: dict[str, Any] = {
1093 "providers": OrderedDict(),
1094 "failures": set(),
1095 "cooldown_until": 0.0,
1096 "awaiting_recovery": False,
1097}
1098
1099# Timeout before flushing stale segments (seconds)
1100FLUSH_TIMEOUT = 3600
1101
1102# State for segment flush (close out dangling agent state after inactivity)
1103_flush_state: dict = {
1104 "last_segment_ts": 0.0, # Wall-clock time of last observe.observed event
1105 "day": None, # Day of last observed segment
1106 "segment": None, # Last observed segment key
1107 "flushed": False, # Whether flush has already run for current segment
1108}
1109
1110
1111def _get_journal_path() -> Path:
1112 return Path(get_journal())
1113
1114
1115def is_supervisor_up() -> bool:
1116 """Return True when supervisor.pid and supervisor.start_time identify a live supervisor process for the current journal."""
1117 health_dir = Path(get_journal()) / "health"
1118 pid_path = health_dir / "supervisor.pid"
1119 try:
1120 pid = int(pid_path.read_text().strip())
1121 except FileNotFoundError:
1122 return False
1123 except (OSError, ValueError):
1124 return False
1125
1126 try:
1127 os.kill(pid, 0)
1128 except ProcessLookupError:
1129 return False
1130 except PermissionError:
1131 return False
1132 except OSError:
1133 return False
1134
1135 start_time_path = health_dir / "supervisor.start_time"
1136 try:
1137 recorded_start = float(start_time_path.read_text().strip())
1138 except FileNotFoundError:
1139 return False
1140 except (OSError, ValueError):
1141 return False
1142
1143 try:
1144 create_time = psutil.Process(pid).create_time()
1145 except psutil.NoSuchProcess:
1146 return False
1147 except psutil.Error:
1148 return False
1149
1150 return abs(recorded_start - create_time) <= START_TIME_TOLERANCE_S
1151
1152
1153class RestartPolicy:
1154 """Track restart attempts and compute backoff delays."""
1155
1156 _SCHEDULE = (0, 1, 5)
1157
1158 def __init__(self) -> None:
1159 self.attempts = 0
1160 self.last_start = 0.0
1161
1162 def record_start(self) -> None:
1163 self.last_start = time.time()
1164
1165 def reset_attempts(self) -> None:
1166 self.attempts = 0
1167
1168 def next_delay(self) -> int:
1169 delay = self._SCHEDULE[min(self.attempts, len(self._SCHEDULE) - 1)]
1170 self.attempts += 1
1171 return delay
1172
1173
1174_RESTART_POLICIES: dict[str, RestartPolicy] = {}
1175
1176
1177def describe_exit(returncode: int) -> str:
1178 """Render a process return code, decoding signals for negative codes."""
1179 if returncode >= 0:
1180 return f"exit {returncode}"
1181 try:
1182 name = signal.Signals(-returncode).name
1183 except ValueError:
1184 return f"exit {returncode} / signal {-returncode}"
1185 return f"exit {returncode} / {name}"
1186
1187
1188def _get_restart_policy(name: str) -> RestartPolicy:
1189 return _RESTART_POLICIES.setdefault(name, RestartPolicy())
1190
1191
1192def _launch_process(
1193 name: str,
1194 cmd: list[str],
1195 *,
1196 restart: bool = False,
1197 shutdown_timeout: int = 15,
1198 ref: str | None = None,
1199 env: dict[str, str] | None = None,
1200) -> RunnerManagedProcess:
1201 # NOTE: All child processes should include -v for verbose logging by default.
1202 # This ensures their output is captured in logs for debugging.
1203 """Launch process with automatic output logging and restart policy tracking."""
1204 policy: RestartPolicy | None = None
1205 if restart:
1206 policy = _get_restart_policy(name)
1207
1208 # Generate ref if not provided
1209 ref = ref if ref else str(now_ms())
1210
1211 # Use unified runner to spawn process (share supervisor's callosum)
1212 try:
1213 managed = RunnerManagedProcess.spawn(
1214 cmd, ref=ref, callosum=_supervisor_callosum, env=env
1215 )
1216 except RuntimeError as exc:
1217 logging.error(str(exc))
1218 raise
1219
1220 if policy:
1221 policy.record_start()
1222 _SERVICE_STATE[name] = {
1223 "restart": restart,
1224 "shutdown_timeout": shutdown_timeout,
1225 }
1226
1227 # Emit started event
1228 if _supervisor_callosum:
1229 _supervisor_callosum.emit(
1230 "supervisor",
1231 "started",
1232 service=name,
1233 pid=managed.process.pid,
1234 ref=managed.ref,
1235 )
1236
1237 return managed
1238
1239
1240def _terminate_managed(
1241 managed: RunnerManagedProcess, timeout: float, *, reason: str
1242) -> None:
1243 logger.info("Terminating %s for %s", managed.name, reason)
1244 try:
1245 managed.terminate(timeout=timeout)
1246 except subprocess.TimeoutExpired:
1247 logger.warning(
1248 "%s did not terminate within %.1fs for %s",
1249 managed.name,
1250 timeout,
1251 reason,
1252 )
1253
1254
1255def _start_termination_thread(
1256 key: str, managed: RunnerManagedProcess, timeout: float, reason: str
1257) -> None:
1258 def run() -> None:
1259 try:
1260 _terminate_managed(managed, timeout, reason=reason)
1261 finally:
1262 with _termination_threads_lock:
1263 if _termination_threads.get(key) is threading.current_thread():
1264 _termination_threads.pop(key, None)
1265
1266 with _termination_threads_lock:
1267 existing = _termination_threads.get(key)
1268 if existing and existing.is_alive():
1269 return
1270
1271 thread = threading.Thread(
1272 target=run,
1273 daemon=True,
1274 name=f"terminate-{key}",
1275 )
1276 _termination_threads[key] = thread
1277 thread.start()
1278
1279
1280def _stop_process(
1281 managed: RunnerManagedProcess, *, timeout_cap: float | None = None
1282) -> None:
1283 timeout = _SERVICE_STATE.get(managed.name, {}).get("shutdown_timeout", 15)
1284 if timeout_cap is not None:
1285 timeout = min(timeout, timeout_cap)
1286 _terminate_managed(managed, timeout, reason="shutdown")
1287 managed.cleanup()
1288
1289
1290def _exit_status_for_code(exit_code: int) -> str:
1291 """Map a scheduled task's process exit code to a scheduler status label.
1292
1293 0 -> "ok"; EXIT_EMPTY -> "empty" (a rollup ran over zero inputs - a distinct,
1294 non-error "nothing to do" outcome); any other non-zero code -> "error".
1295 Timeouts are mapped separately by the caller.
1296 """
1297 if exit_code == 0:
1298 return "ok"
1299 if exit_code == EXIT_EMPTY:
1300 return "empty"
1301 return "error"
1302
1303
1304def _record_scheduler_completion(
1305 scheduler_name: str,
1306 *,
1307 ended_at: float,
1308 exit_status: str,
1309 ref: str,
1310 cmd: list[str],
1311) -> None:
1312 health_dir = Path(get_journal()) / "health"
1313 health_dir.mkdir(parents=True, exist_ok=True)
1314 state_path = health_dir / "scheduler.json"
1315 with _SCHEDULER_JSON_LOCK:
1316 try:
1317 with open(state_path, "r", encoding="utf-8") as file:
1318 state = json.load(file)
1319 except FileNotFoundError:
1320 state = {}
1321 except (json.JSONDecodeError, OSError) as exc:
1322 logger.warning(
1323 "Failed to load scheduler state for completion write: %s", exc
1324 )
1325 state = {}
1326
1327 current = state.get(scheduler_name)
1328 if not isinstance(current, dict):
1329 current = {}
1330 current.update(
1331 {
1332 "last_run": ended_at,
1333 "last_status": exit_status,
1334 "last_ref": ref,
1335 }
1336 )
1337 state[scheduler_name] = current
1338
1339 fd, tmp_path = tempfile.mkstemp(
1340 dir=health_dir, suffix=".tmp", prefix=".scheduler_"
1341 )
1342 tmp_file = Path(tmp_path)
1343 try:
1344 with open(fd, "w", encoding="utf-8") as file:
1345 json.dump(state, file, indent=2)
1346 tmp_file.replace(state_path)
1347 except BaseException:
1348 tmp_file.unlink(missing_ok=True)
1349 raise
1350
1351
1352def _emit_catchup_backoff(
1353 callosum,
1354 *,
1355 day: str | None,
1356 attempts: int,
1357 consecutive: int,
1358 last_outcome: str,
1359) -> None:
1360 if callosum is None:
1361 return
1362 message = f"day {day} stuck after {attempts} attempts, last outcome {last_outcome}"
1363 try:
1364 callosum.emit(
1365 "storage",
1366 "warning",
1367 level="warning",
1368 type="catchup_backoff",
1369 message=message,
1370 current=consecutive,
1371 threshold=STUCK_THRESHOLD,
1372 )
1373 callosum.emit(
1374 "notification",
1375 "show",
1376 title="Catchup stuck",
1377 message=message,
1378 icon="⚠️",
1379 action="/app/health",
1380 )
1381 except Exception:
1382 logging.warning(
1383 "Failed to emit catchup backoff notification for %s", day, exc_info=True
1384 )
1385
1386
1387def _emit_queue_event(cmd_name: str, running_ref: str, queue: list) -> None:
1388 """Emit supervisor.queue event with current queue state for a command.
1389
1390 This is the callback passed to TaskQueue for queue change notifications.
1391 """
1392 if not _supervisor_callosum:
1393 return
1394
1395 _supervisor_callosum.emit(
1396 "supervisor",
1397 "queue",
1398 command=cmd_name,
1399 running=running_ref,
1400 queued=len(queue),
1401 queue=queue,
1402 )
1403
1404
1405def _handle_task_request(message: dict) -> None:
1406 """Handle incoming task request from Callosum."""
1407 if message.get("tract") != "supervisor" or message.get("event") != "request":
1408 return
1409
1410 cmd = message.get("cmd")
1411 if not cmd:
1412 logging.error(f"Invalid task request: missing cmd: {message}")
1413 return
1414
1415 ref = message.get("ref") or str(now_ms())
1416 day = message.get("day")
1417 scheduler_name = message.get("scheduler_name")
1418 if _task_queue:
1419 cmd_name = TaskQueue.get_command_name(cmd)
1420 active_ref = _task_queue.get_active_by_cmd_name(cmd_name)
1421 if active_ref:
1422 with _task_queue._lock:
1423 managed = _task_queue._active.get(active_ref)
1424 cap = _task_queue._effective_cap(cmd_name)
1425 runtime = time.time() - managed.start_time if managed else 0
1426 reason = "wedged" if runtime > 2 * cap else "still_running"
1427 if _supervisor_callosum:
1428 _supervisor_callosum.emit(
1429 "supervisor",
1430 "skipped",
1431 reason=reason,
1432 ref=ref,
1433 active_ref=active_ref,
1434 cmd=cmd,
1435 scheduler_name=scheduler_name,
1436 )
1437 return
1438 _task_queue.submit(cmd, ref, day=day, scheduler_name=scheduler_name)
1439
1440
1441def _restart_service(service: str) -> bool:
1442 """Terminate a managed service to trigger graceful restart.
1443
1444 Returns True if the service was found and running, False if not found
1445 or already exited.
1446 """
1447 for proc in _managed_procs:
1448 if proc.name == service:
1449 if proc.process.poll() is not None:
1450 logging.debug(
1451 f"Ignoring restart for {service}: already exited, awaiting auto-restart"
1452 )
1453 return False
1454
1455 state = _SERVICE_STATE.setdefault(service, {})
1456 state["restart"] = True
1457 timeout = state.get("shutdown_timeout", 15)
1458
1459 logging.info(f"Restart requested for {service}, terminating...")
1460
1461 if _supervisor_callosum:
1462 _supervisor_callosum.emit(
1463 "supervisor",
1464 "restarting",
1465 service=service,
1466 pid=proc.process.pid,
1467 ref=proc.ref,
1468 )
1469
1470 _start_termination_thread(service, proc, timeout=timeout, reason="restart")
1471 return True
1472
1473 logging.warning(f"Cannot restart {service}: not found in managed processes")
1474 return False
1475
1476
1477def _handle_supervisor_request(message: dict) -> None:
1478 """Handle incoming supervisor control messages."""
1479 if message.get("tract") != "supervisor" or message.get("event") != "restart":
1480 return
1481
1482 service = message.get("service")
1483 if not service:
1484 logging.error("Invalid restart request: missing service")
1485 return
1486 if service == "supervisor":
1487 logging.debug("Ignoring restart request for supervisor itself")
1488 return
1489
1490 _restart_service(service)
1491
1492
1493def _handle_supervisor_drain(message: dict) -> None:
1494 """Handle incoming supervisor catchup drain requests."""
1495 if message.get("tract") != "supervisor" or message.get("event") != "drain":
1496 return
1497 if _is_remote_mode:
1498 return
1499
1500 day = message.get("day")
1501 if day:
1502 run_catchup_drain(force_days={day})
1503 else:
1504 run_catchup_drain()
1505
1506
1507def _handle_supervisor_start_local(message: dict) -> None:
1508 """Handle incoming local server start requests."""
1509 if message.get("tract") != "supervisor" or message.get("event") != "start_local":
1510 return
1511 if _is_remote_mode:
1512 return
1513 from solstone.think.providers.local_endpoint import resolve_local_endpoint
1514
1515 if not resolve_local_endpoint().is_bundled:
1516 return
1517
1518 for proc in _managed_procs:
1519 if proc.name in _LOCAL_SERVER_PROCTITLES and proc.is_running():
1520 logging.info("local server already running; ignoring start_local request")
1521 return
1522
1523 proc = start_local_server()
1524 if proc is not None:
1525 _managed_procs.append(proc)
1526 logging.info("started local server from start_local request")
1527
1528
1529def _handle_cortex_outcome(message: dict) -> None:
1530 """Recycle a wedged local model server after sustained generation failures."""
1531 if message.get("tract") != "cortex":
1532 return
1533 event = message.get("event")
1534 if event not in {"start", "finish", "error"}:
1535 return
1536 if _is_remote_mode:
1537 return
1538
1539 use_id = message.get("use_id")
1540 if not use_id:
1541 return
1542
1543 if event == "start":
1544 providers = _wedge_state["providers"]
1545 providers[use_id] = message.get("provider")
1546 while len(providers) > LOCAL_WEDGE_PROVIDER_MAP_CAP:
1547 providers.popitem(last=False)
1548 return
1549
1550 provider = _wedge_state["providers"].get(use_id)
1551 if provider != "local":
1552 return
1553 from solstone.think.providers.local_endpoint import resolve_local_endpoint
1554
1555 if not resolve_local_endpoint().is_bundled:
1556 return
1557
1558 if time.monotonic() < _wedge_state["cooldown_until"]:
1559 return
1560
1561 failures = _wedge_state["failures"]
1562 if event == "finish":
1563 if _wedge_state["awaiting_recovery"]:
1564 logging.info("local server wedge: recovered after recycle")
1565 _wedge_state["awaiting_recovery"] = False
1566 failures.clear()
1567 return
1568
1569 if message.get("reason_code") != "provider_unavailable":
1570 return
1571
1572 failures.add(use_id)
1573 if len(failures) < LOCAL_WEDGE_THRESHOLD:
1574 return
1575
1576 logging.warning(
1577 "local server wedge: declared after %d local provider_unavailable failures "
1578 "(use_ids=%s)",
1579 len(failures),
1580 sorted(failures),
1581 )
1582 port = read_service_port("local")
1583 if port is None:
1584 logging.warning(
1585 "local server wedge: recycle deferred; local service port unavailable"
1586 )
1587 failures.clear()
1588 return
1589
1590 from solstone.think.providers import local_server
1591
1592 state, _ = local_server._probe_health(port)
1593 if state != local_server.STATE_READY:
1594 logging.warning(
1595 "local server wedge: recycle deferred; health state=%s",
1596 state,
1597 )
1598 failures.clear()
1599 return
1600
1601 proctitle = (
1602 MLX_SERVER_PROCESS_NAME
1603 if sys.platform == "darwin"
1604 else LOCAL_SERVER_PROCESS_NAME
1605 )
1606 if _restart_service(proctitle):
1607 logging.warning("local server wedge: recycling %s", proctitle)
1608 failures.clear()
1609 _wedge_state["awaiting_recovery"] = True
1610 _wedge_state["cooldown_until"] = time.monotonic() + LOCAL_WEDGE_RECYCLE_GRACE_S
1611 else:
1612 logging.warning("local server wedge: recycle deferred; service not running")
1613 failures.clear()
1614
1615
1616def get_task_status(ref: str) -> dict:
1617 """Get status of a task.
1618
1619 Args:
1620 ref: Task correlation ID
1621
1622 Returns:
1623 Dict with status info, or {"status": "not_found"} if task doesn't exist
1624 """
1625 if _task_queue:
1626 return _task_queue.get_status(ref)
1627 return {"status": "not_found"}
1628
1629
1630def collect_status(procs: list[RunnerManagedProcess]) -> dict:
1631 """Collect current supervisor status for broadcasting."""
1632 now = time.time()
1633
1634 # Running services
1635 services = []
1636 running_names = set()
1637 for proc in procs:
1638 if proc.process.poll() is None: # Still running
1639 policy = _get_restart_policy(proc.name)
1640 uptime = int(now - policy.last_start) if policy.last_start else 0
1641 services.append(
1642 {
1643 "name": proc.name,
1644 "ref": proc.ref,
1645 "pid": proc.process.pid,
1646 "uptime_seconds": uptime,
1647 }
1648 )
1649 running_names.add(proc.name)
1650
1651 # Prepend supervisor itself
1652 if _supervisor_ref and _supervisor_start:
1653 services.insert(
1654 0,
1655 {
1656 "name": "supervisor",
1657 "ref": _supervisor_ref,
1658 "pid": os.getpid(),
1659 "uptime_seconds": int(now - _supervisor_start),
1660 },
1661 )
1662
1663 # Crashed services (in restart backoff)
1664 crashed = []
1665 for name, policy in _RESTART_POLICIES.items():
1666 if name not in running_names and policy.attempts > 0:
1667 crashed.append(
1668 {
1669 "name": name,
1670 "restart_attempts": policy.attempts,
1671 }
1672 )
1673
1674 # Running tasks
1675 tasks = _task_queue.collect_task_status() if _task_queue else []
1676 queues = _task_queue.collect_queue_counts() if _task_queue else {}
1677
1678 # Scheduled tasks
1679 schedules = scheduler.collect_status()
1680 # Connected callosum clients
1681 callosum_clients = _callosum_server.client_count() if _callosum_server else 0
1682
1683 return {
1684 "services": services,
1685 "crashed": crashed,
1686 "tasks": tasks,
1687 "queues": queues,
1688 "stale_heartbeats": [],
1689 "schedules": schedules,
1690 "callosum_clients": callosum_clients,
1691 }
1692
1693
1694def start_sense() -> RunnerManagedProcess:
1695 """Launch journal sense with output logging."""
1696 return _launch_process("sense", ["journal", "sense", "-v"], restart=True)
1697
1698
1699def _start_mlx_local_server() -> RunnerManagedProcess | None:
1700 """Launch the supervisor-owned mlx-vlm server when artifacts are present."""
1701 from solstone.think.providers import local_server, mlx_install
1702
1703 readiness = mlx_install.inspect_readiness()
1704 readiness_keys = (
1705 "platform_supported",
1706 "package_available",
1707 "ram_sufficient",
1708 "model_installed",
1709 )
1710 if not all(readiness.get(key) for key in readiness_keys):
1711 logging.info(
1712 "MLX local model not ready; skipping mlx-vlm server startup: %s",
1713 {key: readiness.get(key) for key in readiness_keys},
1714 )
1715 return None
1716
1717 runtime_dir = readiness["runtime_dir"]
1718 model_id = readiness["model_id"]
1719 port = find_available_port()
1720 write_service_port("local", port)
1721 script_path = str(Path(sys.executable).with_name(MLX_SERVER_PROCESS_NAME))
1722 cmd = [
1723 script_path,
1724 "--host",
1725 "127.0.0.1",
1726 "--port",
1727 str(port),
1728 "--model",
1729 str(runtime_dir),
1730 ]
1731 if "0.0.0.0" in cmd:
1732 raise RuntimeError("Local server may not bind 0.0.0.0.")
1733
1734 logging.info("Starting mlx-vlm server for %s from %s", model_id, runtime_dir)
1735 managed = _launch_process(MLX_SERVER_PROCESS_NAME, cmd, restart=True)
1736 print(f" {LOCAL_MODEL_WARMING_UP_COPY}", flush=True)
1737
1738 deadline = time.monotonic() + LOCAL_SERVER_READY_TIMEOUT_S
1739 while time.monotonic() < deadline:
1740 if managed.process.poll() is not None:
1741 logging.warning(
1742 "mlx-vlm server exited during warmup with code %s",
1743 managed.process.returncode,
1744 )
1745 return managed
1746 state, error = local_server._probe_health(port)
1747 if state == local_server.STATE_READY:
1748 logging.info("mlx-vlm server ready on port %s", port)
1749 return managed
1750 if state == local_server.STATE_FAILED and error:
1751 logging.debug("mlx-vlm server health probe failed during warmup: %s", error)
1752 time.sleep(LOCAL_SERVER_HEALTH_POLL_INTERVAL_S)
1753
1754 logging.warning(
1755 "mlx-vlm server did not become ready within %.0fs; continuing startup",
1756 LOCAL_SERVER_READY_TIMEOUT_S,
1757 )
1758 return managed
1759
1760
1761def _format_vulkan_devices(devices: list[Any], local_vulkan: Any) -> str:
1762 if not devices:
1763 return "none"
1764 return "; ".join(
1765 (
1766 f"raw_index={device.index} name={device.name!r} "
1767 f"type={local_vulkan.classify(device)} vram_mib={device.vram_mib}"
1768 )
1769 for device in devices
1770 )
1771
1772
1773def _gpu_unavailable_reason(devices: list[Any], override: int | None) -> str:
1774 if not devices:
1775 return "no Vulkan devices enumerated"
1776 if override is not None:
1777 return f"Vulkan override raw index {override} is not an available hardware GPU"
1778 return "only non-hardware or software Vulkan devices were enumerated"
1779
1780
1781def _log_context_assertion(
1782 tier: Any, n_ctx: int | None, total_slots: int | None
1783) -> None:
1784 if n_ctx is None:
1785 logging.info(
1786 "llama-server context assertion skipped: n_ctx unavailable from /props"
1787 )
1788 else:
1789 expected = {tier.context_tokens, tier.context_tokens * tier.parallel_slots}
1790 if n_ctx in expected:
1791 logging.info(
1792 "llama-server context OK: intended -c=%d parallel=%d actual n_ctx=%d",
1793 tier.context_tokens,
1794 tier.parallel_slots,
1795 n_ctx,
1796 )
1797 else:
1798 logging.warning(
1799 "llama-server context MISMATCH: intended -c=%d parallel=%d "
1800 "actual n_ctx=%d",
1801 tier.context_tokens,
1802 tier.parallel_slots,
1803 n_ctx,
1804 )
1805
1806 if isinstance(total_slots, int):
1807 if total_slots == tier.parallel_slots:
1808 logging.info("llama-server slots OK: %d", total_slots)
1809 else:
1810 logging.warning(
1811 "llama-server slots MISMATCH: intended=%d actual=%d",
1812 tier.parallel_slots,
1813 total_slots,
1814 )
1815 else:
1816 logging.info("llama-server slot count not reported; skipped")
1817
1818
1819def start_local_server() -> RunnerManagedProcess | None:
1820 """Launch the supervisor-owned local llama-server when artifacts are present."""
1821 from solstone.think.providers.local_endpoint import resolve_local_endpoint
1822
1823 if not resolve_local_endpoint().is_bundled:
1824 return None
1825
1826 if sys.platform == "darwin":
1827 return _start_mlx_local_server()
1828
1829 from solstone.think.providers import local_install, local_server, local_vulkan
1830
1831 try:
1832 binary_path, gguf_path, mmproj_path = local_install.ensure_artifacts_installed(
1833 LOCAL_MODEL
1834 )
1835 # Defense in depth: refuse to launch a gguf/mmproj pair that does not
1836 # belong to the selected model, even if readiness ever regresses. A
1837 # mixed pair (e.g. a stale gguf from a prior model + the current mmproj)
1838 # aborts llama-server with an n_embd mismatch, so skip startup instead.
1839 expected_dir = local_install.model_dir(LOCAL_MODEL)
1840 if gguf_path.parent != expected_dir or (
1841 mmproj_path is not None and mmproj_path.parent != expected_dir
1842 ):
1843 raise RuntimeError(
1844 f"local model artifacts are not under {expected_dir} "
1845 f"(gguf={gguf_path}, mmproj={mmproj_path}); refusing to launch "
1846 "a mismatched gguf/mmproj pair"
1847 )
1848 except Exception as exc:
1849 logging.info("Local model not ready; skipping llama-server startup: %s", exc)
1850 return None
1851
1852 devices = local_vulkan.detect_gpus()
1853 override = local_install.gpu_device_override()
1854 selected = local_vulkan.select_device(devices, override_index=override)
1855 logging.info(
1856 "Vulkan GPU probe: devices=%s; selected=%s",
1857 _format_vulkan_devices(devices, local_vulkan),
1858 (
1859 f"raw_index={selected.index} name={selected.name!r} "
1860 f"type={local_vulkan.classify(selected)}"
1861 if selected is not None
1862 else "none"
1863 ),
1864 )
1865 if selected is None:
1866 reason = _gpu_unavailable_reason(devices, override)
1867 logging.info("gpu_unavailable: skipping llama-server startup: %s", reason)
1868 return None
1869
1870 port = find_available_port()
1871 write_service_port("local", port)
1872 tier = local_server.select_server_tier(selected.vram_mib)
1873 local_server.write_local_context_window(tier.context_tokens)
1874 logging.info(
1875 "local server tier=%s context=%d parallel=%d cache=%d MiB (vram=%d MiB)",
1876 tier.name,
1877 tier.context_tokens,
1878 tier.parallel_slots,
1879 tier.prompt_cache_mib,
1880 selected.vram_mib,
1881 )
1882 cmd = [
1883 str(binary_path),
1884 "-m",
1885 str(gguf_path),
1886 "--alias",
1887 LOCAL_MODEL,
1888 "--host",
1889 "127.0.0.1",
1890 "--port",
1891 str(port),
1892 "--jinja",
1893 "--n-gpu-layers",
1894 "999",
1895 "-c",
1896 str(tier.context_tokens),
1897 "--parallel",
1898 str(tier.parallel_slots),
1899 "--kv-unified",
1900 "--cache-ram",
1901 str(tier.prompt_cache_mib),
1902 "--no-context-shift",
1903 "--device",
1904 "Vulkan0",
1905 ]
1906 if mmproj_path is not None:
1907 cmd.extend(["--mmproj", str(mmproj_path)])
1908 if "0.0.0.0" in cmd:
1909 raise RuntimeError("Local server may not bind 0.0.0.0.")
1910
1911 env = os.environ | {"GGML_VK_VISIBLE_DEVICES": str(selected.index)}
1912 vram_before_mib = local_vulkan.device_local_used_mib(selected.index)
1913 managed = _launch_process(LOCAL_SERVER_PROCESS_NAME, cmd, restart=True, env=env)
1914 print(f" {LOCAL_MODEL_WARMING_UP_COPY}", flush=True)
1915
1916 def fail_local_server_launch(reason: str) -> None:
1917 logging.warning("local server launch failed: %s", reason)
1918 timeout = _SERVICE_STATE.get(managed.name, {}).get("shutdown_timeout", 15)
1919 _SERVICE_STATE.pop(managed.name, None)
1920 _terminate_managed(
1921 managed,
1922 timeout,
1923 reason="local server launch failed",
1924 )
1925 managed.cleanup()
1926
1927 deadline = time.monotonic() + LOCAL_SERVER_READY_TIMEOUT_S
1928 while time.monotonic() < deadline:
1929 if managed.process.poll() is not None:
1930 fail_local_server_launch(
1931 f"llama-server exited during warmup with code "
1932 f"{managed.process.returncode}"
1933 )
1934 return None
1935 state, error = local_server._probe_health(port)
1936 if state == local_server.STATE_READY:
1937 vram_after_mib = local_vulkan.device_local_used_mib(selected.index)
1938 if vram_before_mib is not None and vram_after_mib is not None:
1939 logging.info(
1940 "local GPU: %s — VRAM used %+d MiB after model load (%d -> %d MiB)",
1941 selected.name,
1942 vram_after_mib - vram_before_mib,
1943 vram_before_mib,
1944 vram_after_mib,
1945 )
1946 else:
1947 logging.info(
1948 "local GPU: %s — VRAM-usage delta unavailable "
1949 "(VK_EXT_memory_budget not reported)",
1950 selected.name,
1951 )
1952 props = local_server.fetch_props(port)
1953 n_ctx = local_server._extract_n_ctx(props) if props is not None else None
1954 total_slots = props.get("total_slots") if isinstance(props, dict) else None
1955 _log_context_assertion(tier, n_ctx, total_slots)
1956 logging.info("llama-server ready on port %s", port)
1957 return managed
1958 if state == local_server.STATE_FAILED and error:
1959 logging.debug("llama-server health probe failed during warmup: %s", error)
1960 time.sleep(LOCAL_SERVER_HEALTH_POLL_INTERVAL_S)
1961
1962 logging.warning(
1963 "llama-server did not become ready within %.0fs; continuing startup",
1964 LOCAL_SERVER_READY_TIMEOUT_S,
1965 )
1966 return managed
1967
1968
1969def start_callosum_in_process() -> CallosumServer:
1970 """Start Callosum message bus server in-process.
1971
1972 Runs the server in a background thread and waits for socket to be ready.
1973
1974 Returns:
1975 CallosumServer instance
1976 """
1977 global _callosum_server, _callosum_thread
1978
1979 server = CallosumServer()
1980 _callosum_server = server
1981
1982 # Pre-delete stale socket to avoid race condition where the ready check
1983 # passes due to an old socket file before the server thread deletes it
1984 socket_path = server.socket_path
1985 socket_path.parent.mkdir(parents=True, exist_ok=True)
1986 if socket_path.exists():
1987 socket_path.unlink()
1988
1989 # Start server in background thread (server.start() is blocking)
1990 thread = threading.Thread(target=server.start, daemon=False, name="callosum-server")
1991 thread.start()
1992 _callosum_thread = thread
1993
1994 # Wait for socket to be ready (with timeout)
1995 for _ in range(50): # Wait up to 500ms
1996 if socket_path.exists():
1997 logging.info(f"Callosum server started on {socket_path}")
1998 return server
1999 time.sleep(0.01)
2000
2001 raise RuntimeError("Callosum server failed to create socket within 500ms")
2002
2003
2004def wait_for_convey_ready(
2005 convey_mp, *, timeout: float = CONVEY_READY_WINDOW_SECONDS, interval: float = 0.1
2006) -> bool:
2007 """Poll until Convey accepts TCP connections, or fail fast on death/timeout."""
2008 start = time.monotonic()
2009 deadline = start + timeout
2010 while time.monotonic() < deadline:
2011 rc = convey_mp.process.poll()
2012 if rc is not None:
2013 logging.error(
2014 "Convey process exited during startup (rc=%d); continuing into supervise loop",
2015 rc,
2016 )
2017 return False
2018 if is_solstone_up(timeout=0.1):
2019 logging.info("Convey ready after %.1fs", time.monotonic() - start)
2020 return True
2021 time.sleep(interval)
2022 alive = convey_mp.process.poll() is None
2023 logging.error(
2024 "Convey not ready after %.1fs (port=%s, pid alive=%s); continuing into supervise loop",
2025 time.monotonic() - start,
2026 read_service_port("convey"),
2027 alive,
2028 )
2029 return False
2030
2031
2032def stop_callosum_in_process(join_timeout: float = 5.0) -> None:
2033 """Stop the in-process Callosum server."""
2034 global _callosum_server, _callosum_thread
2035
2036 if _callosum_server:
2037 logging.info("Stopping Callosum server...")
2038 _callosum_server.stop()
2039
2040 if _callosum_thread:
2041 _callosum_thread.join(timeout=join_timeout)
2042 if _callosum_thread.is_alive():
2043 logging.warning("Callosum server thread did not stop cleanly")
2044
2045 _callosum_server = None
2046 _callosum_thread = None
2047
2048
2049def start_cortex_server() -> RunnerManagedProcess:
2050 """Launch the Cortex WebSocket API server."""
2051 cmd = ["journal", "cortex", "-v"]
2052 return _launch_process("cortex", cmd, restart=True)
2053
2054
2055def start_spl_service() -> RunnerManagedProcess:
2056 """Launch the spl tunnel service."""
2057 cmd = ["journal", "spl", "-v"]
2058 return _launch_process("spl", cmd, restart=True)
2059
2060
2061def start_convey_server(
2062 verbose: bool, debug: bool = False, port: int = 0
2063) -> tuple[RunnerManagedProcess, int]:
2064 """Launch the Convey web application with optional verbose and debug logging.
2065
2066 Returns:
2067 Tuple of (RunnerManagedProcess, resolved_port) where resolved_port is the
2068 actual port being used (auto-selected if port was 0).
2069 """
2070 # Resolve port 0 to an available port before launching
2071 resolved_port = port if port != 0 else find_available_port()
2072
2073 cmd = ["journal", "convey", "--port", str(resolved_port)]
2074 if debug:
2075 cmd.append("-d")
2076 elif verbose:
2077 cmd.append("-v")
2078 return _launch_process("convey", cmd, restart=True), resolved_port
2079
2080
2081def check_runner_exits(
2082 procs: list[RunnerManagedProcess],
2083) -> list[RunnerManagedProcess]:
2084 """Return managed processes that have exited."""
2085
2086 exited: list[RunnerManagedProcess] = []
2087 for managed in procs:
2088 if managed.process.poll() is not None:
2089 exited.append(managed)
2090 return exited
2091
2092
2093async def handle_runner_exits(procs: list[RunnerManagedProcess]) -> None:
2094 """Check for and handle exited processes with restart policy."""
2095 exited = check_runner_exits(procs)
2096 if not exited:
2097 return
2098
2099 exited_names = [managed.name for managed in exited]
2100
2101 # Check if all exits are tempfail (session not ready)
2102 all_tempfail = all(m.process.returncode == EXIT_TEMPFAIL for m in exited)
2103
2104 if all_tempfail:
2105 logging.info("Runner waiting for session: %s", ", ".join(sorted(exited_names)))
2106 else:
2107 parts = []
2108 for m in sorted(exited, key=lambda managed: managed.name):
2109 policy = _get_restart_policy(m.name)
2110 if policy.last_start:
2111 uptime = f"up {time.time() - policy.last_start:.1f}s"
2112 else:
2113 uptime = "up unknown"
2114 parts.append(f"{m.name} ({describe_exit(m.process.returncode)}, {uptime})")
2115 msg = f"Runner process exited: {', '.join(parts)}"
2116 logging.error(msg)
2117
2118 for managed in exited:
2119 returncode = managed.process.returncode
2120 is_tempfail = returncode == EXIT_TEMPFAIL
2121 logging.info("%s exited with code %s", managed.name, returncode)
2122
2123 # Emit stopped event
2124 if _supervisor_callosum:
2125 _supervisor_callosum.emit(
2126 "supervisor",
2127 "stopped",
2128 service=managed.name,
2129 pid=managed.process.pid,
2130 ref=managed.ref,
2131 exit_code=returncode,
2132 )
2133
2134 # Remove from procs list
2135 try:
2136 procs.remove(managed)
2137 except ValueError:
2138 pass
2139
2140 managed.cleanup()
2141
2142 # Handle restart if needed
2143 restart = _SERVICE_STATE.get(managed.name, {}).get("restart", False)
2144 if restart and not shutdown_requested:
2145 # Tempfail: use fixed longer delay, don't burn through backoff
2146 if is_tempfail:
2147 delay = TEMPFAIL_DELAY
2148 else:
2149 policy = _get_restart_policy(managed.name)
2150 uptime = time.time() - policy.last_start if policy.last_start else 0
2151 if uptime >= 60:
2152 policy.reset_attempts()
2153 delay = policy.next_delay()
2154 if delay:
2155 logging.info("Waiting %ss before restarting %s", delay, managed.name)
2156 for _ in range(delay):
2157 if shutdown_requested:
2158 break
2159 await asyncio.sleep(1)
2160 if shutdown_requested:
2161 continue
2162 logging.info("Restarting %s...", managed.name)
2163 try:
2164 state = _SERVICE_STATE.get(managed.name, {})
2165 new_proc = _launch_process(
2166 managed.name,
2167 managed.cmd,
2168 restart=True,
2169 shutdown_timeout=state.get("shutdown_timeout", 15),
2170 )
2171 except Exception as exc:
2172 logging.exception("Failed to restart %s: %s", managed.name, exc)
2173 continue
2174
2175 procs.append(new_proc)
2176 if managed.name in _LOCAL_SERVER_PROCTITLES:
2177 _recovery_state["local_server_down"] = True
2178 logging.info("Restarted %s after exit code %s", managed.name, returncode)
2179 else:
2180 logging.info("Not restarting %s", managed.name)
2181
2182
2183def _nudge_catchup_drain() -> None:
2184 """Ask the supervisor loopback path to drain pending catchup work."""
2185 if _supervisor_callosum is None:
2186 logging.warning("Cannot nudge catchup drain: supervisor callosum unavailable")
2187 return
2188
2189 try:
2190 _supervisor_callosum.emit("supervisor", "drain")
2191 except Exception as exc:
2192 logging.warning("Cannot nudge catchup drain: %s", exc)
2193
2194
2195async def _check_local_server_recovery() -> None:
2196 """Detect a recovered local server after supervisor-managed relaunch."""
2197 if _is_remote_mode or not _recovery_state["local_server_down"]:
2198 return
2199 from solstone.think.providers.local_endpoint import resolve_local_endpoint
2200
2201 if not resolve_local_endpoint().is_bundled:
2202 return
2203
2204 port = read_service_port("local")
2205 if port is None:
2206 return
2207
2208 from solstone.think.providers import local_server
2209
2210 state, _ = await asyncio.to_thread(local_server._probe_health, port)
2211 if state != local_server.STATE_READY:
2212 return
2213
2214 _recovery_state["local_server_down"] = False
2215 _nudge_catchup_drain()
2216
2217
2218def run_catchup_drain(
2219 force_days: Iterable[str] | None = None,
2220 *,
2221 exclude: set[str] | None = None,
2222) -> list[str]:
2223 """Submit catchup daily think tasks for pending, eligible days."""
2224 all_updated = updated_days(exclude=exclude)
2225
2226 def _eligible(day: str) -> bool:
2227 try:
2228 return day_eligible_to_drain(day, KIND_DAILY_CATCHUP)
2229 except Exception:
2230 logging.warning(
2231 "Catchup eligibility check failed for %s; treating as eligible",
2232 day,
2233 )
2234 return True
2235
2236 eligible_natural = [day for day in all_updated if _eligible(day)]
2237 # AC3: force uses the same backoff gate. AC8: cap natural days after
2238 # eligibility; forced eligible days keep importer single-day intent.
2239 freshest = eligible_natural[-MAX_UPDATED_CATCHUP:]
2240 forced_eligible = [day for day in (force_days or []) if _eligible(day)]
2241 merged = set(freshest) | set(forced_eligible)
2242 if not merged:
2243 logging.info("no eligible days to process")
2244 return []
2245
2246 if _task_queue is None:
2247 logging.warning("No task queue available for catchup drain: %s", sorted(merged))
2248 return []
2249
2250 days = sorted(merged)
2251 logging.info("Queuing catchup drain for %d day(s): %s", len(days), days)
2252 for day_str in days:
2253 cmd = ["journal", "think", "-v", "--day", day_str]
2254 _task_queue.submit(cmd, day=day_str)
2255 return days
2256
2257
2258def _startup_catchup_drain() -> None:
2259 try:
2260 transitions = reconcile_interrupted_attempts()
2261 for transition in transitions:
2262 _emit_catchup_backoff(
2263 _supervisor_callosum,
2264 day=transition.day,
2265 attempts=transition.attempts,
2266 consecutive=transition.consecutive_non_completion,
2267 last_outcome=transition.last_outcome,
2268 )
2269 except Exception:
2270 logging.warning("Catchup reconciliation failed", exc_info=True)
2271 run_catchup_drain()
2272
2273
2274def handle_daily_tasks() -> None:
2275 """Check for day change and submit daily think for updated days (non-blocking).
2276
2277 Triggers once when the day rolls over at midnight. Queries ``updated_days()``
2278 for journal days that have new stream data but haven't completed a daily
2279 think yet, then submits up to ``MAX_UPDATED_CATCHUP`` thinks in chronological
2280 order (oldest first, yesterday last) via the TaskQueue.
2281
2282 Think auto-detects updated state and enables ``--refresh`` internally, so we
2283 don't pass it here.
2284
2285 Skipped in remote mode (no local data to process).
2286 """
2287 # Remote mode: no local processing, data is on the server
2288 if _is_remote_mode:
2289 return
2290
2291 today = datetime.now().date()
2292
2293 # Only trigger when day actually changes (at midnight)
2294 if today != _daily_state["last_day"]:
2295 # The day that just ended is what we process
2296 prev_day = _daily_state["last_day"]
2297
2298 # Guard against None (e.g., module reloaded without going through main())
2299 if prev_day is None:
2300 logging.warning("Daily state not initialized, skipping daily processing")
2301 _daily_state["last_day"] = today
2302 return
2303
2304 prev_day_str = prev_day.strftime("%Y%m%d")
2305
2306 # Update state for new day
2307 _daily_state["last_day"] = today
2308
2309 # Flush any dangling segment state from the previous day before daily think
2310 if not _flush_state["flushed"] and _flush_state["day"] == prev_day_str:
2311 _check_segment_flush(force=True)
2312
2313 today_str = today.strftime("%Y%m%d")
2314 run_catchup_drain(exclude={today_str})
2315
2316
2317def _handle_segment_observed(message: dict) -> None:
2318 """Handle segment completion events (from live observation or imports).
2319
2320 Submits journal think in segment mode via task queue, which handles both
2321 generators and segment agents. Also updates flush state to track
2322 segment recency.
2323 """
2324 if message.get("tract") != "observe" or message.get("event") != "observed":
2325 return
2326
2327 segment = message.get("segment") # e.g., "163045_300"
2328 if not segment:
2329 logging.warning("observed event missing segment field")
2330 return
2331
2332 # Use day from event payload, fallback to today (for live observation)
2333 day = message.get("day") or datetime.now().strftime("%Y%m%d")
2334
2335 # Batch/historical re-sensing heals deterministically via daily catchup's
2336 # segment-think pre-phase. A lone volatile segment think for a re-sensed
2337 # past segment can rewind live activity-timeline state, so submit nothing;
2338 # also leave flush state untouched so stale segments cannot reset
2339 # _check_segment_flush or pollute handle_daily_tasks' force-flush gate.
2340 if message.get("batch"):
2341 logging.info(
2342 "Batch observed segment deferred to daily catchup; "
2343 "no volatile segment think submitted: %s/%s",
2344 day,
2345 segment,
2346 )
2347 return
2348
2349 if load_processing_settings().mode == "deferred":
2350 logging.info(
2351 "Deferred mode: live segment %s/%s held for catchup drain; no live think",
2352 day,
2353 segment,
2354 )
2355 return
2356
2357 stream = message.get("stream")
2358
2359 # Update flush state — new segment resets the flush timer
2360 _flush_state["last_segment_ts"] = time.time()
2361 _flush_state["day"] = day
2362 _flush_state["segment"] = segment
2363 _flush_state["stream"] = stream
2364 _flush_state["flushed"] = False
2365
2366 logging.info(f"Segment observed: {day}/{segment}, submitting processing...")
2367
2368 # Submit via task queue — serializes with other think invocations
2369 cmd = ["journal", "think", "-v", "--day", day, "--segment", segment]
2370 if stream:
2371 cmd.extend(["--stream", stream])
2372 if not message.get("batch"):
2373 cmd.append("--live")
2374 if _task_queue:
2375 _task_queue.submit(cmd, day=day)
2376 else:
2377 logging.warning(
2378 "No task queue available for segment processing: %s/%s", day, segment
2379 )
2380
2381
2382def _check_segment_flush(force: bool = False) -> None:
2383 """Check if the last observed segment needs flushing.
2384
2385 If no new segments have arrived within FLUSH_TIMEOUT seconds, runs
2386 ``journal think --flush`` on the last segment to let flush-enabled agents
2387 close out dangling state (e.g., end active activities).
2388
2389 Args:
2390 force: Skip timeout check (used at day boundary to flush
2391 before daily think regardless of elapsed time).
2392
2393 Skipped in remote mode (no local processing).
2394 """
2395 if _is_remote_mode:
2396 return
2397
2398 last_ts = _flush_state["last_segment_ts"]
2399 if not last_ts or _flush_state["flushed"]:
2400 return
2401
2402 if load_processing_settings().mode == "deferred":
2403 return
2404
2405 if not force and time.time() - last_ts < FLUSH_TIMEOUT:
2406 return
2407
2408 day = _flush_state["day"]
2409 segment = _flush_state["segment"]
2410 if not day or not segment:
2411 return
2412
2413 _flush_state["flushed"] = True
2414
2415 stream = _flush_state.get("stream")
2416 cmd = ["journal", "think", "-v", "--day", day, "--segment", segment, "--flush"]
2417 if stream:
2418 cmd.extend(["--stream", stream])
2419 if _task_queue:
2420 _task_queue.submit(cmd, day=day)
2421 logging.info(f"Queued segment flush: {day}/{segment}")
2422 else:
2423 logging.warning(
2424 "No task queue available for segment flush: %s/%s", day, segment
2425 )
2426
2427
2428def _handle_segment_event_log(message: dict) -> None:
2429 """Log observe, think, and activity events with day+segment to segment/events.jsonl.
2430
2431 Any observe, think, or activity tract message with both day and segment fields
2432 gets logged to journal/day/segment/events.jsonl if that directory exists.
2433 """
2434 if message.get("tract") not in {"observe", "think", "activity"}:
2435 return
2436
2437 day = message.get("day")
2438 segment = message.get("segment")
2439
2440 if not day or not segment:
2441 return
2442
2443 stream = message.get("stream")
2444
2445 try:
2446 if stream:
2447 segment_dir = day_path(day, create=False) / stream / segment
2448 else:
2449 segment_dir = day_path(day, create=False) / segment
2450
2451 # Only log if segment directory exists
2452 if not segment_dir.is_dir():
2453 return
2454
2455 events_file = segment_dir / "events.jsonl"
2456
2457 # Append event as JSON line
2458 with open(events_file, "a", encoding="utf-8") as f:
2459 f.write(json.dumps(message, ensure_ascii=False) + "\n")
2460
2461 except Exception as e:
2462 logging.debug(f"Failed to log segment event: {e}")
2463
2464
2465def _handle_activity_recorded(message: dict) -> None:
2466 """Queue a per-activity think task when an activity is recorded.
2467
2468 Listens for activity.recorded events and submits a queued think task
2469 for per-activity agent processing (serialized via TaskQueue).
2470 """
2471 if message.get("tract") != "activity" or message.get("event") != "recorded":
2472 return
2473
2474 record_id = message.get("id")
2475 facet = message.get("facet")
2476 day = message.get("day")
2477
2478 if not record_id or not facet or not day:
2479 logging.warning("activity.recorded event missing required fields")
2480 return
2481
2482 cmd = ["journal", "think", "--activity", record_id, "--facet", facet, "--day", day]
2483
2484 if _task_queue:
2485 _task_queue.submit(cmd, day=day)
2486 logging.info(f"Queued activity think: {record_id} for #{facet}")
2487 else:
2488 logging.warning("No task queue available for activity think: %s", record_id)
2489
2490
2491def _handle_think_daily_complete(message: dict) -> None:
2492 """Submit a heartbeat task after daily think processing completes.
2493
2494 Listens for think.daily_complete events. Skips if a heartbeat process
2495 is already running (PID file guard).
2496 """
2497 if message.get("tract") != "think" or message.get("event") != "daily_complete":
2498 return
2499
2500 # Check if heartbeat is already running via PID file
2501 pid_file = Path(get_journal()) / "health" / "heartbeat.pid"
2502 if pid_file.exists():
2503 try:
2504 existing_pid = int(pid_file.read_text().strip())
2505 os.kill(existing_pid, 0)
2506 logging.info("Heartbeat already running (pid=%d), skipping", existing_pid)
2507 return
2508 except ProcessLookupError:
2509 pass # Stale PID file, proceed
2510 except PermissionError:
2511 logging.info(
2512 "Heartbeat running under different user (pid file exists), skipping"
2513 )
2514 return
2515 except ValueError:
2516 pass # Corrupt PID file, proceed
2517
2518 cmd = ["journal", "heartbeat"]
2519 if _task_queue:
2520 _task_queue.submit(cmd)
2521 logging.info("Queued heartbeat after daily think completion")
2522 else:
2523 logging.warning("No task queue available for heartbeat submission")
2524
2525
2526def _handle_callosum_message(message: dict) -> None:
2527 """Dispatch incoming Callosum messages to appropriate handlers."""
2528 _handle_task_request(message)
2529 _handle_supervisor_request(message)
2530 _handle_supervisor_drain(message)
2531 _handle_supervisor_start_local(message)
2532 _handle_segment_observed(message)
2533 _handle_activity_recorded(message)
2534 _handle_think_daily_complete(message)
2535 _handle_segment_event_log(message)
2536 _handle_cortex_outcome(message)
2537
2538
2539def _run_sync_tick(now: float) -> bool:
2540 """Write this supervisor's heartbeat and stop on live foreign writers."""
2541 global _last_sync_tick, _last_sync_snapshot, _sync_conflict_shutdown
2542 global shutdown_requested
2543
2544 if now - _last_sync_tick < DEFAULT_INTERVAL_SECONDS:
2545 return True
2546
2547 try:
2548 write_self_heartbeat()
2549 result = check_journal_sync(previous=_last_sync_snapshot)
2550 _last_sync_snapshot = result.snapshot
2551 _last_sync_tick = now
2552 if not result.is_conflict:
2553 return True
2554
2555 primary = result.primary_conflict
2556 if primary is None:
2557 return True
2558
2559 machine_prefix = primary.machine_id[:8] if primary.machine_id else "(unknown)"
2560 logging.error(
2561 "Another solstone instance is writing to this journal "
2562 "(host=%s pid=%s machine=%s) - shutting down.",
2563 primary.display_hostname,
2564 primary.pid,
2565 machine_prefix,
2566 )
2567 if _supervisor_callosum:
2568 try:
2569 _supervisor_callosum.emit(
2570 "supervisor",
2571 "sync_conflict",
2572 hostname=primary.display_hostname,
2573 journal_path=primary.journal_path,
2574 pid=primary.pid,
2575 machine_id_prefix=primary.machine_id[:8]
2576 if primary.machine_id
2577 else "",
2578 wall_time=datetime.now(timezone.utc)
2579 .isoformat(timespec="seconds")
2580 .replace("+00:00", "Z"),
2581 )
2582 except Exception:
2583 logging.exception("Failed to emit sync_conflict event")
2584 shutdown_requested = True
2585 _sync_conflict_shutdown = True
2586 return False
2587 except Exception:
2588 logging.exception("Sync conflict check failed (continuing)")
2589 return True
2590
2591
2592def _run_gate_tick(now: float) -> None:
2593 global _last_gate_tick
2594
2595 if now - _last_gate_tick < GATE_TICK_INTERVAL_S:
2596 return
2597 _last_gate_tick = now
2598 if _is_remote_mode:
2599 return
2600 settings = load_processing_settings()
2601 if settings.mode != "deferred":
2602 return
2603 reading = (
2604 poll_display_powersave(time.monotonic())
2605 if settings.gate.display_powersave.enabled
2606 else DISPLAY_POWERSAVE_UNAVAILABLE
2607 )
2608 gate = evaluate_drain_gate(settings, datetime.now(), reading)
2609 if not gate.open:
2610 return
2611 run_catchup_drain()
2612
2613
2614async def supervise(
2615 *,
2616 daily: bool = True,
2617 schedule: bool = True,
2618 procs: list[RunnerManagedProcess] | None = None,
2619) -> None:
2620 """Main supervision loop. Runs at 1-second intervals for responsiveness.
2621
2622 Monitors runner health, emits status, triggers daily processing,
2623 and checks scheduled agents.
2624 """
2625 global _last_gate_tick, _last_sync_tick
2626 global _last_sync_snapshot, _sync_conflict_shutdown
2627
2628 last_status_emit = 0.0
2629 _last_gate_tick = 0.0
2630 reset_display_powersave_monitor()
2631 _last_sync_tick = 0.0
2632 _last_sync_snapshot = None
2633 _sync_conflict_shutdown = False
2634
2635 try:
2636 while (
2637 not shutdown_requested
2638 ): # pragma: no cover - loop checked via unit tests by patching
2639 if _task_queue:
2640 _task_queue.enforce_deadlines(time.time())
2641
2642 # Check for runner exits first (immediate alert)
2643 if procs:
2644 await handle_runner_exits(procs)
2645 await _check_local_server_recovery()
2646
2647 # Emit status every 5 seconds
2648 now = time.time()
2649 if now - last_status_emit >= 5:
2650 if _supervisor_callosum and procs:
2651 try:
2652 status = collect_status(procs)
2653 _supervisor_callosum.emit("supervisor", "status", **status)
2654 except Exception as e:
2655 logging.debug(f"Status emission failed: {e}")
2656 last_status_emit = now
2657
2658 # Check for segment flush (non-blocking, submits via task queue)
2659 _check_segment_flush()
2660
2661 # Check for journal sync conflicts (usually just heartbeat IO)
2662 if not _run_sync_tick(now):
2663 return
2664
2665 # Check for daily processing (non-blocking, submits via task queue)
2666 if daily:
2667 handle_daily_tasks()
2668 _run_gate_tick(now)
2669
2670 # Check periodic task schedules (non-blocking, submits via callosum)
2671 if schedule:
2672 scheduler.check()
2673
2674 # Sleep 1 second before next iteration (responsive to shutdown)
2675 await asyncio.sleep(1)
2676 finally:
2677 pass # Callosum cleanup happens in main()
2678
2679
2680def parse_args() -> argparse.ArgumentParser:
2681 parser = SupervisorArgumentParser(description="Monitor journaling health")
2682 parser.add_argument(
2683 "port",
2684 nargs="?",
2685 type=int,
2686 default=0,
2687 help="Convey port (0 = auto-select available port)",
2688 )
2689 parser.add_argument(
2690 "--threshold",
2691 type=int,
2692 default=DEFAULT_THRESHOLD,
2693 help="Seconds before heartbeat considered stale",
2694 )
2695 parser.add_argument(
2696 "--interval", type=int, default=CHECK_INTERVAL, help="Polling interval seconds"
2697 )
2698 parser.add_argument(
2699 "--no-daily",
2700 action="store_true",
2701 help="Disable daily processing run at midnight",
2702 )
2703 parser.add_argument(
2704 "--no-cortex",
2705 action="store_true",
2706 help="Do not start the Cortex server (run it manually for debugging)",
2707 )
2708 parser.add_argument(
2709 "--no-spl",
2710 action="store_true",
2711 help="Do not start the spl tunnel service",
2712 )
2713 parser.add_argument(
2714 "--no-convey",
2715 action="store_true",
2716 help="Do not start the Convey web application",
2717 )
2718 parser.add_argument(
2719 "--no-schedule",
2720 action="store_true",
2721 help="Disable periodic task scheduler",
2722 )
2723 parser.add_argument(
2724 FLAG,
2725 action="store_true",
2726 help=(
2727 "App-supervised mode: skip all service-unit work and self-exit when "
2728 "the parent process dies (used by the macOS app)."
2729 ),
2730 )
2731 parser.add_argument(
2732 "--remote",
2733 type=str,
2734 help="Remote mode: URL for segment transfer (not yet implemented)",
2735 )
2736 return parser
2737
2738
2739def handle_shutdown(signum, frame):
2740 """Handle shutdown signals gracefully."""
2741 global shutdown_requested
2742 if not shutdown_requested:
2743 shutdown_requested = True
2744 logger.info("shutdown requested via signal %d", signum)
2745 live = [managed for managed in _managed_procs if managed.is_running()]
2746 if live:
2747 logger.info("shutdown: signaling %d managed child(ren)", len(live))
2748 for managed in live:
2749 try:
2750 managed.process.terminate()
2751 except Exception:
2752 logger.exception("shutdown: terminate failed for %s", managed.name)
2753
2754 deadline = time.monotonic() + HANDLE_SHUTDOWN_REAP_S
2755 while time.monotonic() < deadline:
2756 if all(not managed.is_running() for managed in live):
2757 break
2758 time.sleep(0.05)
2759
2760 kills = 0
2761 for managed in live:
2762 if managed.is_running():
2763 try:
2764 managed.process.kill()
2765 logger.warning(
2766 "shutdown: SIGKILL pid=%s name=%s",
2767 managed.process.pid,
2768 managed.name,
2769 )
2770 kills += 1
2771 except Exception:
2772 logger.exception("shutdown: kill failed for %s", managed.name)
2773
2774 cleanly = len(live) - kills
2775 logger.info(
2776 "shutdown: reap complete (%d exited cleanly, %d SIGKILL'd)",
2777 cleanly,
2778 kills,
2779 )
2780 raise KeyboardInterrupt
2781 # Second signal during shutdown: cleanup is already in progress.
2782
2783
2784def _ensure_venv_bin_on_path() -> None:
2785 """Prepend the venv bin dir (sibling of sys.executable) to PATH if absent.
2786
2787 Idempotent — safe to call repeatedly. Lets the supervisor spawn `sol` and
2788 other venv-installed entry points even when the operator's shell PATH does
2789 not include the venv bin dir.
2790 """
2791 venv_bin = os.path.dirname(sys.executable)
2792 parts = os.environ.get("PATH", "").split(os.pathsep)
2793 if parts and parts[0] == venv_bin:
2794 return
2795 parts = [venv_bin] + [p for p in parts if p != venv_bin]
2796 os.environ["PATH"] = os.pathsep.join(parts)
2797
2798
2799def register_baseline_caps(queue: TaskQueue) -> None:
2800 """Register caps that must hold regardless of the schedule_enabled gate.
2801
2802 Reactive partitions (daily/segment/indexer/importer) and on-demand backup
2803 can reach _active even under --no-schedule, so their caps cannot live behind
2804 the schedule-only registration block.
2805 """
2806 for name, seconds in REACTIVE_TASK_CAPS.items():
2807 queue.set_cap(name, seconds)
2808 queue.set_cap(
2809 TaskQueue.get_command_name(BACKUP_RUN_CMD),
2810 parse_duration_seconds(BACKUP_MAX_RUNTIME),
2811 )
2812
2813
2814def main() -> None:
2815 parser = parse_args()
2816
2817 # Capture journal info before setup_cli() hydrates os.environ from journal
2818 # config and strips shell-only managed provider keys.
2819 journal_info = get_journal_info()
2820
2821 args = setup_cli(parser)
2822 app_supervised = is_app_supervised(sys.argv)
2823 _ensure_venv_bin_on_path()
2824
2825 journal_path = _get_journal_path()
2826
2827 log_level = logging.DEBUG if args.debug else logging.INFO
2828 log_path = journal_path / "health" / "supervisor.log"
2829 log_path.parent.mkdir(parents=True, exist_ok=True)
2830 _configure_supervisor_logging(log_path, log_level)
2831
2832 if args.verbose or args.debug:
2833 console_handler = logging.StreamHandler()
2834 console_handler.setLevel(log_level)
2835 console_handler.setFormatter(
2836 logging.Formatter("%(asctime)s %(levelname)s %(message)s")
2837 )
2838 logging.getLogger().addHandler(console_handler)
2839
2840 # Singleton guard: only one supervisor per journal
2841 health_dir = journal_path / "health"
2842 lock_path = health_dir / "supervisor.lock"
2843 pid_path = health_dir / "supervisor.pid"
2844
2845 lock_fd = open(lock_path, "w")
2846 try:
2847 fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
2848 except OSError:
2849 lock_fd.close()
2850 pid_str = ""
2851 try:
2852 pid_str = pid_path.read_text().strip()
2853 except OSError:
2854 pass
2855 pid_msg = f" (PID {pid_str})" if pid_str else ""
2856 if os.environ.get("INVOCATION_ID"):
2857 holder_pid = pid_str or "unknown"
2858 print(
2859 "Supervisor already running "
2860 f"(PID {holder_pid}) - exiting cleanly under systemd activation"
2861 )
2862 sys.exit(0)
2863 sock_path = health_dir / "callosum.sock"
2864 if sock_path.exists():
2865 try:
2866 from solstone.think.health_cli import health_check
2867
2868 print(f"Supervisor already running{pid_msg}\n")
2869 health_check()
2870 except Exception:
2871 print(f"Supervisor already running{pid_msg}")
2872 else:
2873 print(f"Supervisor already running{pid_msg}")
2874 sys.exit(1)
2875
2876 print(
2877 "Checking for other active solstone instances on this journal...",
2878 flush=True,
2879 )
2880 snapshot = check_journal_sync(journal=journal_path)
2881 if snapshot.is_conflict:
2882 print(format_conflict_message(snapshot), file=sys.stderr)
2883 try:
2884 lock_fd.close()
2885 except Exception:
2886 pass
2887 sys.exit(1)
2888 write_self_heartbeat(journal=journal_path)
2889
2890 pid_path.write_text(str(os.getpid()))
2891 start_time_path = health_dir / "supervisor.start_time"
2892 # Stamp THIS process's kernel create_time() so the recorded value equals what
2893 # is_supervisor_up()/_valid_marker() later read via psutil for this pid —
2894 # eliminating (not minimizing) drift from a wall-clock time.time() stamp.
2895 start_time_path.write_text(str(psutil.Process().create_time()))
2896 logging.info("Singleton lock acquired (PID %d)", os.getpid())
2897 _sweep_orphaned_sol_processes(journal_path)
2898
2899 # Set up signal handlers
2900 signal.signal(signal.SIGINT, handle_shutdown)
2901 signal.signal(signal.SIGTERM, handle_shutdown)
2902
2903 # Show journal path and source on startup
2904 path, source = journal_info
2905 print(f"Journal: {path} (from {source})")
2906 logging.info("Supervisor starting...")
2907
2908 global _managed_procs, _supervisor_callosum, _is_remote_mode
2909 global _task_queue
2910 procs: list[RunnerManagedProcess] = []
2911 convey_port = None
2912 convey_proc = None
2913
2914 # Remote mode: run sync instead of local processing
2915 _is_remote_mode = bool(args.remote)
2916
2917 # Run pending journal-maintenance tasks before spawning any writer children.
2918 # Callosum isn't up yet (emit_fn=None); migrations log through supervisor's logger only.
2919 try:
2920 ran, succeeded = run_pending_tasks(journal_path, emit_fn=None)
2921 if ran > 0:
2922 print(f" Ran {ran} maintenance task(s)", flush=True)
2923 if ran == succeeded:
2924 logging.info("Completed %d/%d maintenance task(s)", succeeded, ran)
2925 else:
2926 logging.error(
2927 "Maintenance tasks completed with failures: %d/%d succeeded",
2928 succeeded,
2929 ran,
2930 )
2931 except Exception:
2932 logging.exception("Maintenance runner raised; continuing startup")
2933
2934 try:
2935 from solstone.think.importers.journal_archive import sweep_stale_extract_dirs
2936
2937 swept = sweep_stale_extract_dirs()
2938 if swept > 0:
2939 logging.info("Swept %d stale journal-archive extract dir(s)", swept)
2940 except Exception:
2941 logging.exception("Journal archive extract sweep raised; continuing startup")
2942
2943 # Start Callosum in-process first - it's the message bus that other services depend on
2944 try:
2945 print(" Starting Callosum bus...", flush=True)
2946 start_callosum_in_process()
2947 except RuntimeError as e:
2948 logging.error(f"Failed to start Callosum server: {e}")
2949 parser.error(f"Failed to start Callosum server: {e}")
2950 return
2951
2952 # Connect supervisor's Callosum client to capture startup events from other services
2953 try:
2954 _supervisor_callosum = CallosumConnection(defaults={"rev": get_rev()})
2955 _supervisor_callosum.start(callback=_handle_callosum_message)
2956 logging.info("Supervisor connected to Callosum")
2957 except Exception as e:
2958 logging.warning(f"Failed to start Callosum connection: {e}")
2959
2960 # Mirror supervisor log output to callosum logs tract (best-effort)
2961 supervisor_ref = str(now_ms())
2962 global _supervisor_ref, _supervisor_start
2963 _supervisor_ref = supervisor_ref
2964 _supervisor_start = time.time()
2965 if _supervisor_callosum:
2966 try:
2967 handler = CallosumLogHandler(_supervisor_callosum, supervisor_ref)
2968 handler.setFormatter(
2969 logging.Formatter("%(asctime)s %(levelname)s %(message)s")
2970 )
2971 logging.getLogger().addHandler(handler)
2972 except Exception:
2973 pass
2974
2975 # Initialize task queue with callosum event callback
2976 _task_queue = TaskQueue(on_queue_change=_emit_queue_event, ready=False)
2977 register_baseline_caps(_task_queue)
2978
2979 # Now start other services (their startup events will be captured)
2980 if _is_remote_mode:
2981 # Remote mode: transfer send will be added here
2982 pass
2983 else:
2984 # Local mode: convey first, then sense for file processing
2985 os.environ["SOL_SUPERVISOR_SPAWNED"] = "1"
2986 if not args.no_convey:
2987 print(f" Starting convey on port {args.port}...", flush=True)
2988 convey_proc, convey_port = start_convey_server(
2989 verbose=args.verbose, debug=args.debug, port=args.port
2990 )
2991 procs.append(convey_proc)
2992 wait_for_convey_ready(convey_proc)
2993 print(" Convey ready", flush=True)
2994 from solstone.think.providers.local_endpoint import resolve_local_endpoint
2995
2996 if is_local_provider_needed() and resolve_local_endpoint().is_bundled:
2997 proc = start_local_server()
2998 if proc is not None:
2999 procs.append(proc)
3000 # Sense handles file processing
3001 print(" Starting sense...", flush=True)
3002 procs.append(start_sense())
3003 # Cortex for agent execution
3004 if not args.no_cortex:
3005 print(" Starting cortex...", flush=True)
3006 procs.append(start_cortex_server())
3007 # spl tunnel service (opt-out via --no-spl)
3008 if not args.no_spl:
3009 print(" Starting spl...", flush=True)
3010 procs.append(start_spl_service())
3011
3012 # Make procs accessible to restart handler
3013 _managed_procs = procs
3014
3015 # Initialize daily state to today - think only triggers at midnight when day changes
3016 _daily_state["last_day"] = datetime.now().date()
3017
3018 # Initialize periodic task scheduler
3019 schedule_enabled = not args.no_schedule and not _is_remote_mode
3020 if schedule_enabled and _supervisor_callosum:
3021 try:
3022 maintenance.register_maintenance_schedules()
3023 except Exception:
3024 logging.error("Failed to register maintenance schedules", exc_info=True)
3025 scheduler.init(_supervisor_callosum)
3026 scheduler.register_defaults()
3027 if _task_queue:
3028 for cmd, seconds in scheduler.collect_runtime_caps():
3029 cmd_name = TaskQueue.get_command_name(cmd)
3030 _task_queue.set_cap(cmd_name, seconds)
3031 logging.info(
3032 "Registered max_runtime cap for %s: %ss",
3033 cmd_name,
3034 seconds,
3035 )
3036
3037 if _task_queue:
3038 _task_queue.set_ready()
3039
3040 # Show Convey URL if running
3041 if convey_port:
3042 print(f"Convey: http://localhost:{convey_port}/")
3043
3044 logging.info(f"Started {len(procs)} processes, entering supervision loop")
3045 daily_enabled = not args.no_daily and not _is_remote_mode
3046 if daily_enabled:
3047 logging.info("Daily processing scheduled for midnight")
3048
3049 # Startup catchup: submit thinks for days with pending stream data
3050 if daily_enabled:
3051 _startup_catchup_drain()
3052
3053 # Startup catch-up: submit overdue schedule entries missed while down
3054 if schedule_enabled and _supervisor_callosum:
3055 scheduler.catch_up()
3056
3057 try:
3058 convey_accepting = convey_proc is None or is_solstone_up(timeout=1.0)
3059 if convey_accepting:
3060 print(" Supervisor ready", flush=True)
3061 _sd_notify("READY=1")
3062 signal_ready()
3063 else:
3064 logging.error(
3065 "Convey is not accepting on :%s; withholding readiness marker, "
3066 "continuing into supervise loop",
3067 read_service_port("convey"),
3068 )
3069 if app_supervised:
3070 start_parent_death_watcher()
3071 asyncio.run(
3072 supervise(
3073 daily=daily_enabled,
3074 schedule=schedule_enabled,
3075 procs=procs if procs else None,
3076 )
3077 )
3078 except KeyboardInterrupt:
3079 logging.info("Caught KeyboardInterrupt, shutting down...")
3080 finally:
3081 try:
3082 clear_ready()
3083 except Exception as exc:
3084 logging.warning("Failed to clear readiness marker during shutdown: %s", exc)
3085 try:
3086 if not _sync_conflict_shutdown:
3087 clear_self_heartbeat()
3088 except Exception as exc:
3089 logging.warning("Failed to clear sync heartbeat during shutdown: %s", exc)
3090
3091 logging.info("Stopping all processes...")
3092 print("\nShutting down gracefully (this may take a moment)...", flush=True)
3093
3094 if _task_queue:
3095 task_drain_timeout = APP_SUPERVISED_TASK_DRAIN_S if app_supervised else 10
3096 _task_queue.shutdown(timeout=task_drain_timeout)
3097
3098 # Stop services in reverse order
3099 child_stop_timeout = APP_SUPERVISED_CHILD_STOP_S if app_supervised else None
3100 for managed in reversed(procs):
3101 _stop_process(managed, timeout_cap=child_stop_timeout)
3102
3103 # Disconnect supervisor's Callosum connection
3104 if _supervisor_callosum:
3105 _supervisor_callosum.stop()
3106 logging.info("Supervisor disconnected from Callosum")
3107
3108 # Stop in-process Callosum server last
3109 callosum_join_timeout = (
3110 APP_SUPERVISED_CALLOSUM_JOIN_S if app_supervised else 5.0
3111 )
3112 stop_callosum_in_process(join_timeout=callosum_join_timeout)
3113
3114 logging.info("Supervisor shutdown complete.")
3115 print("Shutdown complete.", flush=True)
3116
3117 if _sync_conflict_shutdown:
3118 sys.exit(2)
3119
3120
3121if __name__ == "__main__":
3122 main()