capa.runtime¶

def runtime_diagnostics(self) -> dict[str, Any]:
    """Snapshot the per-loop / per-bridge / per-worker metrics.

    The dict shape is intended for ``manifest.queue_health`` consumption
    (one entry per queue/bridge/worker keyed by tag), so the existing
    manifest schema works without extension:

    * ``loop.conductor`` — conductor-loop lag percentiles.
    * ``loop.worker:<resource_id>`` — per-worker loop lag (zeros for
      now; workers expose their own LoopLagMetric and a future step
      plumbs it through here).
    * ``bridge.outbound:<resource_id>`` — per-worker outbound bridge
      latency + blocked time.
    * ``worker:<resource_id>`` — tick / poll durations, loop lag,
      command counts, last-sample age.

    Returns an empty dict before :meth:`_run` enters its task group
    (i.e. before bridges and workers exist).
    """
    out: dict[str, dict[str, float]] = {}
    # Runtime tunables — included so status-bar / observers can color
    # health values against the configured thresholds without having
    # to import ConductorConfig defaults.
    out["runtime"] = {
        "saturation_deadline_s": float(self._config.saturation_deadline_s),
        "loop_lag_warn_ms": float(self._config.loop_lag_warn_ms),
    }
    # Conductor loop lag.
    out["loop.conductor"] = {
        "samples": float(self._loop_lag.samples_total),
        "lag_p50_ms": self._loop_lag.p50_ms,
        "lag_p99_ms": self._loop_lag.p99_ms,
        "lag_max_ms": self._loop_lag.max_lag_ms,
    }
    # Per-worker bridges (outbound).
    for rid, bridge in self._bridges.items():
        m = bridge.metrics
        blocked_now = m.blocked_since_ms
        out[f"bridge.outbound:{rid}"] = {
            "depth": float(m.depth),
            "depth_max": float(m.depth_max),
            "capacity": float(bridge.capacity),
            "enqueued_total": float(m.enqueued_total),
            "dequeued_total": float(m.dequeued_total),
            "dropped_total": float(m.dropped_total),
            "blocked_total_ms": float(m.blocked_total_ms),
            "blocked_since_ms": float(blocked_now) if blocked_now is not None else -1.0,
            "latency_p50_ms": float(m.latency_p50_ms),
            "latency_p99_ms": float(m.latency_p99_ms),
        }
    # Per-worker tick/loop metrics. Per-adapter failure policy
    # (``WorkerMetrics.on_failure``) is intentionally not serialized
    # into this float-valued diagnostics block. Per-device
    # stream-silence/fatal-error enforcement can attach here using the
    # resolved policy metadata on each worker.
    for rid, worker in self._pool.workers.items():
        wm = worker.metrics
        out[f"worker:{rid}"] = {
            "samples_emitted": float(wm.samples_emitted),
            "polls_emitted": float(wm.polls_emitted),
            "commands_total": float(wm.commands_total),
            "commands_failed": float(wm.commands_failed),
            "tick_duration_p50_ms": float(wm.tick_duration_p50_ms),
            "tick_duration_p99_ms": float(wm.tick_duration_p99_ms),
            "poll_period_p50_ms": float(wm.poll_period_p50_ms),
            "poll_period_p99_ms": float(wm.poll_period_p99_ms),
            "poll_rate_hz": float(wm.poll_rate_hz),
            "loop_lag_p99_ms": float(wm.loop_lag_ms_p99),
            "last_sample_age_s": float(wm.last_sample_age_s),
        }
    return out

def start(self) -> Future[RunHandle]:
    """Spawn the conductor thread and begin the run.

    Returns a future that resolves with :class:`RunHandle` once the
    run reaches RUNNING state (procedure has started). On failure
    before RUNNING, the handle future rejects and the result future
    resolves with a CRASHED outcome.

    Idempotent only in the failure direction — a second call after
    :meth:`start` was invoked raises :class:`ConductorStateError`.
    """
    if self._thread is not None:
        raise ConductorStateError("Conductor.start() called twice", current=self._state)
    self._thread = threading.Thread(
        target=self._thread_main, name=self._thread_name, daemon=False
    )
    self._thread.start()
    return self._handle_future

