feat(cedar-hitl): approval milestone writers + engine counters

Adds the 14 agent-side approval milestone writers (§11.1) on
``_ProgressWriter`` so Chunk 3's hook integration has a typed API
instead of stringly-typed ``write_agent_milestone`` calls, and the
per-task gate counter / per-container sliding-window rate limit /
denial-injection queue on ``PolicyEngine`` that §6.5 requires.

Why now: the hook work lands cleanly only after these surfaces exist
— every code path in ``pre_tool_use_hook``'s REQUIRE_APPROVAL branch
calls one of these helpers. Shipping them separately lets the hook
commit be about the state machine, not the event-shape bookkeeping.

Engine additions:
  - ``approval_gate_count`` / ``increment_approval_gate_count``: the
    per-task counter §12.9 bounds at ``approvalGateCap``. Session-scoped
    in v1; persistence tracked in §17.
  - ``approvals_in_last_minute`` / ``record_approval_gate_timestamp``:
    sliding-window rate limit (20/min/container, §12.9). Prune on read
    so callers see the current count without a separate tick.
  - ``queue_denial_injection`` / ``drain_denial_injections``: queue
    consumed by ``_denial_between_turns_hook`` at the next Stop seam
    (§6.5). Reason is pre-sanitized upstream by ``DenyTaskFn``.
  - ``mark_ceiling_shrinking_emitted``: emit-once latch for IMPL-26.
  - ``APPROVAL_RATE_LIMIT`` / ``APPROVAL_RATE_WINDOW_S`` module consts
    the hook imports rather than re-deriving.

Milestone writers (§11.1 table, 14 agent-emitted of 15):
  - ``pre_approvals_loaded``, ``approval_requested``,
    ``approval_granted``, ``approval_denied``, ``approval_timed_out``,
    ``approval_stranded``, ``approval_write_failed``,
    ``approval_resume_failed``, ``approval_poll_degraded``,
    ``approval_timeout_capped``, ``approval_ceiling_shrinking``,
    ``approval_cap_exceeded``, ``approval_rate_limit_exceeded``,
    ``approval_late_win``.
  - ``approval_decision_recorded`` (Lambda audit) and
    ``approval_timeout_capped_at_submit`` (CreateTaskFn) stay on the
    Lambda side — Chunk 5 owns those.

