Give the procedure watcher a separate staleness window for Held

apps/api/src/cora/api/_procedure_watcher.py

@@ -12,25 +12,37 @@
 ## What v1 does
 
 Each tick it lists in-conduct procedures (`Running` and `Held`), selects those
-that have sat past the operator-config staleness window without progressing, and
-records one `Decision(context=ProcedureProgress, choice=Stall)` per stall
-EPISODE. It is FLAG-ONLY: it issues NO command (it cannot un-stick a conduct; it
-surfaces the stall so a human acts before an experiment hangs unnoticed
-mid-procedure). Procedure is a distinct aggregate from Run, so this liveness gap
-is one `_run_supervisor` does not cover.
+that have sat past their status's operator-config staleness window without
+progressing, and records one `Decision(context=ProcedureProgress, choice=Stall)`
+per stall EPISODE. It is FLAG-ONLY: it issues NO command (it cannot un-stick a
+conduct; it surfaces the stall so a human acts before an experiment hangs
+unnoticed mid-procedure). Procedure is a distinct aggregate from Run, so this
+liveness gap is one `_run_supervisor` does not cover.
 
 ## Staleness clock and the active-conduct false-positive guard
 
 `stalled_seconds = now - last_progress_at`.
 
+`Running` and `Held` are clocked against SEPARATE operator-config windows
+(`procedure_watcher_stale_after_seconds` and
+`procedure_watcher_held_stale_after_seconds`), because the two statuses mean
+different things: a `Running` procedure sitting idle for an hour is a plausible
+stall, but a `Held` procedure sitting for an hour is routinely a deliberate
+operator pause (a bakeout, waiting on beam, waiting on a collaborator) that can
+legitimately run for days. One shared window would either false-flag an
+ordinary overnight hold or blind the watcher to a genuinely stuck `Running`
+conduct; this mirrors `_campaign_watcher`'s week-long default for its own
+`Held` window.
+
 For `Held` the conduct is paused and accepts no activity, so
 `last_status_changed_at` (the time it was held, on the list projection) is the
-correct clock. For `Running`, `proj_operation_procedure_summary` advances
-`last_status_changed_at` only on real lifecycle transitions and NO-OPs it for
-`ProcedureActivitiesLogbookOpened` / `ProcedureIterationStarted` (activity is
-orthogonal to lifecycle); so keying on it alone would FALSE-FLAG a procedure
-that is actively logging steps. Therefore a `Running` candidate that already
-looks stale by its status timestamp gets ONE per-candidate
+correct clock, and there is no second-chance fold to take: a held conduct
+cannot log activity, structurally. For `Running`, `proj_operation_procedure_summary`
+advances `last_status_changed_at` only on real lifecycle transitions and
+NO-OPs it for `ProcedureActivitiesLogbookOpened` / `ProcedureIterationStarted`
+(activity is orthogonal to lifecycle); so keying on it alone would FALSE-FLAG a
+procedure that is actively logging steps. Therefore a `Running` candidate that
+already looks stale by its status timestamp gets ONE per-candidate
 `read_procedure_activity_recency` to fold in the latest activity `recorded_at`
 before it is flagged. Bounding that read to already-looks-stale `Running`
 candidates keeps the per-tick cost low. This mirrors `_clearance_watcher`
@@ -93,11 +105,12 @@
 _PAGE_LIMIT = 100
 _CHOICE_STALL = "Stall"
 _STATUS_RUNNING = "Running"
+_STATUS_HELD = "Held"
 
 # The two in-conduct lifecycle states the watcher surveys. Defined (registered,
 # not started) and the terminal states (Completed / Aborted / Truncated) are out
 # of scope: only an active or paused conduct can hang mid-flight.
-_WATCHED_STATUSES: tuple[ProcedureStatusFilter, ...] = ("Running", "Held")
+_WATCHED_STATUSES: tuple[ProcedureStatusFilter, ...] = (_STATUS_RUNNING, _STATUS_HELD)
 
 # Stable namespace for deriving the deterministic Decision id from the procedure
 # id + the stall-episode timestamp (0c0c block, distinct from the seed envelope