def stop(self, *, reason: str = "operator_stop") -> Future[RunResult]:
    """Request a cooperative shutdown.

    Sets the completion event on the conductor loop and returns the
    same future as :attr:`result_future` for convenience. Idempotent:
    subsequent calls observe the existing shutdown but the recorded
    ``exit_reason`` is the first caller's.
    """
    if self._stop_requested:
        return self._result_future
    self._stop_requested = True
    self._exit_reason = reason
    # Default outcome assumes operator-initiated; the conductor may
    # override to CRASHED / CRASHED_BUT_SEALED if it discovers the
    # real cause first.
    if self._outcome is RunOutcome.COMPLETED:
        self._outcome = RunOutcome.ABORTED
    loop = self._loop
    ev = self._completion_event
    if loop is not None and ev is not None and not ev.is_set():
        loop.call_soon_threadsafe(ev.set)
    return self._result_future

def dispatch(self, device: str, cmd: DeviceCommand) -> Future[CommandResult]:
    """Run-time command dispatch.

    Refused outside PREPARING / RUNNING. Routes through the pool to
    the worker hosting ``device``.
    """
    if not self._state.permits_dispatch():
        raise ConductorStateError(
            f"dispatch refused in state {self._state}", current=self._state
        )
    return self._pool.dispatch(device, cmd)

def snapshot(self, device: str) -> Future[DeviceEmission]:
    """One-shot snapshot via the pool's worker for ``device``.

    Same state gate as :meth:`dispatch`.
    """
    if not self._state.permits_dispatch():
        raise ConductorStateError(
            f"snapshot refused in state {self._state}", current=self._state
        )
    return self._pool.snapshot(device)

def attach_ui_bridge(self, bridge: ThreadBridge[WorkerEmission]) -> None:
    """Wire a Conductor → UI :class:`ThreadBridge`.

    Must be called BEFORE :meth:`start` so the drain task picks the
    reference up on first dispatch. Headless callers omit this; the
    drain stays a writer-and-bus-only fan-out.

    The UI side owns ``attach_consumer``/``attach_producer``: the
    producer is the conductor's loop, so we register that here in a
    deferred manner — the actual ``attach_producer`` call lands inside
    :meth:`_run` once the conductor's loop is running. The UI side
    calls ``attach_consumer`` from its own loop after attaching.
    """
    if self._thread is not None:
        raise ConductorStateError(
            "attach_ui_bridge must be called before start()",
            current=self._state,
        )
    self._ui_bridge = bridge

def procedure_ui_sink(self) -> ProcedureUiSink:
    """Return a UI-only sink procedures can publish
    :class:`~capa.runtime.emissions.ProcedureTick` payloads through.

    Wraps :meth:`_publish_ui` so the procedure layer stays bridge-
    agnostic: a tick goes to the UI mirror with no writer write,
    no data-bus publish. Safe to call before :meth:`start` —
    no-op when no UI bridge is attached (headless run); the sink
    becomes live the moment :meth:`attach_ui_bridge` runs.

    The returned sink captures ``self`` and is intended to be
    invoked from the conductor loop (where the procedure runs).
    Calling from a different loop is undefined; the underlying
    bridge's ``put_nowait`` is thread-safe but the cross-thread
    timestamp ordering would be unreliable.
    """
    return _ConductorProcedureUiSink(conductor=self)

def join(self, timeout: float | None = None) -> bool:
    """Block until the conductor thread exits.

    Returns ``True`` if the thread joined within ``timeout``, ``False``
    otherwise. Intended for tests and for the CLI driver which awaits
    the result future synchronously."""
    if self._thread is None:
        return True
    self._thread.join(timeout=timeout)
    return not self._thread.is_alive()

@classmethod
def from_runtime(
    cls,
    runtime: RuntimeConfig,
    *,
    saturation_deadline_s: float = DEFAULT_SATURATION_DEADLINE_S,
    saturation_poll_period_s: float = DEFAULT_POLL_PERIOD_S,
) -> ConductorConfig:
    """Build a :class:`ConductorConfig` from the user-facing
    :class:`RuntimeConfig`.

    Saturation parameters remain code defaults. Tests and embedding
    code can override them here for diagnostics; the shipped CLI does
    not expose them as user-facing flags today.
    """
    return cls(
        saturation_deadline_s=saturation_deadline_s,
        saturation_poll_period_s=saturation_poll_period_s,
        shutdown_grace_s=runtime.shutdown_grace_s,
        loop_lag_warn_ms=runtime.loop_lag_warn_ms,
    )

