The witnessed genesis: CORA records a TomoScan capture it only watched

apps/api/src/cora/agent/__init__.py

@@ -87,6 +87,10 @@
     RUN_SUPERVISOR_AGENT_ID,
     seed_run_supervisor_agent,
 )
+from cora.agent.seed_run_witness import (
+    RUN_WITNESS_AGENT_ID,
+    seed_run_witness_agent,
+)
 from cora.agent.tools import register_agent_tools
 from cora.agent.wire import AgentHandlers, wire_agent
 
@@ -102,6 +106,7 @@
     "RATIFICATION_ENFORCER_AGENT_ID",
     "RUN_INITIATOR_AGENT_ID",
     "RUN_SUPERVISOR_AGENT_ID",
+    "RUN_WITNESS_AGENT_ID",
     "AgentHandlers",
     "CautionProposalMalformedError",
     "CautionProposalNotActionableError",
@@ -129,5 +134,6 @@
     "seed_run_debriefer_agent",
     "seed_run_initiator_agent",
     "seed_run_supervisor_agent",
+    "seed_run_witness_agent",
     "wire_agent",
 ]

apps/api/src/cora/agent/seed_run_witness.py

@@ -0,0 +1,108 @@
+"""Bootstrap-time seed for the RunWitness Agent.
+
+The RunWitness runtime (`cora.api._run_witness`) needs an Agent record
+(and its co-registered Actor) to exist at the pinned `RUN_WITNESS_AGENT_ID`
+so it can issue `RecordWitnessedRun` as an agent-kind principal when it
+promotes a BEGUN capture observation to a real witnessed Run. Mirrors
+`cora.agent.seed_run_supervisor.seed_run_supervisor_agent` verbatim except
+for the per-agent constants below; the shared scaffolding lives in
+`cora.agent._agent_seed`.
+
+  - Pinned UUID continues the numeric-mnemonic range RunInitiator opened
+    at `1111` (the lettered `aaaa`/`bbbb`/`cccc`/`dddd`/`eeee`/`ffff`
+    blocks are all claimed); deployment-stable forever.
+  - DETERMINISTIC agent (rule-based, NOT LLM): no prompt template
+    (`prompt_template_id=None`) and a sentinel `ModelRef`
+    (`provider="deterministic"`). Never used to build an LLM: the
+    runtime is a substrate-observation loop, not an LLM subscriber.
+  - Authorization: the runtime issues `RecordWitnessedRun` through the
+    Authorize port like any principal. Under the default AllowAllAuthorize
+    it is permitted; under TrustAuthorize the operator's single configured
+    Policy must include this principal + {RecordWitnessedRun, ListRuns}.
+    ListRuns is the restart-rebuild read: without it a restart cannot
+    rediscover which captures are already open, and would re-promote
+    them. Without the RecordWitnessedRun grant, a real BEGUN observation
+    logs `run_witness.promotion_unauthorized` and stays IDLE (retried on
+    the next BEGUN, same posture as RunInitiator's StartRun grant).
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from uuid import UUID
+
+from cora.agent._agent_seed import AgentSeedIdentity, seed_agent
+from cora.agent.aggregates.agent import ModelRef
+
+if TYPE_CHECKING:
+    from cora.infrastructure.kernel import Kernel
+
+
+# ---------------------------------------------------------------------------
+# RunWitness agent identity (deployment-stable constants)
+# ---------------------------------------------------------------------------
+
+# Treat as FOREVER-STABLE. Same change-cost rationale as
+# `RUN_SUPERVISOR_AGENT_ID` / `RUN_INITIATOR_AGENT_ID`: changing this
+# orphans every prior RunWitness-authored Run's principal_id pointer.
+# UUID continues the `1111`-opened numeric range at `2222` (next unclaimed
+# block).
+RUN_WITNESS_AGENT_ID = UUID("01900000-0000-7000-8000-000022220010")
+RUN_WITNESS_AGENT_NAME = "RunWitness"
+RUN_WITNESS_AGENT_KIND = "RunWitness"
+RUN_WITNESS_AGENT_VERSION = "1.0.0"
+RUN_WITNESS_AGENT_DESCRIPTION = (
+    "Deterministic in-process runtime: promotes a real Witnessed "
+    "Run via record_witnessed_run when it observes an external tool "
+    "(TomoScan) begin a capture, with per-capture-code dedup so a single "
+    "in-progress capture is never promoted twice. Not a control path: it "
+    "never drives the substrate, only records that a capture already "
+    "began."
+)
+
+
+# Sentinel model ref: RunWitness is rule-based, not an LLM agent. The
+# Agent aggregate requires a ModelRef; this value is never used to build
+# an LLM (no subscriber / no build_llm call for this agent).
+_DETERMINISTIC_MODEL_REF = ModelRef(
+    provider="deterministic",
+    model="agent:RunWitness:v1",
+    snapshot_pin=None,
+)
+
+
+# ---------------------------------------------------------------------------
+# Deterministic IDs for the bootstrap write envelope
+# ---------------------------------------------------------------------------
+
+_AGENT_EVENT_ID = UUID("01900000-0000-7000-8000-000022220012")
+_ACTOR_EVENT_ID = UUID("01900000-0000-7000-8000-000022220013")
+_BOOTSTRAP_CORRELATION_ID = UUID("01900000-0000-7000-8000-000022220014")
+
+
+async def seed_run_witness_agent(kernel: Kernel) -> None:
+    """Seed the RunWitness Agent + co-registered Actor (idempotent)."""
+    identity = AgentSeedIdentity(
+        agent_id=RUN_WITNESS_AGENT_ID,
+        name=RUN_WITNESS_AGENT_NAME,
+        kind=RUN_WITNESS_AGENT_KIND,
+        version=RUN_WITNESS_AGENT_VERSION,
+        description=RUN_WITNESS_AGENT_DESCRIPTION,
+        model_ref=_DETERMINISTIC_MODEL_REF,
+        prompt_template_id=None,
+        agent_event_id=_AGENT_EVENT_ID,
+        actor_event_id=_ACTOR_EVENT_ID,
+        correlation_id=_BOOTSTRAP_CORRELATION_ID,
+        command_name="SeedRunWitnessAgent",
+    )
+    await seed_agent(kernel, identity)
+
+
+__all__ = [
+    "RUN_WITNESS_AGENT_DESCRIPTION",
+    "RUN_WITNESS_AGENT_ID",
+    "RUN_WITNESS_AGENT_KIND",
+    "RUN_WITNESS_AGENT_NAME",
+    "RUN_WITNESS_AGENT_VERSION",
+    "seed_run_witness_agent",
+]

apps/api/src/cora/api/_capture_observer.py

@@ -0,0 +1,237 @@
+"""Composition-root bridge: drive the capture observer from ControlPort.
+
+The Run BC's `CaptureObserver` port is BC-local (`cora.run.ports`) and
+the `ControlPort` value-IO it needs is Operation-BC-owned
+(`cora.operation.ports`). tach forbids `cora.run -> cora.operation`, so
+the adapter that bridges the two lives here at the composition root,
+mirroring `_enclosure_permit_observer.py` exactly. If a third cross-BC
+`ControlPort` consumer appears, the rule-of-three move is to hoist
+`ControlPort` to `cora.infrastructure.ports`.
+
+Maps each configured capture code's `status` PV to a `CaptureObservation`
+by looking its decoded text up in the deployment's declared
+`capture_status_phases` table (`classify_capture_status`); a literal
+absent from the table classifies `UNRECOGNIZED` rather than being
+dropped or coerced into a nearby phase.
+
+## One deliberate inversion from the Enclosure precedent
+
+`ControlPortEnclosureObserver` synthesizes an `Unknown` STATUS CLAIM on
+disconnect, because a dead permit signal must fail the run-start gate
+closed rather than leave a stale `Permitted` standing. There is no such
+gate here, and reading a disconnect as a real observation would
+fabricate one: it would either assert a phase no substrate reading
+backs, or, if mapped to `UNRECOGNIZED`, misrepresent a communication
+failure as a vocabulary problem. So `_unreached` here carries NO status
+claim at all (`reported_status=None`, `phase=None`), the same shape a
+probe-only poll tick already uses, exactly mirroring
+`CaptureObservation`'s own port-level contract for the probe-only case.
+
+## Permit probe trail's sibling-poller shape, reused unchanged
+
+Same reasoning as the Enclosure adapter: `_poll` is a SIBLING of
+`_pump`, not nested inside it, because `_pump` returns as soon as its
+subscription ends and `_drain` only re-subscribes after every pump has
+returned; a poller living inside `_pump` would die with it and could
+never observe the PV's recovery.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import TYPE_CHECKING
+
+from cora.operation.ports.control_port import ControlNotConnectedError
+from cora.run.ports.capture_observer import CaptureObservation, CaptureObserverScope, CapturePhase
+from cora.shared.reach import ReachTier
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncGenerator, Mapping
+
+    from cora.operation.ports.control_port import ControlPort, Measurement
+
+_SOURCE_KIND = "EpicsPv"
+_STATUS_ROLE = "status"
+
+
+def classify_capture_status(reported_status: str, status_phases: Mapping[str, str]) -> CapturePhase:
+    """Classify a decoded status literal against the deployment's declared table.
+
+    A literal absent from `status_phases` classifies as
+    `CapturePhase.UNRECOGNIZED`. This also covers the Bad-quality-reading
+    case with no special handling needed: an unresolvable or garbled
+    string is exceedingly unlikely to match a declared literal, so it
+    naturally falls to UNRECOGNIZED, which is the correct signal ("the
+    substrate said something CORA cannot classify"), rather than a
+    fabricated phase.
+    """
+    mapped = status_phases.get(reported_status)
+    if mapped is None:
+        return CapturePhase.UNRECOGNIZED
+    return CapturePhase(mapped)
+
+
+class _PumpDone:
+    """Per-PV sentinel pushed onto the merge queue when a pump exits."""
+
+    __slots__ = ()
+
+
+_PUMP_DONE = _PumpDone()
+
+
+class ControlPortCaptureObserver:
+    """`CaptureObserver` over a `ControlPort` (one `status` PV per capture code).
+
+    `capture_pvs` is code -> role -> PV, matching
+    `Settings.capture_watch_pvs`. Only the `status` role is subscribed
+    in this slice; a code whose PV set has no `status` entry cannot be
+    watched and is silently excluded from scope, mirroring the
+    Enclosure adapter's `if code in self._permit_pvs` filter. The other
+    declared roles (`server_running`, `abort`, `images_saved`,
+    `images_collected`) are read by a later slice; declaring them now
+    costs nothing and lets a deployment's config stabilize ahead of the
+    code that consumes it.
+    """
+
+    def __init__(
+        self,
+        *,
+        control_port: ControlPort,
+        capture_pvs: Mapping[str, Mapping[str, str]],
+        status_phases: Mapping[str, str],
+        tick_seconds: float | None = None,
+    ) -> None:
+        self._control_port = control_port
+        self._status_pvs = {
+            code: roles[_STATUS_ROLE]
+            for code, roles in capture_pvs.items()
+            if _STATUS_ROLE in roles
+        }
+        self._status_phases = dict(status_phases)
+        self._tick_seconds = tick_seconds
+
+    def observe(self, scope: CaptureObserverScope) -> AsyncGenerator[CaptureObservation]:
+        return self._drain(scope)
+
+    async def _drain(self, scope: CaptureObserverScope) -> AsyncGenerator[CaptureObservation]:
+        pvs = [
+            (code, self._status_pvs[code])
+            for code in sorted(scope.capture_codes)
+            if code in self._status_pvs
+        ]
+        if not pvs:
+            return
+        queue: asyncio.Queue[CaptureObservation | _PumpDone] = asyncio.Queue()
+        pump_tasks = [asyncio.create_task(self._pump(code, pv, queue)) for code, pv in pvs]
+        poll_tasks = (
+            [asyncio.create_task(self._poll(code, pv, queue)) for code, pv in pvs]
+            if self._tick_seconds is not None
+            else []
+        )
+        tasks = pump_tasks + poll_tasks
+        # Only pumps ever signal completion; a poller runs until the
+        # `finally` below cancels it, so it must not hold this open.
+        remaining = len(pump_tasks)
+        try:
+            while remaining > 0:
+                item = await queue.get()
+                if isinstance(item, _PumpDone):
+                    remaining -= 1
+                    continue
+                yield item
+            # Every pump has finished, but a still-running poller can have
+            # enqueued a probe in the same instant the final _PumpDone was
+            # read. Drain exactly what is ALREADY queued right now,
+            # synchronously, into a list before yielding any of it: see
+            # `ControlPortEnclosureObserver._drain` for the full reasoning.
+            pending = queue.qsize()
+            leftover = [queue.get_nowait() for _ in range(pending)]
+            for item in leftover:
+                if not isinstance(item, _PumpDone):
+                    yield item
+        finally:
+            for task in tasks:
+                task.cancel()
+            await asyncio.gather(*tasks, return_exceptions=True)
+
+    async def _pump(
+        self,
+        code: str,
+        pv: str,
+        queue: asyncio.Queue[CaptureObservation | _PumpDone],
+    ) -> None:
+        try:
+            async for reading in self._control_port.subscribe(pv):
+                queue.put_nowait(self._from_reading(code, pv, reading))
+            # Clean stream end: no status claim, mirroring a disconnect.
+            queue.put_nowait(self._unreached(code, pv))
+        except ControlNotConnectedError:
+            queue.put_nowait(self._unreached(code, pv))
+        finally:
+            queue.put_nowait(_PUMP_DONE)
+
+    async def _poll(
+        self,
+        code: str,
+        pv: str,
+        queue: asyncio.Queue[CaptureObservation | _PumpDone],
+    ) -> None:
+        """Re-affirm reach to `pv` every `_tick_seconds`, independent of push.
+
+        Never pushes `_PumpDone`: a sibling of `_pump`, not a stage in
+        its lifecycle. See `ControlPortEnclosureObserver._poll`.
+        """
+        assert self._tick_seconds is not None
+        while True:
+            await asyncio.sleep(self._tick_seconds)
+            try:
+                await self._control_port.read(pv)
+            except Exception:  # any read failure is a failed probe, not a bug
+                queue.put_nowait(self._probe_only(code, pv, ReachTier.UNREACHED))
+            else:
+                queue.put_nowait(self._probe_only(code, pv, ReachTier.RELAYED))
+
+    def _from_reading(self, code: str, pv: str, reading: Measurement) -> CaptureObservation:
+        reported_status = str(reading.value)
+        phase = classify_capture_status(reported_status, self._status_phases)
+        return CaptureObservation(
+            capture_code=code,
+            reported_status=reported_status,
+            phase=phase,
+            reach_tier=ReachTier.RELAYED,
+            observed_at=reading.produced_at,
+            source_kind=_SOURCE_KIND,
+            source_id=pv,
+        )
+
+    def _probe_only(self, code: str, pv: str, reach_tier: ReachTier) -> CaptureObservation:
+        """A poll tick's result: reach evidence with no status claim."""
+        return CaptureObservation(
+            capture_code=code,
+            reported_status=None,
+            phase=None,
+            reach_tier=reach_tier,
+            observed_at=None,
+            source_kind=_SOURCE_KIND,
+            source_id=pv,
+        )
+
+    def _unreached(self, code: str, pv: str) -> CaptureObservation:
+        """A disconnect or clean stream end: no status claim, no phase.
+
+        See this module's docstring, "One deliberate inversion from the
+        Enclosure precedent", for why this must not synthesize a phase.
+        """
+        return CaptureObservation(
+            capture_code=code,
+            reported_status=None,
+            phase=None,
+            reach_tier=ReachTier.UNREACHED,
+            observed_at=None,
+            source_kind=_SOURCE_KIND,
+            source_id=pv,
+        )
+
+
+__all__ = ["ControlPortCaptureObserver", "classify_capture_status"]

apps/api/src/cora/api/_run_initiator.py

@@ -274,6 +274,15 @@ async def initiate_tick(
     the steady state. Each start flows through `initiate_run`, so it is authorized,
     attributed, and Decision-linked exactly as a single agent start.
 
+    `running` deliberately counts EVERY Running Run toward `max_in_flight`
+    regardless of `conduct_mode`: unlike the RunSupervisor's hold / resume /
+    liveness mechanisms, which skip a Witnessed Run because CORA does not
+    control it, this cap exists to serialize access to shared single-stage
+    hardware. A witnessed capture IS occupying that hardware, so excluding
+    it here would let the initiator start a second, driven Run on the same
+    stage an external tool is actively driving, the opposite of what the
+    cap is for.
+
     A per-Subject StartRun denial makes `initiate_run` return None (a logged
     no-op); that Subject is not added to `started`, so a transient fault is
     retried next tick. `max_in_flight <= 0` makes the tick inert (returns []); the