Each helper is a thin wrapper over ``_put_event("agent_milestone",
...)`` so the shared circuit-breaker + classifier path (finding aws-samples#6/aws-samples#8)
continues to apply. Metadata keys mirror the §11.1 shapes verbatim
(``maxLifetime_remaining_s`` preserves the design-doc spelling for
downstream parsers).

Tests: +29 total. 17 on ``TestApprovalMilestoneHelpers`` pin the DDB
payload shape for each helper (including the two
``approval_timeout_capped`` reason variants — rule_annotation carries
matching_rule_ids; maxLifetime_ceiling omits the field). 12 on the
engine: counter monotonicity, rate-window prune semantics at window
boundary, denial-queue FIFO + drain-clears, ceiling-shrinking latch
idempotency.

No caller changes — engine and writer surfaces are additive. Hook
integration lands in commit C.

Author: scoropeza

agent/src/policy.py

@@ -58,7 +58,7 @@
 import hashlib
 import json
 import time
-from collections import OrderedDict
+from collections import OrderedDict, deque
 from dataclasses import dataclass
 from enum import StrEnum
 from fnmatch import fnmatch
@@ -83,6 +83,8 @@
 CACHE_MAX_ENTRIES: int = 50  # §12.9: decoupled from approvalGateCap
 CACHE_TTL_S: float = 60.0  # §12.8 sliding-window TTL on DENIED/TIMED_OUT
 POLICIES_MAX_BYTES: int = 64 * 1024  # finding #12: reject blueprints > 64 KB
+APPROVAL_RATE_LIMIT: int = 20  # §12.9 per-container per-minute approval writes
+APPROVAL_RATE_WINDOW_S: float = 60.0  # sliding window paired with APPROVAL_RATE_LIMIT
 
 _SEVERITY_ORDER = {"low": 0, "medium": 1, "high": 2}
 _DEFAULT_SEVERITY = "medium"
@@ -627,6 +629,20 @@ def __init__(
             )
         self._approval_gate_cap = approval_gate_cap
 
+        # §12.9 per-task gate counter + per-container sliding-window rate limit.
+        # Both are session-scoped: the counter survives within the same task
+        # but NOT across container restarts (the design carries-forward tracks
+        # DDB persistence for the counter in Chunk 4/5; for now the §13.6
+        # reconciler + approval_gate_cap bound the worst case).
+        self._approval_gate_count: int = 0
+        self._approvals_last_minute: deque[float] = deque()
+        # §6.5 queue consumed by ``_denial_between_turns_hook``. Each entry
+        # is ``{"request_id", "reason", "decided_at"}``; reason is already
+        # sanitized by DenyTaskFn (§12.6).
+        self._denial_injection_queue: list[dict] = []
+        # IMPL-26: ``approval_ceiling_shrinking`` is emit-once per task.
+        self._emitted_ceiling_shrinking: bool = False
+
         # Validate task_type (non-fatal WARN to match Phase 1 behavior).
         from models import TaskType
 
@@ -800,6 +816,67 @@ def allowlist(self) -> ApprovalAllowlist:
     def recent_decisions(self) -> RecentDecisionCache:
         return self._cache
 
+    # ---- Approval-gate counters + denial queue (§6.5, §12.9) --------------
+
+    @property
+    def approval_gate_count(self) -> int:
+        """Session-scoped count of REQUIRE_APPROVAL gates emitted this task."""
+        return self._approval_gate_count
+
+    def increment_approval_gate_count(self) -> None:
+        """Bump the per-task gate counter (called at row-write time)."""
+        self._approval_gate_count += 1
+
+    def record_approval_gate_timestamp(self, now: float | None = None) -> None:
+        """Record a new approval-gate timestamp for the sliding rate-limit window."""
+        ts = time.monotonic() if now is None else now
+        self._approvals_last_minute.append(ts)
+        self._prune_rate_window(ts)
+
+    def _prune_rate_window(self, now: float) -> None:
+        """Drop timestamps older than ``APPROVAL_RATE_WINDOW_S``."""
+        cutoff = now - APPROVAL_RATE_WINDOW_S
+        while self._approvals_last_minute and self._approvals_last_minute[0] < cutoff:
+            self._approvals_last_minute.popleft()
+
+    @property
+    def approvals_in_last_minute(self) -> int:
+        """Count of approval-gate writes in the last ``APPROVAL_RATE_WINDOW_S``.
+
+        Prunes the window before returning so callers see the current count.
+        """
+        self._prune_rate_window(time.monotonic())
+        return len(self._approvals_last_minute)
+
+    def queue_denial_injection(
+        self, *, request_id: str, reason: str, decided_at: str | None
+    ) -> None:
+        """Append a denial-injection payload for ``_denial_between_turns_hook``.
+
+        Reason is expected to be pre-sanitized upstream (by ``DenyTaskFn``,
+        §12.6). The hook is responsible for XML-escaping at injection time.
+        """
+        self._denial_injection_queue.append(
+            {"request_id": request_id, "reason": reason, "decided_at": decided_at}
+        )
+
+    def drain_denial_injections(self) -> list[dict]:
+        """Pop and return the queued denial-injection payloads."""
+        out = list(self._denial_injection_queue)
+        self._denial_injection_queue.clear()
+        return out
+
+    def mark_ceiling_shrinking_emitted(self) -> bool:
+        """Idempotency latch for ``approval_ceiling_shrinking`` (IMPL-26).
+
+        Returns ``True`` the first time it is called (caller should emit the
+        milestone) and ``False`` on every subsequent call.
+        """
+        if self._emitted_ceiling_shrinking:
+            return False
+        self._emitted_ceiling_shrinking = True
+        return True
+
     # ---- Probes + low-level evaluation ------------------------------------
 
     def _probe_cedar(self, policies_text: str, tier_name: str) -> None:

agent/src/progress_writer.py

@@ -608,3 +608,205 @@ def write_agent_error(self, error_type: str, message: str) -> None:
                 "message_preview": self._preview(message),
             },
         )