async def open(self) -> RunContext:
    """Materialize per-run resources. Returns the :class:`RunContext`
    that the conductor installs into every worker via ``pool.arm_all``.

    Failure here is fatal: the conductor cannot recover and the run
    ends in FAILED state. The session is responsible for cleaning up
    any partial resources before re-raising.
    """
    ...

def set_outcome(self, outcome: RunOutcome, exit_reason: str | None) -> None:
    """Inform the session what to record at finalize. Called by the
    conductor before :meth:`close`. Set to ``COMPLETED`` by default
    until :meth:`close` is called.
    """
    ...

async def dispatch(self, device: str, cmd: DeviceCommand) -> CommandResult:
    """Issue ``cmd`` against ``device``.

    Routes through the conductor when a run is armed and the
    conductor is dispatchable, else through the pool. The future
    returned by either side is bridged into the caller's loop via
    :func:`asyncio.wrap_future`.
    """
    conductor = self._active_conductor()
    try:
        if conductor is not None:
            fut = conductor.dispatch(device, cmd)
        else:
            fut = self._pool.dispatch(device, cmd)
    except KeyError as exc:
        raise UnknownDeviceError(device) from exc
    return await asyncio.wrap_future(fut)

async def snapshot(self, device: str) -> DeviceEmission:
    """One-shot snapshot of ``device``.

    Same routing rules as :meth:`dispatch`.
    """
    conductor = self._active_conductor()
    try:
        if conductor is not None:
            fut = conductor.snapshot(device)
        else:
            fut = self._pool.snapshot(device)
    except KeyError as exc:
        raise UnknownDeviceError(device) from exc
    return await asyncio.wrap_future(fut)

def camera(self, device_name: str) -> Camera | None:
    """Return the underlying :class:`Camera` handle for ``device_name``.

    Preview JPEGs are NOT consumed through this surface — they ride
    a per-camera :class:`~capa.runtime.bridge.ThreadBridge` owned by
    the pool. Device-probe metadata (UVC ranges, supported resolutions,
    per-resolution fps caps) is NOT consumed through this surface
    either — use :meth:`camera_metadata` for a worker-loop-safe
    snapshot.

    Retained for tests that want to assert the wrapper hosts the
    expected handle; UI code should not introduce new callers. The
    returned handle lives on the worker loop, so touching its methods
    from the qasync loop violates the worker-owns-the-handle invariant.

    Returns ``None`` if the device name is unknown or refers to a
    non-camera adapter.
    """
    try:
        worker = self._pool.worker_for(device_name)
    except Exception:
        return None
    adapter: object | None = worker.adapters.get(device_name)
    if not isinstance(adapter, CameraDeviceAdapter):
        return None
    return adapter.camera

async def camera_metadata(self, device_name: str) -> WebcamMetadata | None:
    """Probe a camera's metadata across loops without touching the handle.

    Submits the read to the worker that owns ``device_name``; the
    worker runs ``camera.snapshot_metadata()`` on its own loop and
    signals the resulting future. :func:`asyncio.wrap_future` bridges
    back to the caller's loop — same pattern as :meth:`dispatch` and
    :meth:`snapshot`.

    Returns ``None`` for adapters whose underlying camera does not
    expose a metadata surface (IR cameras today, plus the safety
    net for any future non-webcam adapter routed here by mistake).
    Card code treats ``None`` as "fall back to static widget defaults".

    Raises :class:`UnknownDeviceError` for names that aren't configured
    — same surface as :meth:`dispatch`. Does **not** route through the
    conductor: metadata is a pool-resident probe with no run-state
    semantics, so the same call path applies whether or not a run is
    armed.
    """
    try:
        fut = self._pool.camera_metadata(device_name)
    except KeyError as exc:
        raise UnknownDeviceError(device_name) from exc
    return await asyncio.wrap_future(fut)

