feat(cadence): wire p3_velocity into Stop + UPS dispatchers (#876 PR 3/5)

src/aelfrice/hook.py

@@ -2741,6 +2741,7 @@ def _maybe_fire_cadence_checkpoint(
         POLICY_OFF,
         POLICY_P1_EVERY_K_TURNS,
         POLICY_P2_CTX_THRESHOLD,
+        POLICY_P3_VELOCITY,
         append_shadow_row,
         estimate_transcript_bytes,
         format_shadow_row,
@@ -2749,15 +2750,20 @@ def _maybe_fire_cadence_checkpoint(
         resolve_cadence_ctx_threshold,
         resolve_cadence_enabled,
         resolve_cadence_k,
+        resolve_cadence_p3_velocity_threshold,
         resolve_cadence_policy,
         resolve_cadence_shadow_mode_enabled,
         shadow_log_path,
         should_fire,
         should_fire_p2,
+        should_fire_p3_velocity,
         would_fire_p1,
         would_fire_p2,
     )
-    from aelfrice.session_ring import read_ring_state  # noqa: PLC0415
+    from aelfrice.session_ring import (  # noqa: PLC0415
+        read_ring_state,
+        update_p3_velocity_state,
+    )
 
     cwd_obj = payload.get(_CWD_KEY)
     cwd = (
@@ -2845,6 +2851,66 @@ def _maybe_fire_cadence_checkpoint(
         )
         return
 
+    if policy == POLICY_P3_VELOCITY:
+        threshold = resolve_cadence_p3_velocity_threshold(start=cwd)
+        cfg = CadenceConfig(
+            enabled=True, policy=policy, p3_velocity_threshold=threshold,
+        )
+        state = read_ring_state(session_id)
+        if not isinstance(state, dict):
+            return
+        raw_next: Any = state.get("next_fire_idx")
+        raw_bytes_last: Any = state.get("bytes_at_last_fire", 0)
+        raw_fire_last: Any = state.get("fire_idx_at_last_fire", 0)
+        if (
+            isinstance(raw_next, bool) or not isinstance(raw_next, int)
+            or isinstance(raw_bytes_last, bool) or not isinstance(raw_bytes_last, int)
+            or isinstance(raw_fire_last, bool) or not isinstance(raw_fire_last, int)
+        ):
+            return
+        next_fire_idx = raw_next
+        bytes_at_last_fire = raw_bytes_last
+        fire_idx_at_last_fire = raw_fire_last
+        turns_since_last_fire = next_fire_idx - fire_idx_at_last_fire
+        if turns_since_last_fire <= 0:
+            return
+        tp_obj = payload.get(_TRANSCRIPT_PATH_KEY)
+        tp: Path | None
+        if isinstance(tp_obj, str) and tp_obj:
+            tp = Path(tp_obj)
+        elif isinstance(tp_obj, os.PathLike):
+            tp = Path(tp_obj)
+        else:
+            tp = None
+        transcript_bytes = estimate_transcript_bytes(tp)
+        if not should_fire_p3_velocity(
+            bytes_at_last_fire=bytes_at_last_fire,
+            transcript_bytes=transcript_bytes,
+            turns_since_last_fire=turns_since_last_fire,
+            config=cfg,
+        ):
+            return
+        body = _run_cadence_rebuild(payload, cwd)
+        if body is None:
+            return
+        _write_cadence_resume_cache(body, session_id, policy, serr)
+        # Update both p3-velocity state slots atomically so the next fire's
+        # density calculation sees consistent (bytes, fire_idx) inputs.
+        update_p3_velocity_state(
+            session_id,
+            transcript_bytes=transcript_bytes,
+            fire_idx=next_fire_idx,
+            stderr=serr,
+        )
+        density = (transcript_bytes - bytes_at_last_fire) / turns_since_last_fire
+        print(
+            f"aelfrice: cadence checkpoint fired @ fire_idx={next_fire_idx} "
+            f"(policy={policy}, velocity={density:.1f} bytes/turn, "
+            f"threshold={threshold})",
+            file=serr,
+        )
+        return
+
     # Unknown policy / POLICY_OFF — no-op.
 
 
@@ -2881,16 +2947,23 @@ def _maybe_run_ups_cadence_checkpoint(
         CadenceConfig,
         POLICY_P1_EVERY_K_TURNS,
         POLICY_P2_CTX_THRESHOLD,
+        POLICY_P3_VELOCITY,
+        estimate_transcript_bytes,
         read_last_user_prompt,
         resolve_cadence_ctx_byte_window,
         resolve_cadence_ctx_threshold,
         resolve_cadence_enabled,
         resolve_cadence_k,
+        resolve_cadence_p3_velocity_threshold,
         resolve_cadence_policy,
         should_fire,
         should_fire_p2,
+        should_fire_p3_velocity,
+    )
+    from aelfrice.session_ring import (  # noqa: PLC0415
+        read_ring_state,
+        update_p3_velocity_state,
     )
-    from aelfrice.session_ring import read_ring_state  # noqa: PLC0415
 
     cwd_obj = payload.get(_CWD_KEY)
     cwd = (
@@ -2954,6 +3027,67 @@ def _maybe_run_ups_cadence_checkpoint(
         )
         return body
 
+    if policy == POLICY_P3_VELOCITY:
+        threshold = resolve_cadence_p3_velocity_threshold(start=cwd)
+        cfg = CadenceConfig(
+            enabled=True, policy=policy, p3_velocity_threshold=threshold,
+        )
+        state = read_ring_state(session_id)
+        if not isinstance(state, dict):
+            return None
+        raw_next: Any = state.get("next_fire_idx")
+        raw_bytes_last: Any = state.get("bytes_at_last_fire", 0)
+        raw_fire_last: Any = state.get("fire_idx_at_last_fire", 0)
+        if (
+            isinstance(raw_next, bool) or not isinstance(raw_next, int)
+            or isinstance(raw_bytes_last, bool) or not isinstance(raw_bytes_last, int)
+            or isinstance(raw_fire_last, bool) or not isinstance(raw_fire_last, int)
+        ):
+            return None
+        next_fire_idx = raw_next
+        bytes_at_last_fire = raw_bytes_last
+        fire_idx_at_last_fire = raw_fire_last
+        turns_since_last_fire = next_fire_idx - fire_idx_at_last_fire
+        if turns_since_last_fire <= 0:
+            return None
+        tp_obj = payload.get(_TRANSCRIPT_PATH_KEY)
+        tp: Path | None
+        if isinstance(tp_obj, str) and tp_obj:
+            tp = Path(tp_obj)
+        elif isinstance(tp_obj, os.PathLike):
+            tp = Path(tp_obj)
+        else:
+            tp = None
+        transcript_bytes = estimate_transcript_bytes(tp)
+        if not should_fire_p3_velocity(
+            bytes_at_last_fire=bytes_at_last_fire,
+            transcript_bytes=transcript_bytes,
+            turns_since_last_fire=turns_since_last_fire,
+            config=cfg,
+        ):
+            return None
+        body = _run_cadence_rebuild(payload, cwd)
+        if body is None:
+            return None
+        # Update both p3-velocity state slots atomically — mirrors Stop-side.
+        # When Stop and UPS both fire on the same boundary (the post-#874
+        # counter-sharing pattern), the second writer just overwrites with
+        # identical values, so the race is benign.
+        update_p3_velocity_state(
+            session_id,
+            transcript_bytes=transcript_bytes,
+            fire_idx=next_fire_idx,
+            stderr=serr,
+        )
+        density = (transcript_bytes - bytes_at_last_fire) / turns_since_last_fire
+        print(
+            f"aelfrice: ups cadence checkpoint fired @ fire_idx={next_fire_idx} "
+            f"(policy={policy}, velocity={density:.1f} bytes/turn, "
+            f"threshold={threshold})",
+            file=serr,
+        )
+        return body
+
     # Unknown policy / POLICY_OFF — no-op.
     return None
 

src/aelfrice/session_ring.py

@@ -171,6 +171,7 @@ def _normalize_for_session(
             "next_fire_idx": 0,
             "evicted_total": 0,
             "bytes_at_last_fire": 0,
+            "fire_idx_at_last_fire": 0,
             "classifications": [],
         }
     ring = data.get("ring")
@@ -182,14 +183,23 @@ def _normalize_for_session(
     evicted = data.get("evicted_total")
     if not isinstance(evicted, int) or evicted < 0:
         evicted = 0
-    # P3-velocity state — non-negative int; default 0 for pre-#876 rings
+    # P3-velocity state — non-negative ints; default 0 for pre-#876 rings.
+    # fire_idx_at_last_fire pairs with bytes_at_last_fire so the dispatcher
+    # can compute turns_since_last_fire = next_fire_idx - fire_idx_at_last_fire.
     bytes_at_last_fire = data.get("bytes_at_last_fire")
     if (
         not isinstance(bytes_at_last_fire, int)
         or isinstance(bytes_at_last_fire, bool)
         or bytes_at_last_fire < 0
     ):
         bytes_at_last_fire = 0
+    fire_idx_at_last_fire = data.get("fire_idx_at_last_fire")
+    if (
+        not isinstance(fire_idx_at_last_fire, int)
+        or isinstance(fire_idx_at_last_fire, bool)
+        or fire_idx_at_last_fire < 0
+    ):
+        fire_idx_at_last_fire = 0
     # P3-substantive state — list of bools; default [] for pre-#876 rings
     classifications_raw = data.get("classifications")
     if isinstance(classifications_raw, list):
@@ -203,6 +213,7 @@ def _normalize_for_session(
         "next_fire_idx": next_idx,
         "evicted_total": evicted,
         "bytes_at_last_fire": bytes_at_last_fire,
+        "fire_idx_at_last_fire": fire_idx_at_last_fire,
         "classifications": classifications,
     }
 
@@ -388,20 +399,74 @@ def update_bytes_at_last_fire(
 ) -> bool:
     """Update the p3_velocity ``bytes_at_last_fire`` slot (#876).
 
+    Kept for backward-compat with the PR 2 surface; new callers should
+    prefer :func:`update_p3_velocity_state` which updates both
+    ``bytes_at_last_fire`` and ``fire_idx_at_last_fire`` atomically so
+    the velocity predicate's two state inputs stay in sync.
+
+    Takes the same advisory lock as :func:`append_ids`. Fail-soft.
+    """
+    return _update_p3_velocity_fields(
+        session_id, bytes_at_last_fire=transcript_bytes, fire_idx_at_last_fire=None,
+        stderr=stderr,
+    )
+
+
+def update_p3_velocity_state(
+    session_id: str | None,
+    *,
+    transcript_bytes: int,
+    fire_idx: int,
+    stderr: IO[str] | None = None,
+) -> bool:
+    """Update both p3_velocity state slots atomically (#876).
+
     Called by the cadence dispatcher after a p3_velocity fire — the
-    predicate's next evaluation reads this value to compute density
-    ``(transcript_bytes - bytes_at_last_fire) / turns_since_last_fire``.
+    next predicate evaluation reads both values to compute:
 
-    Takes the same advisory lock as :func:`append_ids` so concurrent
-    UPS + Stop hooks can't clobber each other's writes. Fail-soft:
-    returns False on any error (file missing, lock failure, malformed
-    JSON, etc.); the caller continues without raising.
+      density = (current_transcript_bytes - bytes_at_last_fire)
+              / (current_next_fire_idx - fire_idx_at_last_fire)
 
-    Returns True on successful write, False on no-op or error.
+    Keeping these two in lock-step under a single ring write avoids the
+    sub-window where one is updated and the other isn't (which would
+    distort the density calculation on the next fire).
+
+    Takes the same advisory lock as :func:`append_ids`. Fail-soft:
+    returns False on any error.
+    """
+    if not isinstance(transcript_bytes, int) or transcript_bytes < 0:
+        return False
+    if not isinstance(fire_idx, int) or fire_idx < 0:
+        return False
+    return _update_p3_velocity_fields(
+        session_id,
+        bytes_at_last_fire=transcript_bytes,
+        fire_idx_at_last_fire=fire_idx,
+        stderr=stderr,
+    )
+
+
+def _update_p3_velocity_fields(
+    session_id: str | None,
+    *,
+    bytes_at_last_fire: int | None,
+    fire_idx_at_last_fire: int | None,
+    stderr: IO[str] | None,
+) -> bool:
+    """Shared inner for ``update_bytes_at_last_fire`` + ``update_p3_velocity_state``.
+
+    Either field can be None to mean 'leave unchanged'. Validation of
+    non-None values happens at the public callers.
     """
     if not session_id:
         return False
-    if not isinstance(transcript_bytes, int) or transcript_bytes < 0:
+    if bytes_at_last_fire is not None and (
+        not isinstance(bytes_at_last_fire, int) or bytes_at_last_fire < 0
+    ):
+        return False
+    if fire_idx_at_last_fire is not None and (
+        not isinstance(fire_idx_at_last_fire, int) or fire_idx_at_last_fire < 0
+    ):
         return False
     ring_path = _session_ring_path()
     if ring_path is None:
@@ -422,13 +487,16 @@ def update_bytes_at_last_fire(
         try:
             data = _read_ring_unlocked(ring_path)
             data = _normalize_for_session(data, session_id, ring_max)
-            data["bytes_at_last_fire"] = transcript_bytes
+            if bytes_at_last_fire is not None:
+                data["bytes_at_last_fire"] = bytes_at_last_fire
+            if fire_idx_at_last_fire is not None:
+                data["fire_idx_at_last_fire"] = fire_idx_at_last_fire
             _atomic_write(ring_path, json.dumps(data))
             return True
         except Exception as exc:
             _warn(
                 stderr,
-                f"session_ring: update_bytes_at_last_fire failed (non-fatal): {exc}",
+                f"session_ring: update_p3_velocity_state failed (non-fatal): {exc}",
             )
             return False
         finally: