Extract PRISM coordinator core owners

lab/prism/background_services.py

@@ -0,0 +1,200 @@
+"""Named lifecycle registry for PRISM process-level background loops.
+
+The registry deliberately has no ``start_all`` operation.  The coordinator
+starts named services at the existing recovery boundaries, while this module
+owns the start-once state and the exact thread handles used during shutdown.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import threading
+from typing import Callable, Iterable
+
+
+@dataclass(frozen=True, slots=True)
+class BackgroundServiceSpec:
+    """Immutable construction and shutdown policy for one background loop."""
+
+    name: str
+    thread_name: str
+    target: Callable[[], None]
+    daemon: bool
+    join_timeout: float
+    watchdog_monitored: bool
+    registration_identity: object | None = None
+
+    def __post_init__(self) -> None:
+        if not self.name:
+            raise ValueError("background service name must not be empty")
+        if not self.thread_name:
+            raise ValueError("background service thread name must not be empty")
+        if self.join_timeout < 0:
+            raise ValueError("background service join timeout must be nonnegative")
+
+
+@dataclass(frozen=True, slots=True)
+class BackgroundServiceSnapshot:
+    """Read-only lifecycle state returned without exposing registry records."""
+
+    specification: BackgroundServiceSpec
+    started: bool
+    thread: threading.Thread | None
+
+
+@dataclass(slots=True)
+class _BackgroundServiceRecord:
+    specification: BackgroundServiceSpec
+    started: bool = False
+    thread: threading.Thread | None = None
+    start_hook_completed: bool = False
+
+
+ThreadFactory = Callable[..., threading.Thread]
+
+
+class BackgroundServiceRegistry:
+    """Start named process services once and retain their drain handles."""
+
+    def __init__(
+        self,
+        specifications: Iterable[BackgroundServiceSpec] = (),
+        *,
+        thread_factory: ThreadFactory = threading.Thread,
+    ) -> None:
+        self._lock = threading.Lock()
+        self._thread_factory = thread_factory
+        self._records: dict[str, _BackgroundServiceRecord] = {}
+        self._thread_names: set[str] = set()
+        for specification in specifications:
+            self.register(specification)
+
+    def register(self, specification: BackgroundServiceSpec) -> None:
+        """Register one service without starting it."""
+        with self._lock:
+            self._register_locked(specification)
+
+    def register_if_absent(self, specification: BackgroundServiceSpec) -> bool:
+        """Atomically install an equivalent dynamic service at most once.
+
+        Returns true when this call registered the service. Concurrent callers
+        describing the same lifecycle and registration identity receive false;
+        a conflicting reuse of either name still fails explicitly.
+        """
+        with self._lock:
+            existing = self._records.get(specification.name)
+            if existing is not None:
+                if self._equivalent(existing.specification, specification):
+                    return False
+                raise ValueError(
+                    "incompatible background service registration for name: "
+                    f"{specification.name}"
+                )
+            self._register_locked(specification)
+            return True
+
+    def _register_locked(self, specification: BackgroundServiceSpec) -> None:
+        if specification.name in self._records:
+            raise ValueError(
+                f"background service is already registered: {specification.name}"
+            )
+        if specification.thread_name in self._thread_names:
+            raise ValueError(
+                "background service thread name is already registered: "
+                f"{specification.thread_name}"
+            )
+        self._records[specification.name] = _BackgroundServiceRecord(
+            specification=specification
+        )
+        self._thread_names.add(specification.thread_name)
+
+    @staticmethod
+    def _equivalent(
+        existing: BackgroundServiceSpec,
+        candidate: BackgroundServiceSpec,
+    ) -> bool:
+        target_equivalent = (
+            existing.target is candidate.target
+            if existing.registration_identity is None
+            and candidate.registration_identity is None
+            else existing.registration_identity == candidate.registration_identity
+            and existing.registration_identity is not None
+            and candidate.registration_identity is not None
+        )
+        return bool(
+            target_equivalent
+            and existing.name == candidate.name
+            and existing.thread_name == candidate.thread_name
+            and existing.daemon == candidate.daemon
+            and existing.join_timeout == candidate.join_timeout
+            and existing.watchdog_monitored == candidate.watchdog_monitored
+        )
+
+    def contains(self, name: str) -> bool:
+        with self._lock:
+            return name in self._records
+
+    def service_names(self) -> tuple[str, ...]:
+        with self._lock:
+            return tuple(self._records)
+
+    def snapshot(self, name: str) -> BackgroundServiceSnapshot:
+        with self._lock:
+            record = self._records[name]
+            return BackgroundServiceSnapshot(
+                specification=record.specification,
+                started=record.started,
+                thread=record.thread,
+            )
+
+    def start(
+        self,
+        name: str,
+        *,
+        on_started: Callable[[BackgroundServiceSpec], None] | None = None,
+    ) -> threading.Thread:
+        """Start a named service once, returning the same thread thereafter."""
+        with self._lock:
+            record = self._records[name]
+            if record.thread is not None:
+                if not record.start_hook_completed and on_started is not None:
+                    on_started(record.specification)
+                    record.start_hook_completed = True
+                return record.thread
+            specification = record.specification
+            thread = self._thread_factory(
+                target=specification.target,
+                name=specification.thread_name,
+                daemon=specification.daemon,
+            )
+            thread.start()
+            record.thread = thread
+            record.started = True
+            if on_started is None:
+                record.start_hook_completed = True
+            else:
+                # The live thread remains registered for shutdown if this
+                # nonblocking side-effect fails. A later start call retries
+                # only the hook and never creates a second worker.
+                on_started(specification)
+                record.start_hook_completed = True
+            return thread
+
+    def threads_to_drain(self) -> tuple[tuple[threading.Thread, float], ...]:
+        """Return started threads in stable registration/shutdown order."""
+        with self._lock:
+            return tuple(
+                (record.thread, record.specification.join_timeout)
+                for record in self._records.values()
+                if record.started and record.thread is not None
+            )
+
+    def watchdog_service_names(self, *, started_only: bool = False) -> tuple[str, ...]:
+        """Derive watchdog keys from the same records used to start loops."""
+        with self._lock:
+            return tuple(
+                record.specification.name
+                for record in self._records.values()
+                if record.specification.watchdog_monitored
+                and (record.started or not started_only)
+            )

lab/prism/bounded_executor.py

@@ -0,0 +1,171 @@
+"""Bounded priority executor used for PRISM job delivery."""
+
+from __future__ import annotations
+
+from concurrent.futures import Future
+import heapq
+import queue
+import threading
+from typing import Any, Callable
+
+
+class _DeliveryQueueFull(RuntimeError):
+    """The bounded delivery executor cannot admit another task."""
+
+
+class _BoundedPriorityExecutor:
+    """Small Future-compatible executor with bounded, priority-ordered work."""
+
+    # A short poll keeps idle workers promptly joinable without requiring one
+    # poison-pill queue slot per worker (the queue may be smaller than the
+    # worker pool).
+    _WORKER_POLL_SECONDS = 0.01
+
+    def __init__(
+        self,
+        *,
+        max_workers: int,
+        max_queue_size: int,
+        thread_name_prefix: str = "prism-job-delivery",
+    ) -> None:
+        self.max_workers = max_workers
+        self.max_queue_size = max_queue_size
+        self._queue: queue.PriorityQueue[tuple[object, ...]] = queue.PriorityQueue(
+            maxsize=max_queue_size
+        )
+        self._lock = threading.Lock()
+        self._sequence = 0
+        self._active_workers = 0
+        self._shutdown = False
+        self._cancel_futures_on_shutdown = False
+        self._threads = [
+            threading.Thread(
+                target=self._worker,
+                name=f"{thread_name_prefix}-{index + 1}",
+                daemon=True,
+            )
+            for index in range(max_workers)
+        ]
+        for thread in self._threads:
+            thread.start()
+
+    def submit(
+        self,
+        function: Callable[..., Any],
+        /,
+        *args: object,
+        priority: int = 2,
+        **kwargs: object,
+    ) -> Future[Any]:
+        future: Future[Any] = Future()
+        with self._lock:
+            if self._shutdown:
+                raise RuntimeError("delivery executor is shut down")
+            self._sequence += 1
+            item = (
+                int(priority),
+                self._sequence,
+                future,
+                function,
+                args,
+                kwargs,
+            )
+            try:
+                self._queue.put_nowait(item)
+            except queue.Full as exc:
+                raise _DeliveryQueueFull("delivery executor queue is full") from exc
+        return future
+
+    def _worker(self) -> None:
+        while True:
+            try:
+                item = self._queue.get(timeout=self._WORKER_POLL_SECONDS)
+            except queue.Empty:
+                with self._lock:
+                    if self._shutdown:
+                        return
+                continue
+            _, _, future, function, args, kwargs = item
+            assert isinstance(future, Future)
+            with self._lock:
+                cancel_for_shutdown = (
+                    self._shutdown and self._cancel_futures_on_shutdown
+                )
+            if cancel_for_shutdown:
+                future.cancel()
+                self._queue.task_done()
+                continue
+            if not future.set_running_or_notify_cancel():
+                self._queue.task_done()
+                continue
+            with self._lock:
+                self._active_workers += 1
+            try:
+                result = function(*args, **kwargs)
+            except BaseException as exc:
+                future.set_exception(exc)
+            else:
+                future.set_result(result)
+            finally:
+                with self._lock:
+                    self._active_workers -= 1
+                self._queue.task_done()
+
+    def cancel(self, future: Future[Any]) -> bool:
+        """Cancel ``future`` and immediately discard it when still queued.
+
+        Cancelling a ``Future`` alone does not release bounded queue capacity
+        until a worker dequeues it. Removing the exact entry under the queue
+        mutex makes that capacity available to a replacement immediately.
+        """
+        removed = False
+        with self._queue.mutex:
+            queued_items = self._queue.queue
+            for index, item in enumerate(queued_items):
+                if item[2] is not future:
+                    continue
+                queued_items.pop(index)
+                heapq.heapify(queued_items)
+                self._queue.unfinished_tasks -= 1
+                if self._queue.unfinished_tasks == 0:
+                    self._queue.all_tasks_done.notify_all()
+                self._queue.not_full.notify()
+                removed = True
+                break
+        # Invoke callbacks only after releasing the queue mutex. A cancellation
+        # callback may submit the replacement that consumes the reclaimed slot.
+        future.cancel()
+        return removed
+
+    def stats(self) -> tuple[int, int]:
+        with self._lock:
+            return self._queue.qsize(), self._active_workers
+
+    def shutdown(self, *, wait: bool = True, cancel_futures: bool = False) -> None:
+        with self._lock:
+            if self._shutdown:
+                threads = list(self._threads)
+                already_shutdown = True
+            else:
+                self._shutdown = True
+                self._cancel_futures_on_shutdown = cancel_futures
+                threads = list(self._threads)
+                already_shutdown = False
+        if already_shutdown:
+            if wait:
+                for thread in threads:
+                    thread.join()
+            return
+        if cancel_futures:
+            while True:
+                try:
+                    item = self._queue.get_nowait()
+                except queue.Empty:
+                    break
+                future = item[2]
+                if isinstance(future, Future):
+                    future.cancel()
+                self._queue.task_done()
+        if wait:
+            for thread in threads:
+                thread.join()