async def device_readback(self, device_name: str) -> object:
    """Probe a device's ``read_state_snapshot()`` across loops.

    Returns the adapter-specific snapshot (e.g.
    :class:`~capa.devices.watlow.WatlowStateSnapshot`) or ``None`` for
    adapters that don't implement it. Card code treats ``None`` as
    "fall back to widget defaults".

    Does **not** route through the conductor: this is a pool-resident
    read used by manual-control cards to prefill their widgets, with no
    run-state semantics. Submits to the worker that owns ``device_name``
    and bridges the future back to the caller's loop.
    """
    try:
        fut = self._pool.device_readback(device_name)
    except KeyError as exc:
        raise UnknownDeviceError(device_name) from exc
    return await asyncio.wrap_future(fut)

def exit_code(self) -> int:
    """Map the outcome to the documented process exit code.

    Returns:
        ``0`` on a clean sealed run, ``1`` on operator abort, ``2``
        on procedure / runtime crash, ``3`` on post-seal verification
        failure, ``5`` otherwise (unknown / partial). See
        [Exit codes](../../docs/reference/exit-codes.md).
    """
    if self.bundle_status == "verification_failed":
        return 3
    if self.run_status == "completed" and self.bundle_status == "sealed":
        return 0
    if self.run_status == "aborted":
        return 1
    if self.run_status == "crashed":
        return 2
    return 5

@classmethod
def from_config(
    cls,
    config: ExperimentConfig,
    *,
    runner_factory: Callable[..., WorkerRunner] | None = None,
    preview_consumer_loop: asyncio.AbstractEventLoop | None = None,
    preview_capacity: int = 4,
) -> WorkerPool:
    """Build a pool from a real :class:`ExperimentConfig`.

    Materialization and validation run synchronously here.
    :class:`~capa.devices.materialize.ConfigMaterializationError`
    surfaces adapter construction failures; :class:`ResourceConflict`
    surfaces grouping conflicts. Both fire before any worker is
    constructed, so a misconfigured config has no hardware side
    effects.

    ``preview_consumer_loop``: the loop that will drain preview
    bridges (the qasync UI loop in production; the test loop in
    integration tests). When ``None`` (headless), no preview bridges
    are constructed and the camera adapter's preview lifecycle
    no-ops — zero overhead for headless runs.

    Outbound bridge capacity is derived per worker from each
    adapter's :attr:`DeviceAdapter.expected_emission_rate_hz` inside
    :func:`build_workers`; the previous ``outbound_capacity`` kwarg
    has been retired.
    """
    from capa.devices.materialize import materialize_adapters  # noqa: PLC0415
    from capa.runtime.build import build_workers  # noqa: PLC0415

    materialized = materialize_adapters(config)
    preview_bridges: dict[str, ThreadBridge[PreviewFrame]] = {}
    if preview_consumer_loop is not None:
        for cam in config.hardware.cameras:
            preview_bridges[cam.name] = ThreadBridge(
                name=f"preview-{cam.name}",
                capacity=preview_capacity,
                consumer_loop=preview_consumer_loop,
                policy=BridgePolicy.DROP_OLDEST,
            )
    workers, device_to_resource = build_workers(
        materialized,
        runner_factory=runner_factory,
        preview_bridges=preview_bridges,
    )
    return cls(
        workers=workers,
        device_to_resource=device_to_resource,
        preview_bridges=preview_bridges,
    )

def preview_bridges(self) -> Mapping[str, ThreadBridge[PreviewFrame]]:
    """Per-camera preview bridges, keyed by camera spec name.

    Empty when the pool was constructed without a UI consumer loop
    (headless runs). The mapping itself is read-only — callers that
    need to spawn drainers should iterate the entries.
    """
    return MappingProxyType(self._preview_bridges)

def attach_preview_consumers(self) -> None:
    """Bind every preview bridge's consumer side to the running loop.

    Must be called from the consumer loop (the qasync UI loop) before
    :meth:`open` is awaited — workers attach their producers inside
    :meth:`Worker._open_all_impl` and a producer attach can race a
    consumer-side enqueue. Calling this first builds the
    :class:`asyncio.Queue` so the eventual producer always has a
    target.

    Idempotent: a second call is a no-op.
    """
    if self._preview_consumers_attached:
        return
    for bridge in self._preview_bridges.values():
        bridge.attach_consumer()
    self._preview_consumers_attached = True

