Slice 9: terminals and boundaries for the witnessed path

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

@@ -15,15 +15,31 @@
     (`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
+  - Authorization: the runtime issues four distinct commands 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).
+    all four are permitted; under TrustAuthorize the operator's single
+    configured Policy must include this principal + {RecordWitnessedRun,
+    RecordWitnessedRunOutcome, TruncateRun, 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). Without RecordWitnessedRunOutcome, a real terminal logs
+    `run_witness.outcome_unauthorized` and leaves the Run open (retried
+    on the next BEGUN via truncation). Without TruncateRun, a missed
+    terminal cannot be recovered and logs `run_witness.truncate_unauthorized`,
+    but the new capture still promotes regardless.
+
+    UNLIKE the RecordWitnessedRunOutcome grant, TruncateRun's decider
+    carries no `conduct_mode` gate (it accepts any Running-or-Held Run,
+    same as every other operator-facing terminal). This principal's
+    safety therefore rests on `_run_witness.py`'s own bookkeeping
+    discipline: `_truncate_stale` only ever supplies a `run_id` it
+    popped from its own `_open_captures` dict, which is populated
+    exclusively by this same runtime's own promotions, so it can only
+    ever name a Run it created. A future change to `_run_witness.py`
+    that sources a `run_id` for this call from anywhere else would lose
+    that guarantee with no decider-level backstop to catch it.
 """
 
 from __future__ import annotations

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

@@ -52,6 +52,40 @@
 
 _SOURCE_KIND = "EpicsPv"
 _STATUS_ROLE = "status"
+_ABORT_ROLE = "abort"
+
+# Conventional EPICS binary state labels, mirroring
+# `_enclosure_permit_observer._PERMITTED_LABELS` / `_NOT_PERMITTED_LABELS`
+# exactly: a DBR_ENUM reading through `EpicsCaControlPort` reaches this
+# module as its resolved label, never as its index (2-BM's `AbortScan`
+# is confirmed live as one such ENUM, resolving to `'No'`), so the label
+# is the only thing left to compare against. `CaprotoControlPort`, by
+# contrast, leaves the raw integer unresolved; the `int(value)` fallback
+# below covers that shape too.
+_ASSERTED_LABELS = frozenset({"1", "ON", "TRUE", "YES"})
+_CLEAR_LABELS = frozenset({"0", "OFF", "FALSE", "NO"})
+
+
+def _binary_code(value: object) -> int | None:
+    """Resolve a binary-role reading to 1 / 0, or None when it is neither.
+
+    Unrecognized resolves to `None`, never a guess: same fail-toward-
+    silence posture as `_enclosure_permit_observer._binary_code`, whose
+    docstring documents the production incident (`int('ON')` raising)
+    that made string-label matching necessary in the first place.
+    """
+    if isinstance(value, str):
+        token = value.strip().upper()
+        if token in _ASSERTED_LABELS:
+            return 1
+        if token in _CLEAR_LABELS:
+            return 0
+        return None
+    try:
+        code = int(value)  # type: ignore[call-overload]
+    except (TypeError, ValueError):
+        return None
+    return code if code in (0, 1) else None
 
 
 def classify_capture_status(reported_status: str, status_phases: Mapping[str, str]) -> CapturePhase:
@@ -81,17 +115,23 @@ class _PumpDone:
 
 
 class ControlPortCaptureObserver:
-    """`CaptureObserver` over a `ControlPort` (one `status` PV per capture code).
+    """`CaptureObserver` over a `ControlPort` (`status` + optional `abort` 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.
+    `Settings.capture_watch_pvs`. The `status` role is required; 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 `abort` role is optional
+    per code: when declared, a truthy reading on it is a direct
+    `ABORTED` phase claim, letting a real abort be distinguished from a
+    successful end even where the `status` PV alone cannot (2-BM's
+    `fly_scan` writes the identical `'Scan complete'` literal on both).
+    A code with no `abort` entry watches `status` only, exactly as
+    before this role existed. The remaining declared roles
+    (`server_running`, `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__(
@@ -108,6 +148,9 @@ def __init__(
             for code, roles in capture_pvs.items()
             if _STATUS_ROLE in roles
         }
+        self._abort_pvs = {
+            code: roles[_ABORT_ROLE] for code, roles in capture_pvs.items() if _ABORT_ROLE in roles
+        }
         self._status_phases = dict(status_phases)
         self._tick_seconds = tick_seconds
 
@@ -122,8 +165,15 @@ async def _drain(self, scope: CaptureObserverScope) -> AsyncGenerator[CaptureObs
         ]
         if not pvs:
             return
+        abort_pvs = [
+            (code, self._abort_pvs[code])
+            for code in sorted(scope.capture_codes)
+            if code in self._abort_pvs
+        ]
         queue: asyncio.Queue[CaptureObservation | _PumpDone] = asyncio.Queue()
-        pump_tasks = [asyncio.create_task(self._pump(code, pv, queue)) for code, pv in pvs]
+        pump_tasks = [asyncio.create_task(self._pump(code, pv, queue)) for code, pv in pvs] + [
+            asyncio.create_task(self._pump_abort(code, pv, queue)) for code, pv in abort_pvs
+        ]
         poll_tasks = (
             [asyncio.create_task(self._poll(code, pv, queue)) for code, pv in pvs]
             if self._tick_seconds is not None
@@ -171,6 +221,29 @@ async def _pump(
         finally:
             queue.put_nowait(_PUMP_DONE)
 
+    async def _pump_abort(
+        self,
+        code: str,
+        pv: str,
+        queue: asyncio.Queue[CaptureObservation | _PumpDone],
+    ) -> None:
+        """Sibling pump for the optional `abort` role.
+
+        Unlike `_pump`, not every reading is enqueued: a falsy value (the
+        busy record's idle/reset state) makes no phase claim at all and
+        must not be pushed as a no-op observation, per `_from_abort_reading`.
+        """
+        try:
+            async for reading in self._control_port.subscribe(pv):
+                observation = self._from_abort_reading(code, pv, reading)
+                if observation is not None:
+                    queue.put_nowait(observation)
+            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,
@@ -205,6 +278,36 @@ def _from_reading(self, code: str, pv: str, reading: Measurement) -> CaptureObse
             source_id=pv,
         )
 
+    def _from_abort_reading(
+        self, code: str, pv: str, reading: Measurement
+    ) -> CaptureObservation | None:
+        """An asserted abort-role reading is a direct `ABORTED` claim.
+
+        NOT Python truthiness: 2-BM's `AbortScan` is a DBR_ENUM that
+        resolves to the label `'No'` when idle, and `bool('No')` is
+        `True`. `_binary_code` decodes the conventional EPICS binary
+        labels (or a raw 0/1 index) instead, so "No" correctly resolves
+        to clear, not asserted.
+
+        A clear or unresolvable reading makes no phase claim: it is not
+        "the capture is not aborted" so much as "nothing happened on
+        this PV" (or a label this cannot decode), and reporting it as a
+        phase would let a stale idle read silently arrive between a
+        real `Begun` and its `status` transitions. `None` return means
+        the caller enqueues nothing.
+        """
+        if _binary_code(reading.value) != 1:
+            return None
+        return CaptureObservation(
+            capture_code=code,
+            reported_status=str(reading.value),
+            phase=CapturePhase.ABORTED,
+            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(