@@ -188,7 +201,8 @@ async def _watch_tick(
         return
 
     now = deps.clock.now()
-    stale_after = deps.settings.procedure_watcher_stale_after_seconds
+    stale_after_running = deps.settings.procedure_watcher_stale_after_seconds
+    stale_after_held = deps.settings.procedure_watcher_held_stale_after_seconds
     try:
         items = await _drain_watched_procedures(list_procedures, deps)
     except UnauthorizedError as err:
@@ -209,9 +223,10 @@ async def _watch_tick(
         if base is None:
             # No status-change timestamp recorded: cannot evaluate; defer.
             continue
+        stale_after = stale_after_held if item.status == _STATUS_HELD else stale_after_running
         if not is_stalled(base, now, stale_after):
-            # Fresh by status timestamp. For Running a later activity only makes
-            # it fresher, so skipping here cannot hide a stall.
+            # Fresh by its status's own window. For Running a later activity only
+            # makes it fresher, so skipping here cannot hide a stall.
             continue
         last_progress_at = base
         if item.status == _STATUS_RUNNING:

apps/api/src/cora/infrastructure/config.py

@@ -525,13 +525,20 @@ class Settings(BaseSettings):
     # `procedure_watcher_enabled` gates the ProcedureWatcher background runtime
     # (8th seeded agent, deterministic flag-only). Default off: deployments opt in
     # explicitly. `procedure_watcher_tick_seconds` is the sweep cadence (>= 0.1s).
-    # `procedure_watcher_stale_after_seconds` is how long an in-conduct procedure
-    # (Running / Held) may sit without progressing before it is flagged; live
-    # conduct is far shorter-lived than a clearance or calibration, so the default
-    # is an hour (off by default; an operator sets the real window on enable).
+    # `procedure_watcher_stale_after_seconds` is how long a Running procedure may
+    # sit without progressing before it is flagged; live conduct is far
+    # shorter-lived than a clearance or calibration, so the default is an hour
+    # (off by default; an operator sets the real window on enable).
+    # `procedure_watcher_held_stale_after_seconds` is the separate window for a
+    # Held procedure: a hold is commonly a deliberate operator pause (a bakeout,
+    # waiting on beam, waiting on a collaborator) that legitimately runs far
+    # longer than an hour, and a Held conduct logs no activity to fold in as a
+    # second chance, so it needs its own, much longer default; matches
+    # `campaign_watcher_stale_after_seconds`'s Held precedent, a week.
     procedure_watcher_enabled: bool = False
     procedure_watcher_tick_seconds: float = 300.0
     procedure_watcher_stale_after_seconds: float = 3600.0
+    procedure_watcher_held_stale_after_seconds: float = 604800.0
 
     # `campaign_watcher_enabled` gates the CampaignWatcher background runtime
     # (9th seeded agent, deterministic flag-only). Default off: deployments opt in
@@ -2055,6 +2062,19 @@ def _validate_procedure_watcher_stale_after_seconds(cls, value: float) -> float:
             raise ValueError(msg)
         return value
 
+    @field_validator("procedure_watcher_held_stale_after_seconds")
+    @classmethod
+    def _validate_procedure_watcher_held_stale_after_seconds(cls, value: float) -> float:
+        """Must be positive: a non-positive window would flag every Held
+        procedure."""
+        if value <= 0:
+            msg = (
+                f"procedure_watcher_held_stale_after_seconds must be > 0, got {value}; "
+                "a non-positive window would flag every Held procedure"
+            )
+            raise ValueError(msg)
+        return value
+
     @field_validator("campaign_watcher_tick_seconds")
     @classmethod
     def _validate_campaign_watcher_tick_seconds(cls, value: float) -> float:

apps/api/tests/unit/api/test_procedure_watcher.py

@@ -3,8 +3,10 @@
 Covers the pure staleness rule (is_stalled) on both sides of the inclusive
 boundary, plus a fakes-driven tick that exercises the drain -> flag Decision
 loop for both watched statuses, the Running activity-recency fold (the
-anti-false-flag guard), the Held no-fold path, the defensive status guard, the
-Actor.active revocation gate, idempotency, and the disabled no-op.
+anti-false-flag guard), the Held no-fold path, the separate Running/Held
+staleness windows (a long legitimate Held pause must not flag under the
+default week-long window), the defensive status guard, the Actor.active
+revocation gate, idempotency, and the disabled no-op.
 """
 
 # white-box test of the runtime internals (private functions / constants)
@@ -42,10 +44,12 @@
 
 _NOW = datetime(2026, 6, 22, 12, 0, 0, tzinfo=UTC)
 _STALE_AFTER = 3600.0  # 1 hour
+_HELD_STALE_AFTER = 604800.0  # 1 week, matches the config default for Held
 _OLD = _NOW - timedelta(hours=2)  # clearly stale at a 1-hour window
 _RECENT = _NOW - timedelta(minutes=1)  # fresh
 _STILL_STALE = _NOW - timedelta(minutes=90)  # newer than _OLD but still > window
 _BOUNDARY = _NOW - timedelta(seconds=int(_STALE_AFTER))
+_OVERNIGHT_HELD = _NOW - timedelta(hours=12)  # legitimate pause, still short of a week
 
 
 # ---------- pure rule: is_stalled ----------
@@ -76,10 +80,16 @@ def test_not_stalled_just_under_boundary() -> None:
 # ---------- tick: full loop with fakes ----------
 
 
-def _kernel(*, enabled: bool = False, stale_after: float = _STALE_AFTER) -> Kernel:
+def _kernel(
+    *,
+    enabled: bool = False,
+    stale_after: float = _STALE_AFTER,
+    held_stale_after: float = _HELD_STALE_AFTER,
+) -> Kernel:
     settings = Settings(  # type: ignore[call-arg]
         procedure_watcher_enabled=enabled,
         procedure_watcher_stale_after_seconds=stale_after,
+        procedure_watcher_held_stale_after_seconds=held_stale_after,
     )
     return make_inmemory_kernel(
         settings=settings,
@@ -153,10 +163,12 @@ async def test_tick_flags_stale_running_with_no_activity() -> None:
 @pytest.mark.unit
 async def test_tick_flags_stale_held_without_folding_activity() -> None:
     """A Held conduct accepts no activity, so it is clocked on its status
-    timestamp directly; a (defensively seeded) recent activity is ignored."""
+    timestamp directly; a (defensively seeded) recent activity is ignored.
+    Held's own window is shortened to _STALE_AFTER here (the config default is
+    a week) so that `_OLD` reads as stale for this assertion."""
     from cora.api._procedure_watcher import _watch_tick
 
-    kernel = _kernel()
+    kernel = _kernel(held_stale_after=_STALE_AFTER)
     await seed_procedure_watcher_agent(kernel)
     pid = uuid4()
     list_procedures = _make_list_procedures(
@@ -170,6 +182,31 @@ async def test_tick_flags_stale_held_without_folding_activity() -> None:
     assert await load_decision(kernel.event_store, _derive_decision_id(pid, _OLD)) is not None
 
 
+@pytest.mark.unit
+async def test_tick_does_not_flag_held_overnight_pause() -> None:
+    """Regression test: Held used to share Running's 1-hour window, so any
+    deliberate operator pause longer than an hour (a bakeout, waiting on beam,
+    waiting on a collaborator) raised a false Stall. Under the real default
+    (a week), a 12-hour hold is not stalled."""
+    from cora.api._procedure_watcher import _watch_tick
+
+    kernel = _kernel()  # held_stale_after defaults to the real week-long window
+    await seed_procedure_watcher_agent(kernel)
+    pid = uuid4()
+    list_procedures = _make_list_procedures(
+        [_item(pid, status="Held", last_status_changed_at=_OVERNIGHT_HELD)]
+    )
+
+    await _watch_tick(
+        deps=kernel,
+        list_procedures=list_procedures,
+        activity_lookup=InMemoryProcedureActivityLookup(),
+    )
+
+    decision_id = _derive_decision_id(pid, _OVERNIGHT_HELD)
+    assert await load_decision(kernel.event_store, decision_id) is None
+
+
 @pytest.mark.unit
 async def test_tick_does_not_flag_running_with_recent_activity() -> None:
     """The anti-false-flag fold: a Running procedure that looks stale by its
@@ -462,14 +499,22 @@ def test_procedure_watcher_stale_after_rejects_non_positive() -> None:
         Settings(procedure_watcher_stale_after_seconds=0.0)  # type: ignore[call-arg]
 
 
+@pytest.mark.unit
+def test_procedure_watcher_held_stale_after_rejects_non_positive() -> None:
+    with pytest.raises(ValueError, match="procedure_watcher_held_stale_after_seconds"):
+        Settings(procedure_watcher_held_stale_after_seconds=0.0)  # type: ignore[call-arg]
+
+
 @pytest.mark.unit
 def test_procedure_watcher_settings_accept_valid() -> None:
     settings = Settings(  # type: ignore[call-arg]
         procedure_watcher_tick_seconds=120.0,
         procedure_watcher_stale_after_seconds=7200.0,
+        procedure_watcher_held_stale_after_seconds=1209600.0,
     )
     assert settings.procedure_watcher_tick_seconds == 120.0
     assert settings.procedure_watcher_stale_after_seconds == 7200.0
+    assert settings.procedure_watcher_held_stale_after_seconds == 1209600.0
 
 
 @pytest.mark.unit