def close_preview_bridges(self) -> None:
    """Close every preview bridge. Called from :meth:`close` after
    all workers have closed, so no producer can write afterwards.

    Idempotent: closing an already-closed bridge is a no-op (see
    :meth:`ThreadBridge.close`).
    """
    for bridge in self._preview_bridges.values():
        bridge.close()

def worker_for(self, device_name: str) -> Worker:
    """Return the worker hosting the named adapter.

    Raises :class:`UnknownDeviceError` if the name is not configured.
    """
    try:
        rid = self._device_to_resource[device_name]
    except KeyError as exc:
        raise UnknownDeviceError(device_name, configured_names=self.device_names) from exc
    return self._workers[rid]

async def open(self, progress_callback: OpenProgressCallback | None = None) -> None:
    """Start every worker; on first failure, roll back in reverse order.

    Workers start in parallel — bus-collision avoidance is the
    ``resource_id`` grouping job, not a sequential warm-up
    responsibility. At 6 workers this cuts pool-open from a ~2 s
    sequential warm-up to ~500 ms.

    On any worker's start failure: every already-opened worker is
    closed in reverse order (LIFO of completion), then the original
    exception propagates. The pool returns to :attr:`PoolState.CLOSED`.
    """
    async with self._state_lock:
        self._transition(PoolState.OPENING)

        # Kick off every worker's start in parallel. Use async_start() so
        # we don't need to wrap futures — we get native coroutines that
        # can be gathered directly.
        start_tasks: list[tuple[Worker, asyncio.Task[None]]] = []
        for worker in self._workers.values():
            task = asyncio.create_task(worker.async_start(progress_callback=progress_callback))
            start_tasks.append((worker, task))

        # Wait for ALL to complete (success or failure). We don't want
        # to abandon in-progress starts on first failure — they may
        # leave half-opened threads behind. Collect outcomes, then
        # decide rollback policy.
        await asyncio.gather(*(t for _, t in start_tasks), return_exceptions=True)

        # Categorize.
        opened: list[Worker] = []
        failed: list[tuple[Worker, BaseException]] = []
        for worker, task in start_tasks:
            exc = task.exception()
            if exc is None:
                opened.append(worker)
            else:
                failed.append((worker, exc))

        if not failed:
            self._transition(PoolState.OPEN)
            _logger.info(
                "pool.open",
                worker_count=len(self._workers),
                resources=tuple(self._workers),
            )
            return

        # Rollback path. Close every successfully-opened worker in
        # REVERSE order so paired resources (think watlowlib + serial)
        # tear down LIFO. We swallow rollback errors — they cannot
        # mask the original cause, which is the first failure.
        _logger.warning(
            "pool.open_partial_failure",
            failed_count=len(failed),
            opened_count=len(opened),
            first_error=str(failed[0][1]),
        )
        self._transition(PoolState.CLOSING)
        for worker in reversed(opened):
            # async_close() now returns a structured result rather than
            # raising on adapter-level errors. Any unexpected
            # exception (impl crash) gets swallowed here so the
            # original open failure can propagate; degraded close
            # outcomes are logged but never mask the open cause.
            try:
                rollback_result = await worker.async_close(grace_s=5.0)
            except Exception as rollback_exc:
                _logger.warning(
                    "pool.open_rollback_close_crashed",
                    resource_id=worker.resource_id,
                    error=str(rollback_exc),
                )
                for adapter in worker.adapters.values():
                    self._emit_open_progress(
                        progress_callback,
                        worker=worker,
                        adapter=adapter,
                        status=DeviceInitStatus.ROLLED_BACK,
                        detail=f"rollback close failed: {rollback_exc}",
                    )
                continue
            if not rollback_result.runner_stop.joined or (rollback_result.adapter_close_errors):
                _logger.warning(
                    "pool.open_rollback_close_degraded",
                    resource_id=worker.resource_id,
                    adapter_close_errors=rollback_result.adapter_close_errors,
                    runner_joined=rollback_result.runner_stop.joined,
                )
            detail = "rolled back after another device failed"
            if rollback_result.adapter_close_errors or not rollback_result.runner_stop.joined:
                detail = "rollback degraded; see logs"
            for adapter in worker.adapters.values():
                self._emit_open_progress(
                    progress_callback,
                    worker=worker,
                    adapter=adapter,
                    status=DeviceInitStatus.ROLLED_BACK,
                    detail=detail,
                )
        # Close preview bridges too — producers are gone now, so any
        # pending UI-side drainer wakes with ThreadBridgeClosedError
        # and exits cleanly.
        self.close_preview_bridges()
        self._transition(PoolState.CLOSED)
        # Propagate the first failure — the caller wants the original
        # exception type and message, not a synthetic aggregate.
        raise failed[0][1]