+
+    # -- approval-gate milestones (§11.1, Chunk 3) -----------------------------
+    #
+    # Each helper is a thin wrapper over ``_put_event("agent_milestone", ...)``
+    # carrying the structured metadata shape documented in §11.1. Centralizing
+    # the milestone name + metadata keys here keeps the agent/Lambda/CLI
+    # contract consistent — every approval_* producer goes through a named
+    # method rather than stringly-typed ``write_agent_milestone`` calls.
+    #
+    # ``approval_decision_recorded`` is NOT emitted here: it is the
+    # Lambda-side audit event (written by ApproveTaskFn / DenyTaskFn directly
+    # to TaskEventsTable in Chunk 5). See IMPL-6.
+    #
+    # ``approval_timeout_capped_at_submit`` is emitted by CreateTaskFn and
+    # also lives on the Lambda side (Chunk 5). Agent-side helpers cover the
+    # 14 events the agent runtime emits.
+
+    def _put_approval_milestone(self, milestone: str, metadata: dict) -> None:
+        """Emit an ``agent_milestone`` with ``milestone`` + extra metadata.
+
+        Mirrors the shape of ``write_agent_milestone`` but preserves the
+        structured metadata keys so consumers do not need to parse a
+        free-form ``details`` string.
+        """
+        payload: dict = {"milestone": milestone}
+        payload.update(metadata)
+        self._put_event("agent_milestone", payload)
+
+    def write_approval_pre_approvals_loaded(self, count: int, scopes: list[str]) -> None:
+        """Emit ``pre_approvals_loaded`` (agent init, §11.1)."""
+        self._put_approval_milestone(
+            "pre_approvals_loaded",
+            {"count": count, "scopes": list(scopes)},
+        )
+
+    def write_approval_requested(
+        self,
+        *,
+        request_id: str,
+        tool_name: str,
+        input_preview: str,
+        reason: str,
+        severity: str,
+        timeout_s: int,
+        matching_rule_ids: list[str],
+    ) -> None:
+        """Emit ``approval_requested`` (PENDING row written, §11.1)."""
+        self._put_approval_milestone(
+            "approval_requested",
+            {
+                "request_id": request_id,
+                "tool_name": tool_name,
+                "input_preview": self._preview(input_preview),
+                "reason": self._preview(reason),
+                "severity": severity,
+                "timeout_s": timeout_s,
+                "matching_rule_ids": list(matching_rule_ids),
+            },
+        )
+
+    def write_approval_granted(
+        self,
+        *,
+        request_id: str,
+        scope: str,
+        decided_at: str | None,
+    ) -> None:
+        """Emit ``approval_granted`` (user APPROVED, §11.1)."""
+        self._put_approval_milestone(
+            "approval_granted",
+            {"request_id": request_id, "scope": scope, "decided_at": decided_at},
+        )
+
+    def write_approval_denied(
+        self,
+        *,
+        request_id: str,
+        reason: str,
+        decided_at: str | None,
+    ) -> None:
+        """Emit ``approval_denied`` (user DENIED, §11.1)."""
+        self._put_approval_milestone(
+            "approval_denied",
+            {
+                "request_id": request_id,
+                "reason": self._preview(reason),
+                "decided_at": decided_at,
+            },
+        )
+
+    def write_approval_timed_out(self, *, request_id: str, timeout_s: int) -> None:
+        """Emit ``approval_timed_out`` (timer expired, §11.1)."""
+        self._put_approval_milestone(
+            "approval_timed_out",
+            {"request_id": request_id, "timeout_s": timeout_s},
+        )
+
+    def write_approval_stranded(
+        self,
+        *,
+        request_id: str,
+        age_s: int,
+        reason: str,
+    ) -> None:
+        """Emit ``approval_stranded`` (reconciler surface, §11.1)."""
+        self._put_approval_milestone(
+            "approval_stranded",
+            {"request_id": request_id, "age_s": age_s, "reason": self._preview(reason)},
+        )
+
+    def write_approval_write_failed(self, *, request_id: str | None, error: str) -> None:
+        """Emit ``approval_write_failed`` (TransactWriteItems failure, §11.1)."""
+        self._put_approval_milestone(
+            "approval_write_failed",
+            {"request_id": request_id, "error": self._preview(error)},
+        )
+
+    def write_approval_resume_failed(self, *, request_id: str, error: str) -> None:
+        """Emit ``approval_resume_failed`` (resume transaction failure, §11.1)."""
+        self._put_approval_milestone(
+            "approval_resume_failed",
+            {"request_id": request_id, "error": self._preview(error)},
+        )
+
+    def write_approval_poll_degraded(self, *, request_id: str, consecutive_failures: int) -> None:
+        """Emit ``approval_poll_degraded`` (3+ consecutive GetItem failures)."""
+        self._put_approval_milestone(
+            "approval_poll_degraded",
+            {
+                "request_id": request_id,
+                "consecutive_failures": consecutive_failures,
+            },
+        )
+
+    def write_approval_timeout_capped(
+        self,
+        *,
+        request_id: str,
+        requested_timeout_s: int,
+        effective_timeout_s: int,
+        reason: str,
+        matching_rule_ids: list[str] | None = None,
+    ) -> None:
+        """Emit ``approval_timeout_capped`` when min-wins clips the timeout.
+
+        ``reason`` ∈ {rule_annotation, maxLifetime_ceiling, runtime_jwt_ceiling}.
+        ``matching_rule_ids`` populated when ``reason == "rule_annotation"``
+        (IMPL-26).
+        """
+        payload: dict = {
+            "request_id": request_id,
+            "requested_timeout_s": requested_timeout_s,
+            "effective_timeout_s": effective_timeout_s,
+            "reason": reason,
+        }
+        if matching_rule_ids is not None:
+            payload["matching_rule_ids"] = list(matching_rule_ids)
+        self._put_approval_milestone("approval_timeout_capped", payload)
+
+    def write_approval_ceiling_shrinking(
+        self,
+        *,
+        request_id: str,
+        max_lifetime_remaining_s: int,
+        cleanup_margin_s: int,
+        task_default_timeout_s: int,
+    ) -> None:
+        """Emit once-per-task ``approval_ceiling_shrinking`` (IMPL-26, §11.1)."""
+        self._put_approval_milestone(
+            "approval_ceiling_shrinking",
+            {
+                "request_id": request_id,
+                "maxLifetime_remaining_s": max_lifetime_remaining_s,
+                "cleanup_margin_s": cleanup_margin_s,
+                "task_default_timeout_s": task_default_timeout_s,
+            },
+        )
+
+    def write_approval_cap_exceeded(self, *, request_id: str, count: int, cap: int) -> None:
+        """Emit ``approval_cap_exceeded`` (per-task gate-cap fired)."""
+        self._put_approval_milestone(
+            "approval_cap_exceeded",
+            {"request_id": request_id, "count": count, "cap": cap},
+        )
+
+    def write_approval_rate_limit_exceeded(self, *, request_id: str, rate: int, limit: int) -> None:
+        """Emit ``approval_rate_limit_exceeded`` (per-minute rate limit hit)."""
+        self._put_approval_milestone(
+            "approval_rate_limit_exceeded",
+            {"request_id": request_id, "rate": rate, "limit": limit},
+        )
+
+    def write_approval_late_win(self, *, request_id: str, outcome: str, reason: str) -> None:
+        """Emit ``approval_late_win`` (VM-throttle race mitigation, IMPL-24)."""
+        self._put_approval_milestone(
+            "approval_late_win",
+            {
+                "request_id": request_id,
+                "outcome": outcome,
+                "reason": self._preview(reason),
+            },
+        )