async def shutdown_close(self, *, grace_s: float = 5.0) -> PoolCloseResult:
    """Best-effort close for the :class:`ShutdownCoordinator`.

    Unlike :meth:`close`, this does NOT refuse on non-IDLE workers.
    The coordinator has already attempted to abort the active run by
    the time it calls here, but a worker may still be in
    ``ARMED``/``SAMPLING`` if the conductor's drain itself wedged.
    This method:

    1. Disarms any worker still in ARMED/SAMPLING (parallel, bounded
       by ``grace_s``). Disarm failures are captured as pool errors,
       never raised — the goal is to release as much hardware as we
       can before the coordinator's hard fuse fires.
    2. Closes every worker in parallel, exactly like :meth:`close`.
    3. Returns the aggregate :class:`PoolCloseResult` so the
       coordinator can record non-IDLE entry and any disarm
       timeouts in its :class:`ShutdownResult.errors`.

    The strict :meth:`close` stays for config-reload teardown where
    IDLE is a real invariant.
    """
    async with self._state_lock:
        if self._state is PoolState.CLOSED:
            return PoolCloseResult(clean=True, worker_results=(), errors=())

        pool_errors: list[str] = []

        # Step 1: disarm any non-IDLE workers, in parallel. Workers
        # that are already IDLE skip this step (worker.async_disarm() would
        # raise). Failures are recorded as pool errors but never
        # short-circuit — we still want to close every adapter we can.
        disarm_tasks: dict[str, asyncio.Task[DisarmResult]] = {}
        for rid, worker in self._workers.items():
            if worker.state in (WorkerState.ARMED, WorkerState.SAMPLING):
                disarm_tasks[rid] = asyncio.create_task(worker.async_disarm(grace_s=grace_s))
        if disarm_tasks:
            await asyncio.gather(*disarm_tasks.values(), return_exceptions=True)
            for rid, task in disarm_tasks.items():
                exc = task.exception()
                if exc is not None:
                    msg = f"worker {rid!r} shutdown disarm failed: {exc!r}"
                    pool_errors.append(msg)
                    _logger.warning(
                        "pool.shutdown_close_disarm_failed",
                        resource_id=rid,
                        error=str(exc),
                    )

        # Capture any worker that remained non-IDLE after the disarm
        # attempt — these may still have running stream tasks that
        # close() will see; we record but proceed.
        still_non_idle = [
            (w.resource_id, w.state.value)
            for w in self._workers.values()
            if w.state is not WorkerState.IDLE
        ]
        if still_non_idle:
            pool_errors.append(f"workers non-IDLE entering close: {still_non_idle}")
            _logger.warning(
                "pool.shutdown_close_workers_non_idle",
                non_idle=tuple(still_non_idle),
            )

        self._transition(PoolState.CLOSING)

        # Step 2: close every worker. Only close IDLE workers; non-IDLE
        # workers are recorded as errors.
        idle_workers = [w for w in self._workers.values() if w.state is WorkerState.IDLE]
        for worker in self._workers.values():
            if worker.state is not WorkerState.IDLE:
                pool_errors.append(f"worker {worker.resource_id!r} non-IDLE; close skipped")

        close_results, close_errors = await self._close_workers_parallel(
            idle_workers, grace_s=grace_s
        )
        pool_errors.extend(close_errors)

        self.close_preview_bridges()
        self._transition(PoolState.CLOSED)
        clean = not pool_errors and all(
            not w.adapter_close_errors and not w.adapter_stop_errors and w.runner_stop.joined
            for w in close_results
        )
        _logger.info(
            "shutdown.pool_close_result",
            worker_count=len(self._workers),
            clean=clean,
            pool_errors=tuple(pool_errors),
        )
        return PoolCloseResult(
            clean=clean,
            worker_results=tuple(close_results),
            errors=tuple(pool_errors),
        )