diff --git a/kora_cli/audit/jsonl_sink.py b/kora_cli/audit/jsonl_sink.py
index ff1b253e00c3..497328bbe091 100644
--- a/kora_cli/audit/jsonl_sink.py
+++ b/kora_cli/audit/jsonl_sink.py
@@ -62,12 +62,14 @@
 
 from __future__ import annotations
 
+import atexit
 import json
 import logging
 import os
+import threading
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Any, Dict, Literal, Optional
+from typing import Any, Dict, List, Literal, Optional, Tuple
 
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -203,6 +205,39 @@
     # auto_applied). Source is ``reasoning`` since the cluster
     # input is reasoning audit.
     "promotion.snapshot_field_added",
+    # KR-PROMOTE-ROUTER-TUNING — third promotion loop. Reads per-route
+    # escalation counts from cost_telemetry + the rolling 24h /
+    # monthly windows to surface routes whose Haiku-to-Opus
+    # escalation rate suggests their trigger pattern could be tuned
+    # (tightened to save Opus spend, OR loosened to avoid recurring
+    # operator /opus overrides). Payload carries proposal_id /
+    # route / escalation_count / total_calls / escalation_rate /
+    # recommendation_kind ("tighten_review" | "loosen_review") /
+    # rationale / created_at / status. Source is ``reasoning``.
+    "promotion.router_trigger_proposed",
+    # KR-PROMOTE-TOOL-TRIMMING — fourth promotion loop. Reads
+    # ``reasoning.tool_called`` audit history per (route, tool_name)
+    # over the observation window and proposes adding unused tools
+    # to a route's drop-list (the pre_tool_list_finalized hook
+    # consumer). Payload carries proposal_id / route /
+    # unused_tools (list of names) / total_calls_for_route /
+    # observation_window_days / created_at / status. Source is
+    # ``reasoning``. v1 is propose-only; enforcement of the drop-
+    # list lands in the future KR-PLUGIN-TOOL-DESC-TRIM bucket.
+    "promotion.tool_trim_proposed",
+    # KR-PROMOTE-PROBE-FIX-ENVELOPES — fifth promotion loop. Reads
+    # ``tool.probe_autofix_attempted`` + ``probe.investigation_completed``
+    # audits + clusters recurring probe failures whose investigation
+    # summaries point at a consistent recommended fix. Proposes
+    # adding a new envelope action to ``probes/fix_envelopes.py``.
+    # HIGH-RISK: payload includes the cluster's recurring fix-text +
+    # operator-facing blast radius description; auto-apply is
+    # HARDCODED FALSE — operator MUST review and scaffold manually.
+    # Payload: proposal_id / probe / fix_name_suggestion /
+    # cluster_size / sample_investigation_ids /
+    # recurring_recommendation_text / blast_radius_summary /
+    # created_at / status. Source is ``reasoning``.
+    "promotion.probe_envelope_action_proposed",
 ]
 
 SourceName = Literal[
@@ -326,14 +361,262 @@ def emit_audit(
         return
 
     path = log_path or _resolve_log_path()
+    # KR-CHEAP-AUDIT-BATCHING (R3-4 #9) — route through the batched
+    # sink when batching is enabled (default). The per-emit write
+    # path stays available as the fallback (BATCH_SIZE_ENV=0) and
+    # the immediate-write path inside the sink itself for tests
+    # that pass log_path explicitly + want sync semantics.
+    if _is_batching_enabled():
+        _enqueue_for_batched_flush(entry, path)
+        return
+    _write_entries_sync([entry], path)
+
+
+# ---------------------------------------------------------------------------
+# Batched flusher — KR-CHEAP-AUDIT-BATCHING (R3-4 #9)
+# ---------------------------------------------------------------------------
+#
+# Original behavior: every ``emit_audit`` call opens the JSONL file,
+# appends one line, closes. Fine for low volume but inefficient when
+# the daemon is humming (every reasoning tool call, every probe
+# wake, every promotion proposal emits a row). R3-4 #9 batches:
+# flush at ``KORA_AUDIT_BATCH_SIZE`` events OR ``KORA_AUDIT_FLUSH_
+# INTERVAL_SECONDS`` seconds, whichever first.
+#
+# Lifecycle (STOP-ASK §4 mitigation):
+#   * Daemon process: background thread (daemon=True) ticks every
+#     FLUSH_INTERVAL and drains the queue.
+#   * CLI invocations: same thread starts lazily on the first
+#     emit; atexit handler drains on shutdown so single-shot CLI
+#     processes don't lose pending events.
+#   * Tests that pass an explicit log_path or set BATCH_SIZE=0 stay
+#     on the sync write path.
+#
+# Per-emit interface is UNCHANGED — callers still call ``emit_audit``
+# synchronously; the queue is purely internal.
+
+BATCH_SIZE_ENV = "KORA_AUDIT_BATCH_SIZE"
+FLUSH_INTERVAL_ENV = "KORA_AUDIT_FLUSH_INTERVAL_SECONDS"
+
+DEFAULT_BATCH_SIZE = 100
+DEFAULT_FLUSH_INTERVAL_SECONDS = 5.0
+
+
+_batch_lock = threading.RLock()
+# (entry, log_path) tuples. The path is captured at enqueue-time so
+# callers that pass a per-call log_path override still write to the
+# right file even when batched.
+_batch_queue: List[Tuple["AuditEntry", Path]] = []
+_flusher_thread: Optional[threading.Thread] = None
+_flusher_stop = threading.Event()
+_atexit_registered = False
+
+
+def _read_batch_size() -> int:
+    raw = os.environ.get(BATCH_SIZE_ENV, "").strip()
+    if not raw:
+        return DEFAULT_BATCH_SIZE
+    try:
+        value = int(raw)
+    except ValueError:
+        logger.warning(
+            "[kora.audit.batching] %s=%r not int; using default %d",
+            BATCH_SIZE_ENV,
+            raw,
+            DEFAULT_BATCH_SIZE,
+        )
+        return DEFAULT_BATCH_SIZE
+    if value < 0:
+        return DEFAULT_BATCH_SIZE
+    return value
+
+
+def _read_flush_interval_seconds() -> float:
+    raw = os.environ.get(FLUSH_INTERVAL_ENV, "").strip()
+    if not raw:
+        return DEFAULT_FLUSH_INTERVAL_SECONDS
+    try:
+        value = float(raw)
+    except ValueError:
+        logger.warning(
+            "[kora.audit.batching] %s=%r not numeric; using default %ss",
+            FLUSH_INTERVAL_ENV,
+            raw,
+            DEFAULT_FLUSH_INTERVAL_SECONDS,
+        )
+        return DEFAULT_FLUSH_INTERVAL_SECONDS
+    if value <= 0:
+        return DEFAULT_FLUSH_INTERVAL_SECONDS
+    return value
+
+
+def _is_batching_enabled() -> bool:
+    """Batching is on whenever BATCH_SIZE > 0 (default 100).
+    Setting BATCH_SIZE=0 forces the legacy per-emit write path —
+    useful for tests that want sync semantics."""
+    return _read_batch_size() > 0
+
+
+def _write_entries_sync(
+    entries: List["AuditEntry"], path: Path
+) -> None:
+    """Write a list of entries to ``path`` in one open/close cycle.
+    Fail-soft per OSError; never raises. Empty list is a no-op."""
+    if not entries:
+        return
     try:
         path.parent.mkdir(parents=True, exist_ok=True)
         with path.open("a", encoding="utf-8") as f:
-            f.write(entry.model_dump_json() + "\n")
+            for entry in entries:
+                f.write(entry.model_dump_json() + "\n")
     except OSError as exc:
         logger.warning(
-            "[kora.audit.skipped] JSONL write failed (%s): %r — "
+            "[kora.audit.skipped] JSONL batched write failed (%s): %r — "
             "caller's structured-log line still emitted",
             path,
             exc,
         )
+
+
+def _drain_queue_locked() -> List[Tuple["AuditEntry", Path]]:
+    """Move every queued (entry, path) out of the queue. Caller
+    holds ``_batch_lock``. Returns the drained items so the actual
+    file writes can happen OUTSIDE the lock (writing to disk while
+    holding the lock would block subsequent emits unnecessarily)."""
+    drained = list(_batch_queue)
+    _batch_queue.clear()
+    return drained
+
+
+def _flush_now() -> int:
+    """Drain the queue + write each path's entries in one open/close.
+    Returns total events flushed (0 when queue empty). Safe to call
+    from any thread (the size-triggered flush calls from emit_audit
+    + the interval-triggered flush from the background thread + the
+    atexit handler all share this path)."""
+    with _batch_lock:
+        drained = _drain_queue_locked()
+    if not drained:
+        return 0
+    # Group by path so each file is opened once per flush.
+    by_path: Dict[Path, List["AuditEntry"]] = {}
+    for entry, path in drained:
+        by_path.setdefault(path, []).append(entry)
+    for path, entries in by_path.items():
+        _write_entries_sync(entries, path)
+    return len(drained)
+
+
+def _flusher_loop() -> None:
+    """Background-thread entry. Sleeps the configured interval +
+    flushes; exits when ``_flusher_stop`` is set."""
+    interval = _read_flush_interval_seconds()
+    while not _flusher_stop.is_set():
+        # ``wait(interval)`` returns True if the stop event was
+        # set during the wait → exit promptly. Otherwise it
+        # returns False after the interval and we flush.
+        if _flusher_stop.wait(interval):
+            break
+        try:
+            _flush_now()
+        except Exception as exc:
+            logger.warning(
+                "[kora.audit.batching] background flush raised %r — "
+                "queue retried on next tick",
+                exc,
+            )
+    # Final drain on stop so atexit-initiated shutdown captures
+    # whatever the background thread had pending at the moment
+    # ``_flusher_stop`` was set.
+    try:
+        _flush_now()
+    except Exception as exc:
+        logger.warning(
+            "[kora.audit.batching] final drain raised %r", exc
+        )
+
+
+def _atexit_flush() -> None:
+    """atexit hook — drain the queue + signal the background thread
+    to exit. Best-effort; never raises (atexit handlers that raise
+    are surfaced by the runtime as ugly tracebacks)."""
+    try:
+        _flusher_stop.set()
+        _flush_now()
+    except Exception as exc:
+        logger.warning(
+            "[kora.audit.batching] atexit flush raised %r — events "
+            "may be lost",
+            exc,
+        )
+
+
+def _ensure_flusher_started() -> None:
+    """Lazy thread start on first batched emit. Idempotent — safe
+    to call from every emit_audit. The thread is ``daemon=True`` so
+    a process exit doesn't block on it (the atexit handler does the
+    final drain regardless)."""
+    global _flusher_thread, _atexit_registered
+    with _batch_lock:
+        if _flusher_thread is not None and _flusher_thread.is_alive():
+            return
+        _flusher_stop.clear()
+        _flusher_thread = threading.Thread(
+            target=_flusher_loop,
+            name="kora-audit-flusher",
+            daemon=True,
+        )
+        _flusher_thread.start()
+        if not _atexit_registered:
+            atexit.register(_atexit_flush)
+            _atexit_registered = True
+
+
+def _enqueue_for_batched_flush(entry: "AuditEntry", path: Path) -> None:
+    """Append one (entry, path) to the queue + flush immediately if
+    the size threshold is hit. Otherwise the background thread
+    handles the time-based flush."""
+    _ensure_flusher_started()
+    size_threshold = _read_batch_size()
+    should_flush_immediately = False
+    with _batch_lock:
+        _batch_queue.append((entry, path))
+        if len(_batch_queue) >= size_threshold:
+            should_flush_immediately = True
+    if should_flush_immediately:
+        _flush_now()
+
+
+def flush_for_tests() -> int:
+    """Test surface: synchronously drain the queue. Returns number of
+    events flushed. Production code should NOT call this — the
+    automatic size + time + atexit triggers cover the production
+    flush points."""
+    return _flush_now()
+
+
+def _reset_batching_for_tests() -> None:
+    """Test surface: drain the queue (mirrors atexit shutdown
+    semantics), stop the flusher thread, and reset state so the
+    next test starts clean. Lets the dedicated batching-tests assert
+    "shutdown drains pending events" by calling this helper as the
+    shutdown stand-in.
+    """
+    global _flusher_thread, _atexit_registered
+    # Drain BEFORE stopping the thread so an in-flight queue
+    # reaches disk — atexit's contract is "events written".
+    try:
+        _flush_now()
+    except Exception:
+        # Best-effort; production atexit handler also swallows.
+        pass
+    _flusher_stop.set()
+    if _flusher_thread is not None:
+        _flusher_thread.join(timeout=2.0)
+    _flusher_thread = None
+    with _batch_lock:
+        _batch_queue.clear()
+    # We intentionally leave _atexit_registered True because Python's
+    # atexit API has no unregister-by-function for once-registered
+    # callbacks; the callback short-circuits on a clean queue so
+    # leaving it registered is harmless.
diff --git a/kora_cli/listeners/__init__.py b/kora_cli/listeners/__init__.py
index 3903c21a37d1..66f872b4d1b4 100644
--- a/kora_cli/listeners/__init__.py
+++ b/kora_cli/listeners/__init__.py
@@ -93,3 +93,11 @@
 # order. Default auto-apply OFF (proposes via audit only) per
 # bucket STOP-ASK §4 safety posture.
 from kora_cli.listeners import promote_snapshot_expand_listener  # noqa: F401
+# KR-PROMOTE-LOOPS-COMPLETION-MEGABUCKET — the three remaining
+# promotion loops. All propose-then-approve; all $0-LLM (pure audit
+# scans + threshold math). Order chosen to mirror the spec table
+# (router-tuning → tool-trimming → probe-fix-envelopes). probe-fix
+# is HARDCODED auto-apply FALSE; the others default-OFF auto-apply.
+from kora_cli.listeners import promote_router_tuning_listener  # noqa: F401
+from kora_cli.listeners import promote_tool_trimming_listener  # noqa: F401
+from kora_cli.listeners import promote_probe_fix_envelopes_listener  # noqa: F401
diff --git a/kora_cli/listeners/promote_probe_fix_envelopes_listener.py b/kora_cli/listeners/promote_probe_fix_envelopes_listener.py
new file mode 100644
index 000000000000..2b8affb45010
--- /dev/null
+++ b/kora_cli/listeners/promote_probe_fix_envelopes_listener.py
@@ -0,0 +1,46 @@
+"""Heartbeat-scheduled probe-fix-envelope promotion cycle — KR-PROMOTE-PROBE-FIX-ENVELOPES.
+
+Same listener shape as the other promotion-loop listeners. Cadence
+operator-tunable via ``KORA_PROMOTE_PROBE_FIX_INTERVAL_SEC``.
+
+Auto-apply HARDCODED FALSE — see
+:mod:`kora_cli.promote.probe_fix_envelopes` module docstring for
+the safety rationale.
+"""
+
+from __future__ import annotations
+
+import logging
+
+from kora_cli.listeners.heartbeat import register_periodic_task
+from kora_cli.promote.probe_fix_envelopes.plugin import (
+    get_interval_seconds,
+    run_probe_fix_envelopes_cycle,
+)
+
+logger = logging.getLogger(__name__)
+
+
+async def _periodic_task() -> None:
+    try:
+        summary = await run_probe_fix_envelopes_cycle()
+        logger.debug(
+            "[kora.promote.probe_fix_envelopes.listener] tick complete: "
+            "proposals_persisted=%d expired_count=%d duration_ms=%d",
+            summary.get("proposals_persisted", 0),
+            summary.get("expired_count", 0),
+            summary.get("duration_ms", 0),
+        )
+    except Exception as exc:
+        logger.warning(
+            "[kora.promote.probe_fix_envelopes.listener] tick raised "
+            "%r — next scheduled run will retry",
+            exc,
+        )
+
+
+register_periodic_task(
+    "promote_probe_fix_envelopes_cycle",
+    interval_seconds=float(get_interval_seconds()),
+    callable=_periodic_task,
+)
diff --git a/kora_cli/listeners/promote_router_tuning_listener.py b/kora_cli/listeners/promote_router_tuning_listener.py
new file mode 100644
index 000000000000..d04ed9cc8cda
--- /dev/null
+++ b/kora_cli/listeners/promote_router_tuning_listener.py
@@ -0,0 +1,60 @@
+"""Heartbeat-scheduled router-tuning promotion cycle — KR-PROMOTE-ROUTER-TUNING.
+
+Registers :func:`run_router_tuning_cycle` as a periodic task.
+Cadence operator-tunable via
+``KORA_PROMOTE_ROUTER_TUNING_INTERVAL_SEC`` (default 86400s = 24h).
+Master kill-switch ``KORA_PROMOTE_ROUTER_TUNING_ENABLED=false``
+checked inside the cycle so flipping the env at runtime takes
+effect on the next tick.
+
+# Why a periodic interval, not a cron string
+
+Same rationale as the phrasebook + snapshot-expand listeners —
+the heartbeat scheduler is interval-based; the bucket spec's
+``"0 8 * * *"`` cron suggestion is documented but not honored
+verbatim. Daily-interval is sufficient for a batch promotion
+loop.
+
+# Fail-soft
+
+Cycle exceptions swallowed by the heartbeat scheduler's
+``_loop``; per-proposal exceptions caught inside the cycle so one
+bad proposal doesn't poison the batch.
+"""
+
+from __future__ import annotations
+
+import logging
+
+from kora_cli.listeners.heartbeat import register_periodic_task
+from kora_cli.promote.router_tuning.plugin import (
+    get_interval_seconds,
+    run_router_tuning_cycle,
+)
+
+logger = logging.getLogger(__name__)
+
+
+async def _periodic_task() -> None:
+    try:
+        summary = await run_router_tuning_cycle()
+        logger.debug(
+            "[kora.promote.router_tuning.listener] tick complete: "
+            "proposals_persisted=%d expired_count=%d duration_ms=%d",
+            summary.get("proposals_persisted", 0),
+            summary.get("expired_count", 0),
+            summary.get("duration_ms", 0),
+        )
+    except Exception as exc:
+        logger.warning(
+            "[kora.promote.router_tuning.listener] tick raised %r — "
+            "next scheduled run will retry",
+            exc,
+        )
+
+
+register_periodic_task(
+    "promote_router_tuning_cycle",
+    interval_seconds=float(get_interval_seconds()),
+    callable=_periodic_task,
+)
diff --git a/kora_cli/listeners/promote_tool_trimming_listener.py b/kora_cli/listeners/promote_tool_trimming_listener.py
new file mode 100644
index 000000000000..fa92abdf7cf6
--- /dev/null
+++ b/kora_cli/listeners/promote_tool_trimming_listener.py
@@ -0,0 +1,42 @@
+"""Heartbeat-scheduled tool-trimming promotion cycle — KR-PROMOTE-TOOL-TRIMMING.
+
+Same listener shape as the other promotion-loop listeners. Cadence
+operator-tunable via ``KORA_PROMOTE_TOOL_TRIMMING_INTERVAL_SEC``.
+"""
+
+from __future__ import annotations
+
+import logging
+
+from kora_cli.listeners.heartbeat import register_periodic_task
+from kora_cli.promote.tool_trimming.plugin import (
+    get_interval_seconds,
+    run_tool_trimming_cycle,
+)
+
+logger = logging.getLogger(__name__)
+
+
+async def _periodic_task() -> None:
+    try:
+        summary = await run_tool_trimming_cycle()
+        logger.debug(
+            "[kora.promote.tool_trimming.listener] tick complete: "
+            "proposals_persisted=%d expired_count=%d duration_ms=%d",
+            summary.get("proposals_persisted", 0),
+            summary.get("expired_count", 0),
+            summary.get("duration_ms", 0),
+        )
+    except Exception as exc:
+        logger.warning(
+            "[kora.promote.tool_trimming.listener] tick raised %r — "
+            "next scheduled run will retry",
+            exc,
+        )
+
+
+register_periodic_task(
+    "promote_tool_trimming_cycle",
+    interval_seconds=float(get_interval_seconds()),
+    callable=_periodic_task,
+)
diff --git a/kora_cli/probes/wake_consumer.py b/kora_cli/probes/wake_consumer.py
index 120bfe5d0dcc..f8a13ce20523 100644
--- a/kora_cli/probes/wake_consumer.py
+++ b/kora_cli/probes/wake_consumer.py
@@ -70,6 +70,14 @@
 BYPASS_CRITICAL_ENV = "KORA_PROBE_DEBOUNCE_BYPASS_CRITICAL"
 JOSHUA_SLACK_USER_ID_ENV = "KORA_SLACK_JOSHUA_USER_ID"  # reused from PR #149
 
+# KR-PROBE-DEBOUNCE — consecutive-failure buffering (upgrade from
+# PR #166's flat-window debounce per CC#1's #163 follow-on tracker).
+# Default 2: a probe must fire wake_requested twice within the
+# debounce window before the consumer dispatches an investigation.
+# Operator-tunable; setting to 1 preserves pre-upgrade behavior.
+CONSECUTIVE_REQUIRED_ENV = "KORA_PROBE_DEBOUNCE_CONSECUTIVE_REQUIRED"
+DEFAULT_CONSECUTIVE_REQUIRED = 2
+
 
 def _read_debounce_seconds() -> int:
     raw = os.environ.get(DEBOUNCE_SECONDS_ENV, "").strip()
@@ -98,6 +106,37 @@ def _read_debounce_seconds() -> int:
     return value
 
 
+def _read_consecutive_required() -> int:
+    """Read ``KORA_PROBE_DEBOUNCE_CONSECUTIVE_REQUIRED`` with fail-soft
+    parsing. Defaults to :data:`DEFAULT_CONSECUTIVE_REQUIRED` on
+    malformed values. ``1`` disables the buffering and restores the
+    PR #166 flat-window behavior."""
+    raw = os.environ.get(CONSECUTIVE_REQUIRED_ENV, "").strip()
+    if not raw:
+        return DEFAULT_CONSECUTIVE_REQUIRED
+    try:
+        value = int(raw)
+    except ValueError:
+        logger.warning(
+            "[kora.probe_wake_consumer] %s=%r is not numeric; using "
+            "default %d",
+            CONSECUTIVE_REQUIRED_ENV,
+            raw,
+            DEFAULT_CONSECUTIVE_REQUIRED,
+        )
+        return DEFAULT_CONSECUTIVE_REQUIRED
+    if value < 1:
+        logger.warning(
+            "[kora.probe_wake_consumer] %s=%d must be ≥ 1; using "
+            "default %d",
+            CONSECUTIVE_REQUIRED_ENV,
+            value,
+            DEFAULT_CONSECUTIVE_REQUIRED,
+        )
+        return DEFAULT_CONSECUTIVE_REQUIRED
+    return value
+
+
 def _read_bypass_critical() -> bool:
     raw = os.environ.get(BYPASS_CRITICAL_ENV, "").strip().lower()
     return raw in {"true", "1", "yes", "on"}
@@ -119,6 +158,15 @@ class WakeConsumeOutcome:
     reasoning_invoked: bool  # False when engine None / debounced
     dm_sent: bool  # False when Slack client unavailable
     debounce_skipped: bool = False
+    # KR-PROBE-DEBOUNCE consecutive-failure upgrade. True when an
+    # event was held in the consecutive-failure buffer (didn't yet
+    # meet ``KORA_PROBE_DEBOUNCE_CONSECUTIVE_REQUIRED``) instead of
+    # being dispatched. Distinct from ``debounce_skipped`` (which
+    # remains the flat-window post-dispatch skip) so the audit
+    # / listener telemetry can tell single-tick-flake holds apart
+    # from "already-dispatched recently" skips.
+    buffered_skipped: bool = False
+    buffered_consecutive_count: int = 0
     error: Optional[str] = None
 
 
@@ -157,6 +205,14 @@ def __init__(
         self._debounce_lock = threading.RLock()
         # (probe_name, issue_category) → datetime of last dispatched
         self._last_dispatched: Dict[Tuple[str, str], datetime] = {}
+        # KR-PROBE-DEBOUNCE consecutive-failure buffer state.
+        # (probe, category) → (consecutive_count, first_seen_at).
+        # Reset when the sliding window elapses (proxy for "the
+        # probe went healthy then unhealthy again — the burst was
+        # transient, restart the counter").
+        self._consecutive_buffer: Dict[
+            Tuple[str, str], Tuple[int, datetime]
+        ] = {}
 
     @property
     def debounce_map_size(self) -> int:
@@ -164,11 +220,18 @@ def debounce_map_size(self) -> int:
         return len(self._last_dispatched)
 
     def reset_debounce_state(self) -> None:
-        """Clear the in-memory debounce map. Listener shutdown calls
-        this so subsequent listener start sees a clean slate.
-        Mirrors :meth:`AlertNotifier.reset_dedup_state` (PR #149)."""
+        """Clear the in-memory debounce map + consecutive-failure
+        buffer. Listener shutdown calls this so subsequent listener
+        start sees a clean slate. Mirrors :meth:`AlertNotifier.reset_dedup_state`
+        (PR #149)."""
         with self._debounce_lock:
             self._last_dispatched = {}
+            self._consecutive_buffer = {}
+
+    @property
+    def consecutive_buffer_size(self) -> int:
+        """Read-only view for tests + telemetry."""
+        return len(self._consecutive_buffer)
 
     # ------------------------------------------------------------------
     # Debounce
@@ -197,6 +260,55 @@ def _mark_dispatched(self, probe: str, category: str) -> None:
             self._last_dispatched[(probe, category)] = datetime.now(
                 timezone.utc
             )
+            # Clear the consecutive buffer once we've dispatched —
+            # the post-dispatch flat-window debounce takes over.
+            self._consecutive_buffer.pop((probe, category), None)
+
+    # ------------------------------------------------------------------
+    # Consecutive-failure buffer (KR-PROBE-DEBOUNCE upgrade)
+    # ------------------------------------------------------------------
+
+    def _record_failure_and_check_threshold(
+        self, probe: str, category: str, severity: str
+    ) -> Tuple[bool, int]:
+        """Update the consecutive-failure buffer for one (probe,
+        category) event.
+
+        Returns ``(threshold_met, count_after_update)``:
+          * ``threshold_met`` is True when this event brings the
+            buffer up to the configured required count → caller
+            should dispatch.
+          * ``count_after_update`` is the post-update buffered
+            count, surfaced in the outcome for tests / telemetry.
+
+        Window semantics: the buffer entry expires after the
+        existing :data:`DEBOUNCE_SECONDS_ENV` window elapsed since
+        ``first_seen_at`` — re-using the existing debounce window
+        keeps the operator-tunable surface minimal. A fresh event
+        AFTER expiry restarts the count at 1 (proxy for "probe
+        went healthy then unhealthy again").
+
+        Critical-severity bypass: when ``severity == "critical"``
+        AND ``KORA_PROBE_DEBOUNCE_BYPASS_CRITICAL`` is truthy, the
+        threshold is implicitly 1 (caller never reaches this
+        method; see :meth:`_should_dispatch_now`).
+        """
+        required = _read_consecutive_required()
+        window = _read_debounce_seconds()
+        now = datetime.now(timezone.utc)
+        with self._debounce_lock:
+            entry = self._consecutive_buffer.get((probe, category))
+            if entry is None:
+                self._consecutive_buffer[(probe, category)] = (1, now)
+                return (required <= 1, 1)
+            count, first_seen = entry
+            # Expired window → restart count.
+            if window > 0 and (now - first_seen).total_seconds() >= window:
+                self._consecutive_buffer[(probe, category)] = (1, now)
+                return (required <= 1, 1)
+            new_count = count + 1
+            self._consecutive_buffer[(probe, category)] = (new_count, first_seen)
+            return (new_count >= required, new_count)
 
     # ------------------------------------------------------------------
     # Public entry point
@@ -239,6 +351,41 @@ async def consume_wake_event(
                 debounce_skipped=True,
             )
 
+        # KR-PROBE-DEBOUNCE consecutive-failure buffering. Critical
+        # wakes can optionally bypass via BYPASS_CRITICAL_ENV (same
+        # opt-in env as the flat-window bypass — operator already
+        # tunes one knob for "trust critical urgency").
+        bypass_critical = (
+            severity == "critical" and _read_bypass_critical()
+        )
+        if not bypass_critical:
+            threshold_met, buffered_count = (
+                self._record_failure_and_check_threshold(
+                    probe, category, severity
+                )
+            )
+            if not threshold_met:
+                logger.debug(
+                    "[kora.probe_wake_consumer] buffered probe=%s "
+                    "category=%s severity=%s count=%d required=%d "
+                    "— waiting for consecutive failure",
+                    probe,
+                    category,
+                    severity,
+                    buffered_count,
+                    _read_consecutive_required(),
+                )
+                return WakeConsumeOutcome(
+                    probe=probe,
+                    category=category,
+                    severity=severity,
+                    dispatched=False,
+                    reasoning_invoked=False,
+                    dm_sent=False,
+                    buffered_skipped=True,
+                    buffered_consecutive_count=buffered_count,
+                )
+
         # KR-PROBE-INVESTIGATION-DATA-COMPLETION — wall-clock start
         # so the investigation_completed audit can carry
         # investigation_duration_ms. Also the lower bound for the
diff --git a/kora_cli/promote/_shared/__init__.py b/kora_cli/promote/_shared/__init__.py
new file mode 100644
index 000000000000..520e54eb8676
--- /dev/null
+++ b/kora_cli/promote/_shared/__init__.py
@@ -0,0 +1,6 @@
+"""Shared helpers for the 3 propose-then-approve promotion loops added
+in KR-PROMOTE-LOOPS-COMPLETION-MEGABUCKET (router-tuning, tool-trimming,
+probe-fix-envelopes). Keeps each loop's per-module footprint small
+while preserving the phrasebook (#186) on-disk pending/approved/
+rejected/expired layout operators already know.
+"""
diff --git a/kora_cli/promote/_shared/proposal_store.py b/kora_cli/promote/_shared/proposal_store.py
new file mode 100644
index 000000000000..f790af024077
--- /dev/null
+++ b/kora_cli/promote/_shared/proposal_store.py
@@ -0,0 +1,255 @@
+"""Generic propose-then-approve file-backed store for promotion loops.
+
+Extracted from the phrasebook (#186) store to avoid duplicating the
+same pending/approved/rejected/expired filesystem layout across the
+3 new propose-then-approve loops introduced in
+KR-PROMOTE-LOOPS-COMPLETION-MEGABUCKET.
+
+# Why per-loop subdirs (not one shared dir)
+
+Each loop's proposal payload has a distinct shape; mixing them in one
+directory would require a discriminator field + per-row type guards.
+Per-loop subdirs keep operator triage trivial:
+
+  ``${KORA_HOME}/promotions/<loop_name>/{pending,approved,rejected,expired}``
+
+# Payload contract
+
+The store is payload-agnostic: it accepts a JSON-serializable dict
+keyed by ``proposal_id``. Each loop owns its own dataclass + (de)serialize
+helpers; this module only handles atomic-write, list, status
+transition, and expiry.
+
+# Out of scope
+
+  * Audit emission — each loop owns its own audit seam + emit call
+    (the store is filesystem-only)
+  * Operator-edit overrides — phrasebook (#186) needed a whitelist for
+    pattern/reply_template/category overrides at approve-time; the
+    new loops either don't take overrides (router-tuning, probe-
+    envelopes) or take simpler ones (tool-trimming may take a
+    per-tool retain decision but those land via the payload)
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Tuple
+
+logger = logging.getLogger(__name__)
+
+
+PROMOTIONS_ROOT_ENV = "KORA_PROMOTIONS_DIR"
+_PROMOTIONS_RELATIVE = Path("promotions")
+
+STATUS_VALUES: Tuple[str, ...] = (
+    "pending",
+    "approved",
+    "rejected",
+    "expired",
+)
+
+
+class ProposalNotFound(LookupError):
+    """Raised when an endpoint references a proposal_id that doesn't
+    exist in any of the status subdirectories."""
+
+
+def _root(loop_name: str) -> Path:
+    """Resolve the loop's promotions root.
+
+    Honors ``KORA_PROMOTIONS_DIR`` for tests that don't want to use
+    the canonical KORA_HOME path; falls back to
+    ``${KORA_HOME}/promotions/<loop_name>`` otherwise.
+    """
+    override = os.environ.get(PROMOTIONS_ROOT_ENV, "").strip()
+    if override:
+        return Path(override) / loop_name
+    from kora_constants import get_kora_home
+
+    return get_kora_home() / _PROMOTIONS_RELATIVE / loop_name
+
+
+def _status_dir(loop_name: str, status: str) -> Path:
+    if status not in STATUS_VALUES:
+        raise ValueError(f"unknown proposal status: {status!r}")
+    return _root(loop_name) / status
+
+
+def save_pending(
+    *, loop_name: str, proposal_id: str, payload: Dict[str, Any]
+) -> Path:
+    """Atomic-write a pending proposal. Returns the file path.
+
+    ``payload`` is written verbatim. Caller is responsible for
+    setting ``status="pending"`` inside it if the loop's shape
+    carries a status field — the store doesn't enforce that.
+    """
+    target_dir = _status_dir(loop_name, "pending")
+    target_dir.mkdir(parents=True, exist_ok=True)
+    target = target_dir / f"{proposal_id}.json"
+    tmp = target.with_suffix(".json.tmp")
+    tmp.write_text(
+        json.dumps(payload, indent=2, sort_keys=True),
+        encoding="utf-8",
+    )
+    os.replace(tmp, target)
+    return target
+
+
+def list_by_status(
+    *, loop_name: str, status: str
+) -> List[Dict[str, Any]]:
+    """Return all proposals in a status directory (unordered).
+
+    Caller can sort by whatever proposal-shape field it cares about
+    (confidence / cluster_size / created_at). Loops with a wire-
+    stable ordering convention apply it in their endpoint handler.
+    """
+    target_dir = _status_dir(loop_name, status)
+    if not target_dir.is_dir():
+        return []
+    out: List[Dict[str, Any]] = []
+    for path in sorted(target_dir.iterdir()):
+        if not path.is_file() or path.suffix != ".json":
+            continue
+        try:
+            payload = json.loads(path.read_text(encoding="utf-8"))
+        except Exception as exc:
+            logger.warning(
+                "[kora.promote.store] %s unreadable, skipped: %r",
+                path,
+                exc,
+            )
+            continue
+        if isinstance(payload, dict):
+            out.append(payload)
+    return out
+
+
+def load(*, loop_name: str, proposal_id: str) -> Tuple[str, Dict[str, Any]]:
+    """Look up a proposal across all status directories.
+
+    Returns ``(current_status, payload)``. Raises
+    :class:`ProposalNotFound` when the file is missing.
+    """
+    for status in STATUS_VALUES:
+        path = _status_dir(loop_name, status) / f"{proposal_id}.json"
+        if path.is_file():
+            try:
+                payload = json.loads(path.read_text(encoding="utf-8"))
+            except Exception as exc:
+                raise ProposalNotFound(
+                    f"proposal {proposal_id!r} exists at {path} but "
+                    f"failed to load: {exc!r}"
+                ) from exc
+            if not isinstance(payload, dict):
+                raise ProposalNotFound(
+                    f"proposal {proposal_id!r} at {path} is not a JSON object"
+                )
+            return status, payload
+    raise ProposalNotFound(f"proposal {proposal_id!r} not found")
+
+
+def transition(
+    *,
+    loop_name: str,
+    proposal_id: str,
+    new_status: str,
+    payload_mutator: Optional[Any] = None,
+) -> Tuple[str, Dict[str, Any]]:
+    """Move a proposal from its current status to ``new_status``.
+
+    Returns ``(old_status, post_transition_payload)``. Optionally
+    runs ``payload_mutator(payload)`` to mutate the dict in place
+    before persisting at the new status (callers use this to stamp
+    review_notes / approver / etc.).
+
+    Write-before-unlink shape — a crash mid-transition leaves BOTH
+    files; recovery is operator-readable (list both directories).
+    """
+    if new_status not in STATUS_VALUES:
+        raise ValueError(f"unknown proposal status: {new_status!r}")
+    old_status, payload = load(loop_name=loop_name, proposal_id=proposal_id)
+    if callable(payload_mutator):
+        payload_mutator(payload)
+    # The store-level status field is optional; we stamp it for
+    # callers that don't bother in payload_mutator.
+    payload.setdefault("_store_status", new_status)
+    payload["_store_status"] = new_status
+
+    target_dir = _status_dir(loop_name, new_status)
+    target_dir.mkdir(parents=True, exist_ok=True)
+    target = target_dir / f"{proposal_id}.json"
+    tmp = target.with_suffix(".json.tmp")
+    tmp.write_text(
+        json.dumps(payload, indent=2, sort_keys=True),
+        encoding="utf-8",
+    )
+    os.replace(tmp, target)
+
+    old_path = _status_dir(loop_name, old_status) / f"{proposal_id}.json"
+    if old_path != target:
+        try:
+            old_path.unlink()
+        except OSError as exc:
+            logger.warning(
+                "[kora.promote.store] old path %s unlink failed: %r — "
+                "operator can clean manually",
+                old_path,
+                exc,
+            )
+    return old_status, payload
+
+
+def expire_older_than(*, loop_name: str, days: int) -> int:
+    """Move pending proposals older than ``days`` to ``expired/``.
+
+    Returns count moved. Read ``created_at`` from the payload (ISO
+    8601). Payloads without a parseable created_at are left in
+    place (defensive; future loops may not carry that field).
+    """
+    cutoff = datetime.now(timezone.utc).timestamp() - days * 86400
+    expired = 0
+    for payload in list_by_status(loop_name=loop_name, status="pending"):
+        proposal_id = str(payload.get("proposal_id") or "")
+        if not proposal_id:
+            continue
+        ts_raw = payload.get("created_at")
+        if not isinstance(ts_raw, str) or not ts_raw:
+            continue
+        try:
+            if ts_raw.endswith("Z"):
+                ts_raw_parsed = ts_raw[:-1] + "+00:00"
+            else:
+                ts_raw_parsed = ts_raw
+            ts = datetime.fromisoformat(ts_raw_parsed).timestamp()
+        except ValueError:
+            continue
+        if ts < cutoff:
+            try:
+                transition(
+                    loop_name=loop_name,
+                    proposal_id=proposal_id,
+                    new_status="expired",
+                    payload_mutator=lambda p: p.update(
+                        {
+                            "review_notes": (
+                                f"auto-expired after {days} days pending"
+                            )
+                        }
+                    ),
+                )
+                expired += 1
+            except Exception as exc:
+                logger.warning(
+                    "[kora.promote.store] expire transition failed for "
+                    "%s: %r",
+                    proposal_id,
+                    exc,
+                )
+    return expired
diff --git a/kora_cli/promote/probe_fix_envelopes/__init__.py b/kora_cli/promote/probe_fix_envelopes/__init__.py
new file mode 100644
index 000000000000..48aefb589977
--- /dev/null
+++ b/kora_cli/promote/probe_fix_envelopes/__init__.py
@@ -0,0 +1,50 @@
+"""Probe-fix-envelope promotion loop — KR-PROMOTE-PROBE-FIX-ENVELOPES.
+
+Fifth and final promotion loop. Observes recurring probe failures
+where Kora's investigation summaries point at a consistent
+recommended fix, and proposes adding a new envelope action to
+``probes/fix_envelopes.py``.
+
+# Loop shape
+
+  1. :mod:`.observer` — read ``probe.investigation_completed``
+     (#184) + ``tool.probe_autofix_attempted`` (#182) audit rows.
+     Cluster by ``probe`` + ``issue_category`` over the
+     observation window; carry ``investigation_summary_text`` and
+     fix-attempt outcomes so the proposer can see the pattern.
+  2. :mod:`.proposer` — for each cluster ≥ min_cluster_size,
+     propose a new envelope action with:
+       * ``fix_name_suggestion`` — derived from probe + category
+       * ``recurring_recommendation_text`` — extracted from the
+         cluster's investigation summaries
+       * ``blast_radius_summary`` — operator-facing risk
+         description (defaults to "operator must review")
+  3. Store + audit + endpoint follow the phrasebook (#186) shape.
+  4. :mod:`.plugin` — orchestrator + listener wiring.
+
+# Cost discipline
+
+$0 LLM. Pure audit-log scan + lexical cluster + text projection.
+
+# Auto-apply (HARDCODED FALSE per spec)
+
+Per spec §2 deliverable C: auto-apply is HARDCODED FALSE for v1.
+Adding operator-authorized fix actions to Kora's envelope file
+mutates what Kora is permitted to do to production infra without
+operator-in-the-loop. Per
+``feedback-fail-closed-by-default-for-security-infra`` and
+``feedback-promotion-loops-self-improving-subsystems``: never
+auto-apply high-risk loops.
+
+# Approval path (manual scaffolding — STOP-ASK §4 mitigation)
+
+Per spec STOP-ASK §4: writing to declarative ``fix_envelopes.py``
+via codegen is fragile. The approve endpoint transitions the
+proposal status + emits ``promotion.approved`` — it does NOT
+modify ``fix_envelopes.py``. Operator manually scaffolds the
+approved envelope into the file (the proposal payload carries
+the suggested ``FixEnvelope(...)`` shape verbatim so the copy-
+paste is mechanical). Approved proposal sits in the
+``promotions/probe_fix_envelopes/approved/`` directory as the
+audit trail.
+"""
diff --git a/kora_cli/promote/probe_fix_envelopes/observer.py b/kora_cli/promote/probe_fix_envelopes/observer.py
new file mode 100644
index 000000000000..923855c32ea6
--- /dev/null
+++ b/kora_cli/promote/probe_fix_envelopes/observer.py
@@ -0,0 +1,112 @@
+"""Probe-fix observation collector — KR-PROMOTE-PROBE-FIX-ENVELOPES.
+
+Reads ``probe.investigation_completed`` (#184) audit rows and
+projects them into the shape the proposer clusters on.
+
+# Why investigation_completed (not probe.wake_requested)
+
+The wake_requested rows fire BEFORE Kora investigates. The
+investigation_completed rows carry Kora's actual recommendation
+text — the operator-decision-relevant signal the loop wants to
+cluster. wake_requested rows are still relevant for "frequency of
+this issue category" but the recommendation text only lives in
+investigation_completed.
+
+# autofix_attempted cross-reference
+
+Investigations that already triggered an autofix attempt
+(``autofix_attempted=True``) are SKIPPED — the loop's purpose is
+to propose NEW envelope actions, not reinforce existing ones.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from typing import List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True, slots=True)
+class InvestigationObservation:
+    """One investigation projected for clustering."""
+
+    probe: str
+    issue_category: str
+    severity: str
+    investigation_summary_text: str
+    caller_session_id: str
+    timestamp: datetime
+
+
+async def collect_recent_investigations(
+    *, since: Optional[datetime] = None
+) -> List[InvestigationObservation]:
+    """Read recent ``probe.investigation_completed`` audit rows.
+
+    Args:
+      since: Lower bound (aware datetime). Defaults to 14 days
+        before now — broad enough to surface recurring issues
+        while small enough that the audit JSONL read stays cheap.
+
+    Returns observations sorted by ``timestamp`` ascending.
+    Investigations that already triggered an autofix attempt
+    (``autofix_attempted=True``) are skipped — see module
+    docstring.
+    """
+    if since is None:
+        since = datetime.now(timezone.utc) - timedelta(days=14)
+    if since.tzinfo is None:
+        since = since.replace(tzinfo=timezone.utc)
+
+    try:
+        from kora_cli.audit.jsonl_reader import read_audit_entries
+    except Exception as exc:
+        logger.warning(
+            "[kora.promote.probe_fix_envelopes.observer] audit reader "
+            "import failed: %r — no observations",
+            exc,
+        )
+        return []
+
+    try:
+        entries = read_audit_entries(
+            seam="probe.investigation_completed", since=since
+        )
+    except Exception as exc:
+        logger.warning(
+            "[kora.promote.probe_fix_envelopes.observer] "
+            "read_audit_entries raised %r — no observations",
+            exc,
+        )
+        return []
+
+    out: List[InvestigationObservation] = []
+    for entry in entries:
+        details = entry.details or {}
+        if bool(details.get("autofix_attempted")):
+            continue
+        probe = details.get("probe")
+        category = details.get("issue_category")
+        if not isinstance(probe, str) or not probe:
+            continue
+        if not isinstance(category, str) or not category:
+            continue
+        summary = details.get("investigation_summary_text") or ""
+        if not isinstance(summary, str) or not summary.strip():
+            continue
+        out.append(
+            InvestigationObservation(
+                probe=str(probe),
+                issue_category=str(category),
+                severity=str(details.get("severity") or "warning"),
+                investigation_summary_text=summary,
+                caller_session_id=str(entry.caller_session_id or ""),
+                timestamp=entry.emitted_at,
+            )
+        )
+
+    out.sort(key=lambda o: o.timestamp)
+    return out
diff --git a/kora_cli/promote/probe_fix_envelopes/plugin.py b/kora_cli/promote/probe_fix_envelopes/plugin.py
new file mode 100644
index 000000000000..719000041fd4
--- /dev/null
+++ b/kora_cli/promote/probe_fix_envelopes/plugin.py
@@ -0,0 +1,202 @@
+"""Probe-fix-envelope cycle orchestrator — KR-PROMOTE-PROBE-FIX-ENVELOPES.
+
+Called by the periodic-task heartbeat (registered by
+:mod:`kora_cli.listeners.promote_probe_fix_envelopes_listener`).
+
+# Env
+
+  * ``KORA_PROMOTE_PROBE_FIX_ENABLED`` (default ``true``)
+  * ``KORA_PROMOTE_PROBE_FIX_INTERVAL_SEC`` (default 86400 = 24h)
+  * ``KORA_PROMOTE_PROBE_FIX_EXPIRY_DAYS`` (default 14)
+  * ``KORA_PROMOTE_PROBE_FIX_MIN_CLUSTER`` (default 3)
+
+# Auto-apply
+
+HARDCODED FALSE (no env). See module ``__init__`` for the
+safety rationale.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import time
+from datetime import datetime, timedelta, timezone
+from typing import Any, Dict, Optional
+
+from kora_cli.promote._shared.proposal_store import (
+    expire_older_than,
+    save_pending,
+)
+
+from .observer import collect_recent_investigations
+from .proposer import (
+    ProbeEnvelopeProposal,
+    generate_proposals,
+    proposal_to_dict,
+)
+
+logger = logging.getLogger(__name__)
+
+
+LOOP_NAME = "probe_fix_envelopes"
+
+ENABLED_ENV = "KORA_PROMOTE_PROBE_FIX_ENABLED"
+INTERVAL_SEC_ENV = "KORA_PROMOTE_PROBE_FIX_INTERVAL_SEC"
+EXPIRY_DAYS_ENV = "KORA_PROMOTE_PROBE_FIX_EXPIRY_DAYS"
+OBSERVATION_WINDOW_DAYS_ENV = "KORA_PROMOTE_PROBE_FIX_WINDOW_DAYS"
+
+DEFAULT_INTERVAL_SEC = 86400  # once daily
+DEFAULT_EXPIRY_DAYS = 14
+DEFAULT_OBSERVATION_WINDOW_DAYS = 14
+
+
+def _is_enabled() -> bool:
+    raw = os.environ.get(ENABLED_ENV, "true").strip().lower()
+    return raw in {"true", "1", "yes", "on", ""}
+
+
+def _int_env(name: str, default: int, *, minimum: int = 1) -> int:
+    raw = os.environ.get(name, "").strip()
+    if not raw:
+        return default
+    try:
+        value = int(raw)
+    except ValueError:
+        return default
+    if value < minimum:
+        return default
+    return value
+
+
+def get_interval_seconds() -> int:
+    return _int_env(INTERVAL_SEC_ENV, DEFAULT_INTERVAL_SEC, minimum=60)
+
+
+def _emit_audit(proposal: ProbeEnvelopeProposal) -> None:
+    try:
+        from kora_cli.audit.jsonl_sink import emit_audit
+    except Exception as exc:
+        logger.warning(
+            "[kora.promote.probe_fix_envelopes] audit import failed: %r",
+            exc,
+        )
+        return
+    try:
+        emit_audit(
+            "promotion.probe_envelope_action_proposed",
+            proposal_to_dict(proposal),
+            caller_session_id=(
+                f"promotion:probe_fix_envelopes:{proposal.proposal_id}"
+            ),
+            source="reasoning",
+        )
+    except Exception as exc:
+        logger.warning(
+            "[kora.promote.probe_fix_envelopes] emit_audit raised %r — "
+            "proposal persisted; audit row missing",
+            exc,
+        )
+
+
+async def run_probe_fix_envelopes_cycle(
+    *, now: Optional[datetime] = None
+) -> Dict[str, Any]:
+    """One cycle of the probe-fix-envelope promotion loop."""
+    started_dt = now or datetime.now(timezone.utc)
+    started_monotonic = time.monotonic()
+
+    summary: Dict[str, Any] = {
+        "enabled": True,
+        "observations_read": 0,
+        "proposals_generated": 0,
+        "proposals_persisted": 0,
+        "expired_count": 0,
+        "started_at": started_dt.strftime("%Y-%m-%dT%H:%M:%SZ"),
+        "duration_ms": 0,
+        # HARDCODED FALSE auto-apply for this loop — surface in
+        # summary so cycle log makes the discipline explicit.
+        "auto_apply_mode": False,
+    }
+
+    if not _is_enabled():
+        summary["enabled"] = False
+        logger.info(
+            "[kora.promote.probe_fix_envelopes] disabled (%s=false) — "
+            "skipping",
+            ENABLED_ENV,
+        )
+        summary["duration_ms"] = int(
+            (time.monotonic() - started_monotonic) * 1000
+        )
+        return summary
+
+    window_days = _int_env(
+        OBSERVATION_WINDOW_DAYS_ENV,
+        DEFAULT_OBSERVATION_WINDOW_DAYS,
+        minimum=1,
+    )
+
+    try:
+        observations = await collect_recent_investigations(
+            since=started_dt - timedelta(days=window_days),
+        )
+        summary["observations_read"] = len(observations)
+    except Exception as exc:
+        logger.warning(
+            "[kora.promote.probe_fix_envelopes] observer failed: %r", exc
+        )
+        summary["duration_ms"] = int(
+            (time.monotonic() - started_monotonic) * 1000
+        )
+        return summary
+
+    try:
+        proposals = generate_proposals(observations, now=started_dt)
+        summary["proposals_generated"] = len(proposals)
+    except Exception as exc:
+        logger.warning(
+            "[kora.promote.probe_fix_envelopes] proposer failed: %r",
+            exc,
+        )
+        summary["duration_ms"] = int(
+            (time.monotonic() - started_monotonic) * 1000
+        )
+        return summary
+
+    for proposal in proposals:
+        try:
+            save_pending(
+                loop_name=LOOP_NAME,
+                proposal_id=proposal.proposal_id,
+                payload=proposal_to_dict(proposal),
+            )
+            summary["proposals_persisted"] += 1
+        except Exception as exc:
+            logger.warning(
+                "[kora.promote.probe_fix_envelopes] persist failed for "
+                "%s: %r",
+                proposal.proposal_id,
+                exc,
+            )
+        _emit_audit(proposal)
+
+    try:
+        summary["expired_count"] = expire_older_than(
+            loop_name=LOOP_NAME,
+            days=_int_env(EXPIRY_DAYS_ENV, DEFAULT_EXPIRY_DAYS, minimum=1),
+        )
+    except Exception as exc:
+        logger.warning(
+            "[kora.promote.probe_fix_envelopes] expire_older_than "
+            "raised %r",
+            exc,
+        )
+
+    summary["duration_ms"] = int(
+        (time.monotonic() - started_monotonic) * 1000
+    )
+    logger.info(
+        "[kora.promote.probe_fix_envelopes] cycle complete: %s", summary
+    )
+    return summary
diff --git a/kora_cli/promote/probe_fix_envelopes/proposer.py b/kora_cli/promote/probe_fix_envelopes/proposer.py
new file mode 100644
index 000000000000..d7086fe678ed
--- /dev/null
+++ b/kora_cli/promote/probe_fix_envelopes/proposer.py
@@ -0,0 +1,181 @@
+"""Probe-fix envelope proposal generator — KR-PROMOTE-PROBE-FIX-ENVELOPES.
+
+Input: :class:`InvestigationObservation` from the observer.
+Output: :class:`ProbeEnvelopeProposal` records — one per
+(probe, issue_category) cluster ≥ min_cluster_size.
+
+# Cluster shape
+
+Exact-match on ``(probe, issue_category)`` — coarse but operator-
+reviewable. The proposer emits the most-common short snippet of
+the cluster's investigation summaries as the
+``recurring_recommendation_text`` so operator can see the
+recurring suggestion at-a-glance.
+
+# Blast-radius default
+
+v1 always emits a conservative ``"operator must review — proposed
+envelope action has not been classified"`` blast-radius summary.
+The operator-reviewing-the-proposal step IS the blast-radius
+review; the loop is propose-only and never auto-applies, so
+defaulting to "review-required" is fine. Operator-edit-at-
+approve-time can refine this in the persisted record.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import re
+import uuid
+from collections import Counter, defaultdict
+from dataclasses import asdict, dataclass, field
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Literal
+
+from .observer import InvestigationObservation
+
+logger = logging.getLogger(__name__)
+
+
+MIN_CLUSTER_SIZE_ENV = "KORA_PROMOTE_PROBE_FIX_MIN_CLUSTER"
+DEFAULT_MIN_CLUSTER_SIZE = 3  # lower than other loops — probe failures are
+# rarer + recurring ones are higher-signal
+
+
+ProposalStatus = Literal["pending", "approved", "rejected", "expired"]
+
+
+_DEFAULT_BLAST_RADIUS = (
+    "operator must review — proposed envelope action has not been "
+    "classified for production-mutation risk; treat as broad-impact "
+    "by default until operator narrows the scope"
+)
+
+
+@dataclass(frozen=True, slots=True)
+class ProbeEnvelopeProposal:
+    """Wire-stable proposal shape."""
+
+    proposal_id: str
+    probe: str
+    issue_category: str
+    fix_name_suggestion: str
+    cluster_size: int
+    sample_caller_session_ids: List[str]
+    recurring_recommendation_text: str
+    blast_radius_summary: str
+    confidence: float
+    created_at: datetime
+    status: ProposalStatus = "pending"
+    review_notes: str = ""
+
+
+def _format_iso(dt: datetime) -> str:
+    return dt.strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def proposal_to_dict(p: ProbeEnvelopeProposal) -> Dict[str, Any]:
+    out = asdict(p)
+    out["created_at"] = _format_iso(p.created_at)
+    out["sample_caller_session_ids"] = list(p.sample_caller_session_ids)
+    return out
+
+
+def _int_env(name: str, default: int, *, minimum: int = 1) -> int:
+    raw = os.environ.get(name, "").strip()
+    if not raw:
+        return default
+    try:
+        value = int(raw)
+    except ValueError:
+        return default
+    if value < minimum:
+        return default
+    return value
+
+
+def _suggest_fix_name(probe: str, issue_category: str) -> str:
+    """Derive a short stable id from (probe, issue_category).
+
+    The id mirrors the existing ``FixEnvelope.fix_name`` convention
+    (snake_case, narrow scope). Operator MUST rename on approval —
+    this is a placeholder rather than a final identity.
+    """
+    safe_probe = re.sub(r"[^a-z0-9]+", "_", probe.lower()).strip("_")
+    safe_cat = re.sub(r"[^a-z0-9]+", "_", issue_category.lower()).strip("_")
+    return f"proposed_{safe_probe}_{safe_cat}"
+
+
+_SUMMARY_FALLBACK_CHARS = 240
+
+
+def _recurring_recommendation_text(
+    observations: List[InvestigationObservation],
+) -> str:
+    """Pick the most-common short summary across the cluster.
+
+    Falls back to the first observation's leading 240 chars if no
+    shared substring emerges. Operator edits the result on
+    approve.
+    """
+    if not observations:
+        return ""
+    # The summaries are usually short paragraphs; cluster by first-
+    # 240-char projection so near-identical wording bundles.
+    rolled: Counter = Counter()
+    for o in observations:
+        head = o.investigation_summary_text.strip()
+        rolled[head[:_SUMMARY_FALLBACK_CHARS]] += 1
+    most_common, _count = rolled.most_common(1)[0]
+    return most_common
+
+
+def generate_proposals(
+    observations: List[InvestigationObservation],
+    *,
+    now: datetime,
+) -> List[ProbeEnvelopeProposal]:
+    """Cluster + filter + propose. Returns proposals sorted by
+    confidence descending."""
+    min_cluster_size = _int_env(
+        MIN_CLUSTER_SIZE_ENV, DEFAULT_MIN_CLUSTER_SIZE, minimum=2
+    )
+
+    clusters: Dict[tuple, List[InvestigationObservation]] = defaultdict(list)
+    for o in observations:
+        clusters[(o.probe, o.issue_category)].append(o)
+
+    out: List[ProbeEnvelopeProposal] = []
+    for (probe, category), members in clusters.items():
+        if len(members) < min_cluster_size:
+            continue
+        sample_ids: List[str] = []
+        seen = set()
+        for o in members:
+            if not o.caller_session_id or o.caller_session_id in seen:
+                continue
+            seen.add(o.caller_session_id)
+            sample_ids.append(o.caller_session_id)
+            if len(sample_ids) >= 3:
+                break
+        confidence = min(1.0, len(members) / (2 * min_cluster_size))
+        out.append(
+            ProbeEnvelopeProposal(
+                proposal_id=str(uuid.uuid4()),
+                probe=probe,
+                issue_category=category,
+                fix_name_suggestion=_suggest_fix_name(probe, category),
+                cluster_size=len(members),
+                sample_caller_session_ids=sample_ids,
+                recurring_recommendation_text=(
+                    _recurring_recommendation_text(members)
+                ),
+                blast_radius_summary=_DEFAULT_BLAST_RADIUS,
+                confidence=round(confidence, 4),
+                created_at=now,
+                status="pending",
+            )
+        )
+    out.sort(key=lambda p: (-p.confidence, -p.cluster_size, p.probe))
+    return out
diff --git a/kora_cli/promote/router_tuning/__init__.py b/kora_cli/promote/router_tuning/__init__.py
new file mode 100644
index 000000000000..5b27431415df
--- /dev/null
+++ b/kora_cli/promote/router_tuning/__init__.py
@@ -0,0 +1,36 @@
+"""Router-tuning promotion loop — KR-PROMOTE-ROUTER-TUNING.
+
+Third promotion loop. Reads per-route escalation telemetry +
+proposes operator review for routes whose Haiku-to-Opus escalation
+pattern suggests trigger tuning.
+
+# Loop shape
+
+  1. :mod:`.observer` — read ``cost_telemetry.snapshot()`` per-route
+     ``escalation_count`` / ``calls_count`` for the rolling 24h
+     window. Per-route quality data ("was the Opus reply materially
+     better than Haiku?") isn't exposed today — see ``plugin`` for
+     the v1 scope decision.
+  2. :mod:`.proposer` — score each route by escalation_rate +
+     volume. High-rate routes get a ``tighten_review``
+     recommendation; routes with notably-low escalation despite
+     repeated operator ``/opus`` overrides (future-data) would get a
+     ``loosen_review``. v1 only ships ``tighten_review`` since the
+     override-observation data isn't yet collected.
+  3. Store + audit + endpoint follow the phrasebook (#186)
+     pending/approve/reject template via
+     :mod:`kora_cli.promote._shared.proposal_store`.
+  4. :mod:`.plugin` — orchestrator + listener wiring.
+
+# Cost discipline
+
+$0 LLM. The proposer reads telemetry counters + emits proposals
+purely from threshold math. Per cycle: $0. Combined with the other
+4 loops, still inside the [[feedback-promotion-loops-self-improving-
+subsystems]] $0.01-0.05/day target.
+
+# Auto-apply
+
+DEFAULT FALSE. Router-tuning changes trigger patterns that affect
+every reasoning call; operator MUST review before any change.
+"""
diff --git a/kora_cli/promote/router_tuning/observer.py b/kora_cli/promote/router_tuning/observer.py
new file mode 100644
index 000000000000..971a607296f6
--- /dev/null
+++ b/kora_cli/promote/router_tuning/observer.py
@@ -0,0 +1,106 @@
+"""Per-route escalation observer — KR-PROMOTE-ROUTER-TUNING.
+
+Reads the live :func:`kora_cli.telemetry.get_telemetry` singleton's
+:meth:`snapshot` and projects per-route rollups for the proposer.
+
+# v1 scope (STOP-ASK §4)
+
+The bucket STOP-ASK §4 anticipated: the observer wants per-call
+"was the Opus reply materially better than Haiku's would be?" data.
+That data isn't exposed today — collecting it would need either:
+
+  1. A second Haiku call per Opus call to do post-hoc quality
+     scoring (doubles spend on every escalation), OR
+  2. A new audit row whenever the operator manually issues
+     ``/opus`` to fix a Haiku miss (not yet emitted).
+
+v1 ships with the data we DO have: ``calls_count`` +
+``escalation_count`` per route. That's enough to surface "this
+route escalates 60% of the time — please review the trigger" for
+operator-attention; the actual tuning decision stays operator-
+gated regardless. A future bucket can wire option 2 (cheap; one
+audit row per operator override) to refine the rationale.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from typing import Dict, List
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True, slots=True)
+class RouteEscalationRollup:
+    """One route's projection from cost_telemetry's rolling_24h window."""
+
+    route: str
+    calls_count: int
+    escalation_count: int
+    escalation_rate: float  # 0.0..1.0; 0.0 when calls_count == 0
+    cost_estimate_usd_total: float
+
+
+def _safe_rate(escalations: int, total: int) -> float:
+    if total <= 0:
+        return 0.0
+    return min(1.0, escalations / total)
+
+
+def collect_route_rollups() -> List[RouteEscalationRollup]:
+    """Return per-route rollups from the live cost_telemetry singleton.
+
+    Fail-soft: telemetry singleton unavailable / snapshot raises →
+    empty list. Proposer treats that as "no data this cycle, no
+    proposals" — same fail-soft contract as the other promotion
+    loops.
+
+    Routes are sorted alphabetically for stable test assertions;
+    the proposer reorders by score before emitting.
+    """
+    try:
+        from kora_cli.telemetry import (
+            WINDOW_ROLLING_24H,
+            get_telemetry,
+        )
+    except Exception as exc:
+        logger.debug(
+            "[kora.promote.router_tuning.observer] telemetry import "
+            "failed: %r — no rollups",
+            exc,
+        )
+        return []
+
+    try:
+        all_windows = get_telemetry().snapshot()
+    except Exception as exc:
+        logger.warning(
+            "[kora.promote.router_tuning.observer] telemetry.snapshot() "
+            "raised %r — no rollups",
+            exc,
+        )
+        return []
+
+    window_data = all_windows.get(WINDOW_ROLLING_24H, {})
+    if not isinstance(window_data, dict):
+        return []
+
+    out: List[RouteEscalationRollup] = []
+    for route, counters in window_data.items():
+        if not isinstance(counters, dict):
+            continue
+        calls = int(counters.get("calls_count") or 0)
+        escs = int(counters.get("escalation_count") or 0)
+        cost = float(counters.get("cost_estimate_usd_total") or 0.0)
+        out.append(
+            RouteEscalationRollup(
+                route=str(route),
+                calls_count=calls,
+                escalation_count=escs,
+                escalation_rate=_safe_rate(escs, calls),
+                cost_estimate_usd_total=round(cost, 6),
+            )
+        )
+    out.sort(key=lambda r: r.route)
+    return out
diff --git a/kora_cli/promote/router_tuning/plugin.py b/kora_cli/promote/router_tuning/plugin.py
new file mode 100644
index 000000000000..3b2059639d0b
--- /dev/null
+++ b/kora_cli/promote/router_tuning/plugin.py
@@ -0,0 +1,198 @@
+"""Router-tuning cycle orchestrator — KR-PROMOTE-ROUTER-TUNING.
+
+Called by the periodic-task heartbeat (registered by
+:mod:`kora_cli.listeners.promote_router_tuning_listener`). One
+cycle:
+
+  1. Read rolling-24h escalation rollups via
+     :func:`observer.collect_route_rollups`.
+  2. Generate proposals via :func:`proposer.generate_proposals`.
+  3. Persist each proposal via the shared store + emit
+     ``promotion.router_trigger_proposed`` audit row.
+  4. Expire pending proposals older than ``EXPIRY_DAYS_ENV``
+     (default 14) so the operator's review queue stays bounded.
+  5. Log cycle summary.
+
+# Env
+
+  * ``KORA_PROMOTE_ROUTER_TUNING_ENABLED`` (default ``true``)
+  * ``KORA_PROMOTE_ROUTER_TUNING_INTERVAL_SEC`` (default 86400 = 24h)
+  * ``KORA_PROMOTE_ROUTER_TUNING_EXPIRY_DAYS`` (default 14)
+
+# Fail-soft
+
+Cycle exceptions log + are swallowed by the heartbeat scheduler.
+Per-proposal exceptions caught so one bad proposal doesn't poison
+the batch.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import time
+from datetime import datetime, timezone
+from typing import Any, Dict, Optional
+
+from kora_cli.promote._shared.proposal_store import (
+    expire_older_than,
+    save_pending,
+)
+
+from .observer import collect_route_rollups
+from .proposer import RouterTuningProposal, generate_proposals, proposal_to_dict
+
+logger = logging.getLogger(__name__)
+
+
+LOOP_NAME = "router_tuning"
+
+ENABLED_ENV = "KORA_PROMOTE_ROUTER_TUNING_ENABLED"
+INTERVAL_SEC_ENV = "KORA_PROMOTE_ROUTER_TUNING_INTERVAL_SEC"
+EXPIRY_DAYS_ENV = "KORA_PROMOTE_ROUTER_TUNING_EXPIRY_DAYS"
+
+DEFAULT_INTERVAL_SEC = 86400  # once daily
+DEFAULT_EXPIRY_DAYS = 14
+
+
+def _is_enabled() -> bool:
+    raw = os.environ.get(ENABLED_ENV, "true").strip().lower()
+    return raw in {"true", "1", "yes", "on", ""}
+
+
+def _int_env(name: str, default: int, *, minimum: int = 1) -> int:
+    raw = os.environ.get(name, "").strip()
+    if not raw:
+        return default
+    try:
+        value = int(raw)
+    except ValueError:
+        return default
+    if value < minimum:
+        return default
+    return value
+
+
+def get_interval_seconds() -> int:
+    return _int_env(INTERVAL_SEC_ENV, DEFAULT_INTERVAL_SEC, minimum=60)
+
+
+def _emit_audit(proposal: RouterTuningProposal) -> None:
+    try:
+        from kora_cli.audit.jsonl_sink import emit_audit
+    except Exception as exc:
+        logger.warning(
+            "[kora.promote.router_tuning] audit import failed: %r — "
+            "promotion.router_trigger_proposed skipped",
+            exc,
+        )
+        return
+    try:
+        emit_audit(
+            "promotion.router_trigger_proposed",
+            proposal_to_dict(proposal),
+            caller_session_id=(
+                f"promotion:router_tuning:{proposal.proposal_id}"
+            ),
+            source="reasoning",
+        )
+    except Exception as exc:
+        logger.warning(
+            "[kora.promote.router_tuning] emit_audit raised %r — "
+            "proposal persisted; audit row missing",
+            exc,
+        )
+
+
+async def run_router_tuning_cycle(
+    *, now: Optional[datetime] = None
+) -> Dict[str, Any]:
+    """One cycle of the router-tuning promotion loop.
+
+    Returns a summary dict the heartbeat scheduler logs at INFO.
+    """
+    started_dt = now or datetime.now(timezone.utc)
+    started_monotonic = time.monotonic()
+
+    summary: Dict[str, Any] = {
+        "enabled": True,
+        "rollups_observed": 0,
+        "proposals_generated": 0,
+        "proposals_persisted": 0,
+        "expired_count": 0,
+        "started_at": started_dt.strftime("%Y-%m-%dT%H:%M:%SZ"),
+        "duration_ms": 0,
+    }
+
+    if not _is_enabled():
+        summary["enabled"] = False
+        logger.info(
+            "[kora.promote.router_tuning] disabled (%s=false) — skipping",
+            ENABLED_ENV,
+        )
+        summary["duration_ms"] = int(
+            (time.monotonic() - started_monotonic) * 1000
+        )
+        return summary
+
+    try:
+        rollups = collect_route_rollups()
+        summary["rollups_observed"] = len(rollups)
+    except Exception as exc:
+        logger.warning(
+            "[kora.promote.router_tuning] observer failed: %r — "
+            "no proposals generated",
+            exc,
+        )
+        summary["duration_ms"] = int(
+            (time.monotonic() - started_monotonic) * 1000
+        )
+        return summary
+
+    try:
+        proposals = generate_proposals(rollups, now=started_dt)
+        summary["proposals_generated"] = len(proposals)
+    except Exception as exc:
+        logger.warning(
+            "[kora.promote.router_tuning] proposer failed: %r", exc
+        )
+        summary["duration_ms"] = int(
+            (time.monotonic() - started_monotonic) * 1000
+        )
+        return summary
+
+    for proposal in proposals:
+        try:
+            save_pending(
+                loop_name=LOOP_NAME,
+                proposal_id=proposal.proposal_id,
+                payload=proposal_to_dict(proposal),
+            )
+            summary["proposals_persisted"] += 1
+        except Exception as exc:
+            logger.warning(
+                "[kora.promote.router_tuning] persist failed for "
+                "%s: %r — proposal lost (audit row still emitted)",
+                proposal.proposal_id,
+                exc,
+            )
+        _emit_audit(proposal)
+
+    try:
+        summary["expired_count"] = expire_older_than(
+            loop_name=LOOP_NAME,
+            days=_int_env(EXPIRY_DAYS_ENV, DEFAULT_EXPIRY_DAYS, minimum=1),
+        )
+    except Exception as exc:
+        logger.warning(
+            "[kora.promote.router_tuning] expire_older_than raised %r",
+            exc,
+        )
+
+    summary["duration_ms"] = int(
+        (time.monotonic() - started_monotonic) * 1000
+    )
+    logger.info(
+        "[kora.promote.router_tuning] cycle complete: %s", summary
+    )
+    return summary
diff --git a/kora_cli/promote/router_tuning/proposer.py b/kora_cli/promote/router_tuning/proposer.py
new file mode 100644
index 000000000000..a57147ff79cb
--- /dev/null
+++ b/kora_cli/promote/router_tuning/proposer.py
@@ -0,0 +1,179 @@
+"""Router-tuning proposal generator — KR-PROMOTE-ROUTER-TUNING.
+
+Input: per-route :class:`RouteEscalationRollup` from the observer.
+Output: zero or more :class:`RouterTuningProposal` records — one
+per route whose escalation pattern crosses an operator-attention
+threshold.
+
+# Thresholds (operator-tunable via env)
+
+  * ``KORA_PROMOTE_ROUTER_TUNING_MIN_CALLS`` (default 20) —
+    minimum calls in window before a route gets considered.
+    Below this, sample size is too noisy.
+  * ``KORA_PROMOTE_ROUTER_TUNING_TIGHTEN_THRESHOLD`` (default 0.40)
+    — escalation_rate ≥ this on an eligible route → tighten_review
+    proposal. (Default 40% — well above the natural escalation
+    baseline of <15% from healthy decision-language patterns.)
+
+# Why no loosen_review in v1
+
+The signal for ``loosen_review`` is "operator overrode Haiku to
+Opus via /opus N times" — that observation doesn't have its own
+audit row yet (it lives in the routing decision logs, not the
+JSONL audit). Future bucket can emit a ``router.operator_override``
+seam; the proposer here would then surface routes with high
+override-rate as loosen candidates. Documented in
+``__init__.py`` v1 scope.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import uuid
+from dataclasses import asdict, dataclass
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Literal, Tuple
+
+from .observer import RouteEscalationRollup
+
+logger = logging.getLogger(__name__)
+
+
+MIN_CALLS_ENV = "KORA_PROMOTE_ROUTER_TUNING_MIN_CALLS"
+TIGHTEN_THRESHOLD_ENV = "KORA_PROMOTE_ROUTER_TUNING_TIGHTEN_THRESHOLD"
+
+DEFAULT_MIN_CALLS = 20
+DEFAULT_TIGHTEN_THRESHOLD = 0.40
+
+
+ProposalStatus = Literal["pending", "approved", "rejected", "expired"]
+RecommendationKind = Literal["tighten_review", "loosen_review"]
+
+
+@dataclass(frozen=True, slots=True)
+class RouterTuningProposal:
+    """Wire-stable proposal shape. Mirrors the snapshot_expand /
+    phrasebook proposal shape conventions (proposal_id /
+    cluster_size / confidence / created_at / status)."""
+
+    proposal_id: str
+    route: str
+    calls_count: int
+    escalation_count: int
+    escalation_rate: float  # 0.0..1.0
+    cost_estimate_usd_total: float
+    recommendation_kind: RecommendationKind
+    rationale: str
+    confidence: float  # derived from sample size; 0.0..1.0
+    created_at: datetime
+    status: ProposalStatus = "pending"
+    review_notes: str = ""
+
+
+def _format_iso(dt: datetime) -> str:
+    return dt.strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def proposal_to_dict(p: RouterTuningProposal) -> Dict[str, Any]:
+    out = asdict(p)
+    out["created_at"] = _format_iso(p.created_at)
+    return out
+
+
+def _float_env(name: str, default: float, *, minimum: float = 0.0) -> float:
+    raw = os.environ.get(name, "").strip()
+    if not raw:
+        return default
+    try:
+        value = float(raw)
+    except ValueError:
+        logger.warning(
+            "[kora.promote.router_tuning.proposer] %s=%r not numeric — "
+            "using default %f",
+            name,
+            raw,
+            default,
+        )
+        return default
+    if value < minimum:
+        return default
+    return value
+
+
+def _int_env(name: str, default: int, *, minimum: int = 0) -> int:
+    raw = os.environ.get(name, "").strip()
+    if not raw:
+        return default
+    try:
+        value = int(raw)
+    except ValueError:
+        return default
+    if value < minimum:
+        return default
+    return value
+
+
+def _confidence_from_calls(calls: int, *, threshold_calls: int) -> float:
+    """Map the call count to a 0..1 confidence band.
+
+    At threshold_calls we land at 0.5 (just enough); 4× threshold
+    earns near-1.0; below threshold isn't proposed anyway.
+    """
+    if calls <= 0:
+        return 0.0
+    return min(1.0, (calls / (2 * threshold_calls)))
+
+
+def generate_proposals(
+    rollups: List[RouteEscalationRollup],
+    *,
+    now: datetime,
+) -> List[RouterTuningProposal]:
+    """Cluster + filter + propose. Returns proposals sorted by
+    confidence descending (matches the phrasebook convention so the
+    cockpit can use a single ordering rule across all loops)."""
+    min_calls = _int_env(MIN_CALLS_ENV, DEFAULT_MIN_CALLS, minimum=2)
+    tighten_threshold = _float_env(
+        TIGHTEN_THRESHOLD_ENV,
+        DEFAULT_TIGHTEN_THRESHOLD,
+        minimum=0.0,
+    )
+    if tighten_threshold > 1.0:
+        tighten_threshold = 1.0
+
+    out: List[RouterTuningProposal] = []
+    for r in rollups:
+        if r.calls_count < min_calls:
+            continue
+        if r.escalation_rate < tighten_threshold:
+            continue
+        confidence = _confidence_from_calls(
+            r.calls_count, threshold_calls=min_calls
+        )
+        rationale = (
+            f"Route {r.route!r} escalated to Opus on "
+            f"{r.escalation_count}/{r.calls_count} calls "
+            f"({r.escalation_rate * 100:.1f}%) in the rolling 24h "
+            f"window. Per-call escalations cost a full Opus turn on "
+            f"top of the original Haiku turn. Operator review of the "
+            f"escalation trigger pattern for this route is "
+            f"recommended; spend so far: ${r.cost_estimate_usd_total:.4f}."
+        )
+        out.append(
+            RouterTuningProposal(
+                proposal_id=str(uuid.uuid4()),
+                route=r.route,
+                calls_count=r.calls_count,
+                escalation_count=r.escalation_count,
+                escalation_rate=round(r.escalation_rate, 4),
+                cost_estimate_usd_total=r.cost_estimate_usd_total,
+                recommendation_kind="tighten_review",
+                rationale=rationale,
+                confidence=round(confidence, 4),
+                created_at=now,
+                status="pending",
+            )
+        )
+    out.sort(key=lambda p: (-p.confidence, -p.escalation_rate))
+    return out
diff --git a/kora_cli/promote/tool_trimming/__init__.py b/kora_cli/promote/tool_trimming/__init__.py
new file mode 100644
index 000000000000..c4d14db87217
--- /dev/null
+++ b/kora_cli/promote/tool_trimming/__init__.py
@@ -0,0 +1,34 @@
+"""Tool-trimming promotion loop — KR-PROMOTE-TOOL-TRIMMING.
+
+Fourth promotion loop. Reads ``reasoning.tool_called`` audit rows
+per (route, tool_name) over an observation window and proposes
+adding unused tools to a route's drop-list — so the LLM doesn't
+spend tokens on tool descriptions it never invokes for that route.
+
+# Loop shape
+
+  1. :mod:`.observer` — tally tool calls by (route, tool_name) over
+     the last N days from the audit JSONL.
+  2. :mod:`.proposer` — for each route with ≥ min_total_calls,
+     identify tools registered for the route that had ZERO calls in
+     the window. Propose adding them to a drop-list. The list of
+     tools registered per route is sourced from a snapshot of the
+     reasoning engine's tool registry (or — when unavailable —
+     the union of all tool names observed across routes).
+  3. Store + audit + endpoint follow the phrasebook (#186) shape
+     via :mod:`kora_cli.promote._shared.proposal_store`.
+  4. :mod:`.plugin` — orchestrator + listener wiring.
+
+# Cost discipline
+
+$0 LLM. Pure audit-log scan + set diff. Per cycle: $0.
+
+# Enforcement (deferred)
+
+v1 is propose-only. Actual enforcement of the per-route drop-list
+lands in the future KR-PLUGIN-TOOL-DESC-TRIM bucket, which will
+read the approved set from the operator-curated config and
+respect it inside the ``pre_tool_list_finalized`` hook (which is
+a no-op today — see ``kora_cli/reasoning/kora_hermes_plugin/
+plugin.py::_pre_tool_list_finalized``).
+"""
diff --git a/kora_cli/promote/tool_trimming/observer.py b/kora_cli/promote/tool_trimming/observer.py
new file mode 100644
index 000000000000..319e975f4bdf
--- /dev/null
+++ b/kora_cli/promote/tool_trimming/observer.py
@@ -0,0 +1,110 @@
+"""Tool-call usage observer — KR-PROMOTE-TOOL-TRIMMING.
+
+Reads ``reasoning.tool_called`` audit rows and tallies them by
+(route, tool_name) over the observation window. Returns a nested
+dict the proposer consumes.
+
+# Route attribution
+
+The ``reasoning.tool_called`` writer (see
+``kora_cli/reasoning/anthropic_engine.py::_emit_tool_called_audit``)
+stamps ``route`` inside the audit ``details`` dict from the
+engine's per-call source mapping (post KR-PROMOTE-EXPAND-AND-
+TELEMETRY-WIRES). Rows without a ``route`` field bucket into
+``"unknown"`` per the cost-telemetry taxonomy.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections import defaultdict
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from typing import Dict, List, Optional, Set
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True, slots=True)
+class RouteToolUsage:
+    """Per-route tool-usage rollup over the observation window."""
+
+    route: str
+    total_calls: int
+    tools_called: Set[str]  # tool names with ≥1 call
+    per_tool_calls: Dict[str, int]  # tool_name → call count
+
+
+async def collect_route_tool_usage(
+    *, since: Optional[datetime] = None
+) -> List[RouteToolUsage]:
+    """Tally ``reasoning.tool_called`` entries by (route, tool_name).
+
+    Args:
+      since: Lower bound (aware datetime). Defaults to 30 days
+        before now — long enough to surface durably-unused tools
+        without being polluted by a single bursty week.
+    """
+    if since is None:
+        since = datetime.now(timezone.utc) - timedelta(days=30)
+    if since.tzinfo is None:
+        since = since.replace(tzinfo=timezone.utc)
+
+    try:
+        from kora_cli.audit.jsonl_reader import read_audit_entries
+    except Exception as exc:
+        logger.warning(
+            "[kora.promote.tool_trimming.observer] audit reader import "
+            "failed: %r — no rollups",
+            exc,
+        )
+        return []
+
+    try:
+        entries = read_audit_entries(seam="reasoning.tool_called", since=since)
+    except Exception as exc:
+        logger.warning(
+            "[kora.promote.tool_trimming.observer] read_audit_entries "
+            "raised %r — no rollups",
+            exc,
+        )
+        return []
+
+    by_route_tool: Dict[str, Dict[str, int]] = defaultdict(
+        lambda: defaultdict(int)
+    )
+    for entry in entries:
+        details = entry.details or {}
+        tool_name = details.get("tool_name")
+        if not isinstance(tool_name, str) or not tool_name:
+            continue
+        # Route can live either in ``details["route"]`` (KR-PROMOTE-
+        # EXPAND-AND-TELEMETRY-WIRES future write site) or fall back
+        # to a derived caller_session_id prefix; v1 reads both.
+        route = details.get("route")
+        if not isinstance(route, str) or not route:
+            csid = entry.caller_session_id or ""
+            if csid.startswith("probe:"):
+                route = "probe_investigation"
+            elif csid.startswith("email:"):
+                route = "email_inbound"
+            elif csid.startswith("mcp:"):
+                route = "mcp_tool"
+            elif csid:
+                route = "slack_dm"
+            else:
+                route = "unknown"
+        by_route_tool[route][tool_name] += 1
+
+    out: List[RouteToolUsage] = []
+    for route, tool_map in sorted(by_route_tool.items()):
+        total = sum(tool_map.values())
+        out.append(
+            RouteToolUsage(
+                route=route,
+                total_calls=total,
+                tools_called=set(tool_map.keys()),
+                per_tool_calls=dict(tool_map),
+            )
+        )
+    return out
diff --git a/kora_cli/promote/tool_trimming/plugin.py b/kora_cli/promote/tool_trimming/plugin.py
new file mode 100644
index 000000000000..64604171f9f2
--- /dev/null
+++ b/kora_cli/promote/tool_trimming/plugin.py
@@ -0,0 +1,191 @@
+"""Tool-trimming cycle orchestrator — KR-PROMOTE-TOOL-TRIMMING.
+
+Called by the periodic-task heartbeat (registered by
+:mod:`kora_cli.listeners.promote_tool_trimming_listener`).
+
+# Env
+
+  * ``KORA_PROMOTE_TOOL_TRIMMING_ENABLED`` (default ``true``)
+  * ``KORA_PROMOTE_TOOL_TRIMMING_INTERVAL_SEC`` (default 86400 = 24h)
+  * ``KORA_PROMOTE_TOOL_TRIMMING_EXPIRY_DAYS`` (default 14)
+  * Proposer-side: ``KORA_PROMOTE_TOOL_TRIMMING_MIN_CALLS``,
+    ``KORA_PROMOTE_TOOL_TRIMMING_WINDOW_DAYS``
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import time
+from datetime import datetime, timedelta, timezone
+from typing import Any, Dict, Optional
+
+from kora_cli.promote._shared.proposal_store import (
+    expire_older_than,
+    save_pending,
+)
+
+from .observer import collect_route_tool_usage
+from .proposer import (
+    DEFAULT_OBSERVATION_WINDOW_DAYS,
+    OBSERVATION_WINDOW_DAYS_ENV,
+    ToolTrimProposal,
+    generate_proposals,
+    proposal_to_dict,
+)
+
+logger = logging.getLogger(__name__)
+
+
+LOOP_NAME = "tool_trimming"
+
+ENABLED_ENV = "KORA_PROMOTE_TOOL_TRIMMING_ENABLED"
+INTERVAL_SEC_ENV = "KORA_PROMOTE_TOOL_TRIMMING_INTERVAL_SEC"
+EXPIRY_DAYS_ENV = "KORA_PROMOTE_TOOL_TRIMMING_EXPIRY_DAYS"
+
+DEFAULT_INTERVAL_SEC = 86400  # once daily
+DEFAULT_EXPIRY_DAYS = 14
+
+
+def _is_enabled() -> bool:
+    raw = os.environ.get(ENABLED_ENV, "true").strip().lower()
+    return raw in {"true", "1", "yes", "on", ""}
+
+
+def _int_env(name: str, default: int, *, minimum: int = 1) -> int:
+    raw = os.environ.get(name, "").strip()
+    if not raw:
+        return default
+    try:
+        value = int(raw)
+    except ValueError:
+        return default
+    if value < minimum:
+        return default
+    return value
+
+
+def get_interval_seconds() -> int:
+    return _int_env(INTERVAL_SEC_ENV, DEFAULT_INTERVAL_SEC, minimum=60)
+
+
+def _emit_audit(proposal: ToolTrimProposal) -> None:
+    try:
+        from kora_cli.audit.jsonl_sink import emit_audit
+    except Exception as exc:
+        logger.warning(
+            "[kora.promote.tool_trimming] audit import failed: %r", exc
+        )
+        return
+    try:
+        emit_audit(
+            "promotion.tool_trim_proposed",
+            proposal_to_dict(proposal),
+            caller_session_id=(
+                f"promotion:tool_trimming:{proposal.proposal_id}"
+            ),
+            source="reasoning",
+        )
+    except Exception as exc:
+        logger.warning(
+            "[kora.promote.tool_trimming] emit_audit raised %r — "
+            "proposal persisted; audit row missing",
+            exc,
+        )
+
+
+async def run_tool_trimming_cycle(
+    *, now: Optional[datetime] = None
+) -> Dict[str, Any]:
+    """One cycle of the tool-trimming promotion loop."""
+    started_dt = now or datetime.now(timezone.utc)
+    started_monotonic = time.monotonic()
+
+    summary: Dict[str, Any] = {
+        "enabled": True,
+        "rollups_observed": 0,
+        "proposals_generated": 0,
+        "proposals_persisted": 0,
+        "expired_count": 0,
+        "started_at": started_dt.strftime("%Y-%m-%dT%H:%M:%SZ"),
+        "duration_ms": 0,
+    }
+
+    if not _is_enabled():
+        summary["enabled"] = False
+        logger.info(
+            "[kora.promote.tool_trimming] disabled (%s=false) — skipping",
+            ENABLED_ENV,
+        )
+        summary["duration_ms"] = int(
+            (time.monotonic() - started_monotonic) * 1000
+        )
+        return summary
+
+    window_days = _int_env(
+        OBSERVATION_WINDOW_DAYS_ENV,
+        DEFAULT_OBSERVATION_WINDOW_DAYS,
+        minimum=1,
+    )
+
+    try:
+        rollups = await collect_route_tool_usage(
+            since=started_dt - timedelta(days=window_days),
+        )
+        summary["rollups_observed"] = len(rollups)
+    except Exception as exc:
+        logger.warning(
+            "[kora.promote.tool_trimming] observer failed: %r", exc
+        )
+        summary["duration_ms"] = int(
+            (time.monotonic() - started_monotonic) * 1000
+        )
+        return summary
+
+    try:
+        proposals = generate_proposals(rollups, now=started_dt)
+        summary["proposals_generated"] = len(proposals)
+    except Exception as exc:
+        logger.warning(
+            "[kora.promote.tool_trimming] proposer failed: %r", exc
+        )
+        summary["duration_ms"] = int(
+            (time.monotonic() - started_monotonic) * 1000
+        )
+        return summary
+
+    for proposal in proposals:
+        try:
+            save_pending(
+                loop_name=LOOP_NAME,
+                proposal_id=proposal.proposal_id,
+                payload=proposal_to_dict(proposal),
+            )
+            summary["proposals_persisted"] += 1
+        except Exception as exc:
+            logger.warning(
+                "[kora.promote.tool_trimming] persist failed for "
+                "%s: %r",
+                proposal.proposal_id,
+                exc,
+            )
+        _emit_audit(proposal)
+
+    try:
+        summary["expired_count"] = expire_older_than(
+            loop_name=LOOP_NAME,
+            days=_int_env(EXPIRY_DAYS_ENV, DEFAULT_EXPIRY_DAYS, minimum=1),
+        )
+    except Exception as exc:
+        logger.warning(
+            "[kora.promote.tool_trimming] expire_older_than raised %r",
+            exc,
+        )
+
+    summary["duration_ms"] = int(
+        (time.monotonic() - started_monotonic) * 1000
+    )
+    logger.info(
+        "[kora.promote.tool_trimming] cycle complete: %s", summary
+    )
+    return summary
diff --git a/kora_cli/promote/tool_trimming/proposer.py b/kora_cli/promote/tool_trimming/proposer.py
new file mode 100644
index 000000000000..326455763594
--- /dev/null
+++ b/kora_cli/promote/tool_trimming/proposer.py
@@ -0,0 +1,140 @@
+"""Tool-trim proposal generator — KR-PROMOTE-TOOL-TRIMMING.
+
+For each route with sufficient call volume, identifies tools that
+were never called in the window and proposes adding them to the
+route's drop-list.
+
+# "Tools registered for a route" sourcing
+
+v1 uses the **union of all tools observed across routes** as the
+"available tools" set. Reason: extracting the per-route registered
+tool list from the live engine requires importing a heavy chain
+(``listeners.mcp_tools`` → MCP transport) that's neither cheap
+nor available outside the daemon. The union-across-observation is
+a conservative proxy — it gives operator the set of tools that
+SOMEONE called, but THIS route didn't, which is exactly the
+operator-attention signal.
+
+A future bucket can replace ``_available_tool_names`` with a real
+per-route registered-tool projection once
+``pre_tool_list_finalized`` ships a side-channel that records the
+registered list per route. The proposer's interface (the
+``available_tool_names`` set) is the seam.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import uuid
+from dataclasses import asdict, dataclass, field
+from datetime import datetime, timezone
+from typing import Any, Dict, Iterable, List, Literal, Set
+
+from .observer import RouteToolUsage
+
+logger = logging.getLogger(__name__)
+
+
+MIN_CALLS_ENV = "KORA_PROMOTE_TOOL_TRIMMING_MIN_CALLS"
+OBSERVATION_WINDOW_DAYS_ENV = "KORA_PROMOTE_TOOL_TRIMMING_WINDOW_DAYS"
+
+DEFAULT_MIN_CALLS = 20
+DEFAULT_OBSERVATION_WINDOW_DAYS = 30
+
+
+ProposalStatus = Literal["pending", "approved", "rejected", "expired"]
+
+
+@dataclass(frozen=True, slots=True)
+class ToolTrimProposal:
+    """Wire-stable proposal shape."""
+
+    proposal_id: str
+    route: str
+    unused_tools: List[str]  # alphabetical
+    total_calls_for_route: int
+    observation_window_days: int
+    confidence: float  # derived from sample size
+    created_at: datetime
+    status: ProposalStatus = "pending"
+    review_notes: str = ""
+
+
+def _format_iso(dt: datetime) -> str:
+    return dt.strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def proposal_to_dict(p: ToolTrimProposal) -> Dict[str, Any]:
+    out = asdict(p)
+    out["created_at"] = _format_iso(p.created_at)
+    out["unused_tools"] = list(p.unused_tools)
+    return out
+
+
+def _int_env(name: str, default: int, *, minimum: int = 0) -> int:
+    raw = os.environ.get(name, "").strip()
+    if not raw:
+        return default
+    try:
+        value = int(raw)
+    except ValueError:
+        return default
+    if value < minimum:
+        return default
+    return value
+
+
+def _available_tool_names(rollups: List[RouteToolUsage]) -> Set[str]:
+    """Union-across-routes of tools observed in the window. v1 proxy
+    for "tools registered for any route" — see module docstring for
+    the rationale + future swap-in seam."""
+    out: Set[str] = set()
+    for r in rollups:
+        out.update(r.tools_called)
+    return out
+
+
+def generate_proposals(
+    rollups: List[RouteToolUsage],
+    *,
+    now: datetime,
+    available_tool_names: Iterable[str] | None = None,
+) -> List[ToolTrimProposal]:
+    """For each eligible route, propose dropping tools the route
+    never called. Returns proposals sorted by route (stable
+    ordering for tests + operator triage)."""
+    min_calls = _int_env(MIN_CALLS_ENV, DEFAULT_MIN_CALLS, minimum=2)
+    window_days = _int_env(
+        OBSERVATION_WINDOW_DAYS_ENV,
+        DEFAULT_OBSERVATION_WINDOW_DAYS,
+        minimum=1,
+    )
+
+    if available_tool_names is None:
+        available = _available_tool_names(rollups)
+    else:
+        available = set(available_tool_names)
+
+    out: List[ToolTrimProposal] = []
+    for r in rollups:
+        if r.total_calls < min_calls:
+            continue
+        unused = sorted(available - r.tools_called)
+        if not unused:
+            continue
+        confidence = min(1.0, r.total_calls / (2 * min_calls))
+        out.append(
+            ToolTrimProposal(
+                proposal_id=str(uuid.uuid4()),
+                route=r.route,
+                unused_tools=unused,
+                total_calls_for_route=r.total_calls,
+                observation_window_days=window_days,
+                confidence=round(confidence, 4),
+                created_at=now,
+                status="pending",
+            )
+        )
+    out.sort(key=lambda p: p.route)
+    return out
diff --git a/kora_cli/web_server.py b/kora_cli/web_server.py
index 49d36bd8eef7..a80fef6614fb 100644
--- a/kora_cli/web_server.py
+++ b/kora_cli/web_server.py
@@ -6810,6 +6810,261 @@ async def reject_phrasebook_proposal(
     }
 
 
+# ---------------------------------------------------------------------------
+# Generic promotion-loop endpoints — KR-PROMOTE-LOOPS-COMPLETION-MEGABUCKET
+# ---------------------------------------------------------------------------
+#
+# Three additional loops landed in this bucket (router-tuning,
+# tool-trimming, probe-fix-envelopes). Each follows the phrasebook
+# (#186) pattern but without the loop-specific "approve also writes
+# to live config" step — these are propose-only at v1. The approve
+# endpoint transitions the proposal status + emits ``promotion.approved``;
+# reject does likewise with ``promotion.rejected``. Operator
+# scaffolds the actual config change manually (router prompts, tool
+# manifest, fix_envelopes.py) using the persisted proposal payload
+# as the spec.
+#
+# DRY via :func:`_promotion_loop_pending` etc. — the per-loop GET +
+# POST handlers are 3-line wrappers around the generic helpers.
+#
+# Drift-guard: ``_PROMOTION_STATUS_VALUES`` (defined above for the
+# phrasebook endpoints) is shared.
+
+
+def _promotion_loop_pending(loop_name: str) -> Dict[str, Any]:
+    from kora_cli.promote._shared.proposal_store import list_by_status
+
+    proposals = list_by_status(loop_name=loop_name, status="pending")
+    # Highest-confidence first when payloads carry that field;
+    # falls back to filesystem-name order otherwise.
+    proposals.sort(
+        key=lambda p: (
+            -float(p.get("confidence") or 0.0),
+            -int(p.get("cluster_size") or 0),
+        )
+    )
+    return {
+        "proposals": proposals,
+        "status_values": list(_PROMOTION_STATUS_VALUES),
+        "loop_name": loop_name,
+    }
+
+
+def _promotion_loop_transition(
+    *,
+    loop_name: str,
+    proposal_id: str,
+    new_status: str,
+    audit_seam: str,
+    payload: Optional[Dict[str, Any]],
+) -> Any:
+    """Shared transition helper for the 3 new loops. Validates
+    pending state, mutates payload review_notes if present, emits
+    audit row, returns the canonical response shape."""
+    from kora_cli.audit import emit_audit
+    from kora_cli.promote._shared.proposal_store import (
+        ProposalNotFound,
+        load,
+        transition,
+    )
+
+    review_notes = ""
+    if isinstance(payload, dict):
+        notes_raw = payload.get("review_notes")
+        if isinstance(notes_raw, str):
+            review_notes = notes_raw
+
+    try:
+        current_status, _ = load(loop_name=loop_name, proposal_id=proposal_id)
+    except ProposalNotFound:
+        return JSONResponse(
+            status_code=404,
+            content={"error": "proposal_not_found", "proposal_id": proposal_id},
+        )
+    if current_status != "pending":
+        return JSONResponse(
+            status_code=409,
+            content={
+                "error": "proposal_not_pending",
+                "proposal_id": proposal_id,
+                "current_status": current_status,
+            },
+        )
+
+    def _mutate(p: Dict[str, Any]) -> None:
+        p["status"] = new_status
+        if review_notes:
+            p["review_notes"] = review_notes
+
+    _, updated = transition(
+        loop_name=loop_name,
+        proposal_id=proposal_id,
+        new_status=new_status,
+        payload_mutator=_mutate,
+    )
+
+    try:
+        emit_audit(
+            seam=audit_seam,
+            details=updated,
+            caller_session_id=(
+                f"promotion:{loop_name}:{proposal_id}"
+            ),
+            source="reasoning",
+        )
+    except Exception as exc:
+        logger.warning(
+            "[kora.promote] %s emit raised %r — transition persisted; "
+            "audit row missing",
+            audit_seam,
+            exc,
+        )
+
+    return {
+        "proposal_id": proposal_id,
+        "status": new_status,
+        "review_notes": review_notes,
+    }
+
+
+# --- Router-tuning ---------------------------------------------------------
+
+
+@app.get("/api/promotions/router-tuning/pending")
+async def list_pending_router_tuning_proposals() -> Dict[str, Any]:
+    """Return pending router-tuning proposals.
+
+    Payload shape per ``kora_cli.promote.router_tuning.proposer``:
+    proposal_id / route / calls_count / escalation_count /
+    escalation_rate / recommendation_kind / rationale / confidence /
+    created_at / status.
+    """
+    return _promotion_loop_pending("router_tuning")
+
+
+@app.post("/api/promotions/router-tuning/{proposal_id}/approve")
+async def approve_router_tuning_proposal(
+    proposal_id: str, payload: Optional[Dict[str, Any]] = None
+) -> Any:
+    """Approve a router-tuning proposal. Transitions the proposal
+    + emits ``promotion.approved``; does NOT mutate router config —
+    operator scaffolds the trigger-pattern change manually from the
+    proposal rationale."""
+    return _promotion_loop_transition(
+        loop_name="router_tuning",
+        proposal_id=proposal_id,
+        new_status="approved",
+        audit_seam="promotion.approved",
+        payload=payload,
+    )
+
+
+@app.post("/api/promotions/router-tuning/{proposal_id}/reject")
+async def reject_router_tuning_proposal(
+    proposal_id: str, payload: Optional[Dict[str, Any]] = None
+) -> Any:
+    return _promotion_loop_transition(
+        loop_name="router_tuning",
+        proposal_id=proposal_id,
+        new_status="rejected",
+        audit_seam="promotion.rejected",
+        payload=payload,
+    )
+
+
+# --- Tool-trimming ---------------------------------------------------------
+
+
+@app.get("/api/promotions/tool-trimming/pending")
+async def list_pending_tool_trimming_proposals() -> Dict[str, Any]:
+    """Return pending tool-trimming proposals.
+
+    Payload per ``kora_cli.promote.tool_trimming.proposer``:
+    proposal_id / route / unused_tools / total_calls_for_route /
+    observation_window_days / confidence / created_at / status.
+    """
+    return _promotion_loop_pending("tool_trimming")
+
+
+@app.post("/api/promotions/tool-trimming/{proposal_id}/approve")
+async def approve_tool_trimming_proposal(
+    proposal_id: str, payload: Optional[Dict[str, Any]] = None
+) -> Any:
+    """Approve a tool-trim proposal. v1 transitions status + emits
+    audit only — actual drop-list enforcement lands in the future
+    KR-PLUGIN-TOOL-DESC-TRIM bucket which reads the approved
+    proposals."""
+    return _promotion_loop_transition(
+        loop_name="tool_trimming",
+        proposal_id=proposal_id,
+        new_status="approved",
+        audit_seam="promotion.approved",
+        payload=payload,
+    )
+
+
+@app.post("/api/promotions/tool-trimming/{proposal_id}/reject")
+async def reject_tool_trimming_proposal(
+    proposal_id: str, payload: Optional[Dict[str, Any]] = None
+) -> Any:
+    return _promotion_loop_transition(
+        loop_name="tool_trimming",
+        proposal_id=proposal_id,
+        new_status="rejected",
+        audit_seam="promotion.rejected",
+        payload=payload,
+    )
+
+
+# --- Probe-fix-envelopes ---------------------------------------------------
+
+
+@app.get("/api/promotions/probe-envelopes/pending")
+async def list_pending_probe_envelope_proposals() -> Dict[str, Any]:
+    """Return pending probe-fix-envelope proposals.
+
+    Payload per ``kora_cli.promote.probe_fix_envelopes.proposer``:
+    proposal_id / probe / issue_category / fix_name_suggestion /
+    cluster_size / recurring_recommendation_text /
+    blast_radius_summary / confidence / created_at / status.
+
+    HIGH-RISK loop — see module docstring. Operator manually
+    scaffolds approved envelopes into ``probes/fix_envelopes.py``
+    using the persisted payload as the spec.
+    """
+    return _promotion_loop_pending("probe_fix_envelopes")
+
+
+@app.post("/api/promotions/probe-envelopes/{proposal_id}/approve")
+async def approve_probe_envelope_proposal(
+    proposal_id: str, payload: Optional[Dict[str, Any]] = None
+) -> Any:
+    """Approve a probe-fix-envelope proposal. v1 transitions status
+    + emits audit only — Kora's ``fix_envelopes.py`` MUST be edited
+    by hand. The approved/ proposal file is the audit trail for
+    when the manual scaffold lands."""
+    return _promotion_loop_transition(
+        loop_name="probe_fix_envelopes",
+        proposal_id=proposal_id,
+        new_status="approved",
+        audit_seam="promotion.approved",
+        payload=payload,
+    )
+
+
+@app.post("/api/promotions/probe-envelopes/{proposal_id}/reject")
+async def reject_probe_envelope_proposal(
+    proposal_id: str, payload: Optional[Dict[str, Any]] = None
+) -> Any:
+    return _promotion_loop_transition(
+        loop_name="probe_fix_envelopes",
+        proposal_id=proposal_id,
+        new_status="rejected",
+        audit_seam="promotion.rejected",
+        payload=payload,
+    )
+
+
 # ---------------------------------------------------------------------------
 # Email-intent audit lens (KR-FE-EMAIL-INTENT-LOG-PANEL)
 # ---------------------------------------------------------------------------
diff --git a/tests/kora_cli/audit/test_jsonl_sink.py b/tests/kora_cli/audit/test_jsonl_sink.py
index 9909bcc6cd83..106c8460a685 100644
--- a/tests/kora_cli/audit/test_jsonl_sink.py
+++ b/tests/kora_cli/audit/test_jsonl_sink.py
@@ -30,14 +30,30 @@
 from pydantic import ValidationError
 
 from kora_cli.audit.jsonl_sink import (
-    LOG_PATH_ENV,
     AUDIT_LOG_FILENAME,
+    BATCH_SIZE_ENV,
+    FLUSH_INTERVAL_ENV,
+    LOG_PATH_ENV,
     AuditEntry,
+    _reset_batching_for_tests,
     _resolve_log_path,
     emit_audit,
+    flush_for_tests,
 )
 
 
+# KR-CHEAP-AUDIT-BATCHING — most legacy tests assume sync per-emit
+# write semantics (they emit then immediately read the file). Force
+# BATCH_SIZE=0 to preserve that for the legacy suite; the dedicated
+# batching-behavior tests below opt back into batching explicitly.
+@pytest.fixture(autouse=True)
+def _disable_batching_by_default(monkeypatch):
+    monkeypatch.setenv(BATCH_SIZE_ENV, "0")
+    _reset_batching_for_tests()
+    yield
+    _reset_batching_for_tests()
+
+
 # ---------------------------------------------------------------------------
 # AuditEntry shape
 # ---------------------------------------------------------------------------
@@ -537,3 +553,165 @@ def test_dual_write_both_surfaces_fire(tmp_path, caplog, monkeypatch):
     [entry] = _read_jsonl_lines(path)
     assert entry["seam"] == "reasoning.tool_called"
     assert entry["details"]["tool_name"] == "kora__get_operational_state"
+
+
+# ===========================================================================
+# KR-CHEAP-AUDIT-BATCHING (R3-4 #9) — batched writer behavior
+# ===========================================================================
+
+
+@pytest.fixture
+def batching_enabled(monkeypatch):
+    """Opt into batching for the per-test scope. Tiny batch size (3)
+    so the size-triggered flush is easy to exercise; tiny interval
+    (0.1s) so the time-triggered flush completes within the test's
+    timeout window."""
+    monkeypatch.setenv(BATCH_SIZE_ENV, "3")
+    monkeypatch.setenv(FLUSH_INTERVAL_ENV, "0.1")
+    _reset_batching_for_tests()
+    yield
+    _reset_batching_for_tests()
+
+
+def test_batching_size_trigger_flushes_when_threshold_hit(
+    batching_enabled, tmp_path
+):
+    """Hitting BATCH_SIZE events flushes synchronously inside the
+    triggering emit_audit call — no need to wait for the interval."""
+    path = tmp_path / "audit.jsonl"
+    # 2 emits queue; on the 3rd, batch_size=3 triggers immediate flush.
+    for i in range(3):
+        emit_audit(
+            seam="reasoning.tool_called",
+            details={"tool_name": f"tool_{i}", "tool_status": "ok"},
+            source="reasoning",
+            log_path=path,
+        )
+    entries = _read_jsonl_lines(path)
+    assert len(entries) == 3
+    assert [e["details"]["tool_name"] for e in entries] == [
+        "tool_0",
+        "tool_1",
+        "tool_2",
+    ]
+
+
+def test_batching_below_size_threshold_does_not_write_immediately(
+    batching_enabled, tmp_path
+):
+    """Emits below the batch_size threshold are queued — file stays
+    empty until either the interval-tick or an explicit flush."""
+    path = tmp_path / "audit.jsonl"
+    emit_audit(
+        seam="reasoning.tool_called",
+        details={"tool_name": "tool_a", "tool_status": "ok"},
+        source="reasoning",
+        log_path=path,
+    )
+    emit_audit(
+        seam="reasoning.tool_called",
+        details={"tool_name": "tool_b", "tool_status": "ok"},
+        source="reasoning",
+        log_path=path,
+    )
+    # Below batch_size=3 — file should not exist yet.
+    assert not path.exists() or _read_jsonl_lines(path) == []
+    # Synchronous test-only drain proves the queue held both rows.
+    flushed = flush_for_tests()
+    assert flushed == 2
+    entries = _read_jsonl_lines(path)
+    assert len(entries) == 2
+
+
+def test_batching_time_trigger_flushes_after_interval(
+    batching_enabled, tmp_path
+):
+    """Below-threshold queue gets drained by the background-thread
+    interval tick. Allows up to ~0.5s for the thread to fire."""
+    import time
+
+    path = tmp_path / "audit.jsonl"
+    emit_audit(
+        seam="reasoning.tool_called",
+        details={"tool_name": "tool_x", "tool_status": "ok"},
+        source="reasoning",
+        log_path=path,
+    )
+    # Interval is 0.1s; allow up to 0.6s for the thread to fire.
+    deadline = time.monotonic() + 0.6
+    entries: List[Dict[str, Any]] = []
+    while time.monotonic() < deadline:
+        entries = _read_jsonl_lines(path)
+        if entries:
+            break
+        time.sleep(0.05)
+    assert len(entries) == 1
+    assert entries[0]["details"]["tool_name"] == "tool_x"
+
+
+def test_batching_groups_entries_per_path(batching_enabled, tmp_path):
+    """When emits target multiple paths in the same batch, each
+    file is opened once + receives only its own rows."""
+    a = tmp_path / "a.jsonl"
+    b = tmp_path / "b.jsonl"
+    # 3 events total → triggers the size-based flush. 2 to A + 1 to B.
+    emit_audit(
+        seam="reasoning.tool_called",
+        details={"tool_name": "a1", "tool_status": "ok"},
+        source="reasoning",
+        log_path=a,
+    )
+    emit_audit(
+        seam="reasoning.tool_called",
+        details={"tool_name": "b1", "tool_status": "ok"},
+        source="reasoning",
+        log_path=b,
+    )
+    emit_audit(
+        seam="reasoning.tool_called",
+        details={"tool_name": "a2", "tool_status": "ok"},
+        source="reasoning",
+        log_path=a,
+    )
+    rows_a = _read_jsonl_lines(a)
+    rows_b = _read_jsonl_lines(b)
+    assert [r["details"]["tool_name"] for r in rows_a] == ["a1", "a2"]
+    assert [r["details"]["tool_name"] for r in rows_b] == ["b1"]
+
+
+def test_batching_shutdown_drain_flushes_remaining(
+    batching_enabled, tmp_path
+):
+    """The test-reset path mirrors what atexit does in production —
+    signals the thread to stop + drains the queue."""
+    path = tmp_path / "audit.jsonl"
+    emit_audit(
+        seam="reasoning.tool_called",
+        details={"tool_name": "in_flight", "tool_status": "ok"},
+        source="reasoning",
+        log_path=path,
+    )
+    # Trigger the reset (which calls flusher_stop.set + waits for the
+    # thread; the thread's final loop iteration drains the queue
+    # before exiting).
+    _reset_batching_for_tests()
+    entries = _read_jsonl_lines(path)
+    assert len(entries) == 1
+    assert entries[0]["details"]["tool_name"] == "in_flight"
+
+
+def test_batching_disabled_writes_synchronously(tmp_path, monkeypatch):
+    """BATCH_SIZE=0 (explicit opt-out, default in legacy tests):
+    emits write to the file before emit_audit returns."""
+    monkeypatch.setenv(BATCH_SIZE_ENV, "0")
+    _reset_batching_for_tests()
+    path = tmp_path / "audit.jsonl"
+    emit_audit(
+        seam="reasoning.tool_called",
+        details={"tool_name": "sync_only", "tool_status": "ok"},
+        source="reasoning",
+        log_path=path,
+    )
+    # File should exist immediately, no flush call needed.
+    [entry] = _read_jsonl_lines(path)
+    assert entry["details"]["tool_name"] == "sync_only"
diff --git a/tests/kora_cli/conftest.py b/tests/kora_cli/conftest.py
index 3afaed552e21..3a260fd014de 100644
--- a/tests/kora_cli/conftest.py
+++ b/tests/kora_cli/conftest.py
@@ -44,3 +44,32 @@ def _suppress_concurrent_hermes_gate(request, monkeypatch):
     monkeypatch.setattr(
         _cli_main, "_detect_concurrent_hermes_instances", lambda *_a, **_k: []
     )
+
+
+@pytest.fixture(autouse=True)
+def _audit_sync_writes_by_default(request, monkeypatch):
+    """KR-CHEAP-AUDIT-BATCHING — default tests to sync writes.
+
+    Production default is batching ON (BATCH_SIZE=100, 5s interval)
+    but most test suites read the audit JSONL immediately after the
+    emit call and assume sync semantics. Force BATCH_SIZE=0 globally;
+    the dedicated batching-behavior tests in
+    ``tests/kora_cli/audit/test_jsonl_sink.py`` opt back in with their
+    own ``batching_enabled`` fixture.
+
+    Tests that want to exercise the production batched path opt out
+    of this default with ``@pytest.mark.audit_batching_default``.
+    """
+    if request.node.get_closest_marker("audit_batching_default"):
+        return
+    try:
+        from kora_cli.audit.jsonl_sink import (
+            BATCH_SIZE_ENV,
+            _reset_batching_for_tests,
+        )
+    except Exception:
+        return
+    monkeypatch.setenv(BATCH_SIZE_ENV, "0")
+    _reset_batching_for_tests()
+    yield
+    _reset_batching_for_tests()
diff --git a/tests/kora_cli/probes/test_wake_consumer.py b/tests/kora_cli/probes/test_wake_consumer.py
index fff26d243079..44a40a02b199 100644
--- a/tests/kora_cli/probes/test_wake_consumer.py
+++ b/tests/kora_cli/probes/test_wake_consumer.py
@@ -66,9 +66,27 @@
 
 @pytest.fixture(autouse=True)
 def _isolate_env(monkeypatch):
+    from kora_cli.audit.jsonl_sink import (
+        BATCH_SIZE_ENV,
+        _reset_batching_for_tests,
+    )
+
     monkeypatch.setenv(JOSHUA_SLACK_USER_ID_ENV, _JOSHUA_USER_ID)
     monkeypatch.delenv(DEBOUNCE_SECONDS_ENV, raising=False)
     monkeypatch.delenv(BYPASS_CRITICAL_ENV, raising=False)
+    # KR-PROBE-DEBOUNCE — default-behavior tests in this file predate
+    # the consecutive-failure upgrade. Force required=1 here so
+    # single-tick scenarios still dispatch; dedicated consecutive-
+    # buffering tests set their own env explicitly.
+    monkeypatch.setenv("KORA_PROBE_DEBOUNCE_CONSECUTIVE_REQUIRED", "1")
+    # KR-CHEAP-AUDIT-BATCHING — these tests read the audit JSONL
+    # immediately after the consumer emits + assume sync semantics.
+    # Force per-emit writes to keep that contract; dedicated
+    # batching tests live in tests/kora_cli/audit/test_jsonl_sink.py.
+    monkeypatch.setenv(BATCH_SIZE_ENV, "0")
+    _reset_batching_for_tests()
+    yield
+    _reset_batching_for_tests()
 
 
 def _make_event(
@@ -906,3 +924,100 @@ def _boom(*args, **kwargs):
         )
         is None
     )
+
+
+# ===========================================================================
+# KR-PROBE-DEBOUNCE — consecutive-failure buffering upgrade
+# ===========================================================================
+
+
+@pytest.mark.asyncio
+async def test_consecutive_first_failure_buffered_not_dispatched(monkeypatch):
+    """With required=2, the first wake_requested event for a
+    (probe, category) pair is held in the buffer rather than
+    dispatched. Prevents single-tick flakes from waking Kora."""
+    monkeypatch.setenv("KORA_PROBE_DEBOUNCE_CONSECUTIVE_REQUIRED", "2")
+    # Severity warning so the critical-bypass path doesn't apply.
+    consumer = _make_consumer(engine=_make_engine(), slack=_make_slack())
+    out = await consumer.consume_wake_event(
+        _make_event(severity="warning")
+    )
+    assert out.dispatched is False
+    assert out.buffered_skipped is True
+    assert out.buffered_consecutive_count == 1
+    assert consumer.consecutive_buffer_size == 1
+
+
+@pytest.mark.asyncio
+async def test_consecutive_second_failure_dispatches(monkeypatch):
+    """The second event within the window pushes the buffer to the
+    threshold → dispatch fires."""
+    monkeypatch.setenv("KORA_PROBE_DEBOUNCE_CONSECUTIVE_REQUIRED", "2")
+    engine = _make_engine()
+    slack = _make_slack()
+    consumer = _make_consumer(engine=engine, slack=slack)
+    out1 = await consumer.consume_wake_event(
+        _make_event(severity="warning")
+    )
+    assert out1.dispatched is False
+    out2 = await consumer.consume_wake_event(
+        _make_event(severity="warning")
+    )
+    assert out2.dispatched is True
+    assert out2.buffered_skipped is False
+    # Buffer cleared after dispatch (post-dispatch flat-window
+    # debounce takes over).
+    assert consumer.consecutive_buffer_size == 0
+
+
+@pytest.mark.asyncio
+async def test_consecutive_critical_bypass_dispatches_first(monkeypatch):
+    """Critical-severity wakes with the bypass env truthy skip the
+    consecutive-failure buffer + dispatch on first event."""
+    monkeypatch.setenv("KORA_PROBE_DEBOUNCE_CONSECUTIVE_REQUIRED", "5")
+    monkeypatch.setenv(BYPASS_CRITICAL_ENV, "true")
+    consumer = _make_consumer(engine=_make_engine(), slack=_make_slack())
+    out = await consumer.consume_wake_event(
+        _make_event(severity="critical")
+    )
+    assert out.dispatched is True
+
+
+@pytest.mark.asyncio
+async def test_consecutive_different_pairs_independent(monkeypatch):
+    """Distinct (probe, category) pairs accumulate independently —
+    one buffered, the other unrelated."""
+    monkeypatch.setenv("KORA_PROBE_DEBOUNCE_CONSECUTIVE_REQUIRED", "2")
+    consumer = _make_consumer(engine=_make_engine(), slack=_make_slack())
+    await consumer.consume_wake_event(
+        _make_event(probe="fly", category="machine_down", severity="warning")
+    )
+    out = await consumer.consume_wake_event(
+        _make_event(probe="vercel", category="deploy_fail", severity="warning")
+    )
+    assert out.dispatched is False
+    assert out.buffered_consecutive_count == 1
+    assert consumer.consecutive_buffer_size == 2
+
+
+@pytest.mark.asyncio
+async def test_consecutive_required_one_preserves_legacy_behavior(monkeypatch):
+    """required=1 disables buffering — single failure dispatches."""
+    monkeypatch.setenv("KORA_PROBE_DEBOUNCE_CONSECUTIVE_REQUIRED", "1")
+    consumer = _make_consumer(engine=_make_engine(), slack=_make_slack())
+    out = await consumer.consume_wake_event(
+        _make_event(severity="warning")
+    )
+    assert out.dispatched is True
+
+
+@pytest.mark.asyncio
+async def test_consecutive_reset_debounce_state_clears_buffer(monkeypatch):
+    """``reset_debounce_state`` clears the consecutive buffer too —
+    listener shutdown restart should see a clean slate."""
+    monkeypatch.setenv("KORA_PROBE_DEBOUNCE_CONSECUTIVE_REQUIRED", "2")
+    consumer = _make_consumer(engine=_make_engine(), slack=_make_slack())
+    await consumer.consume_wake_event(_make_event(severity="warning"))
+    assert consumer.consecutive_buffer_size == 1
+    consumer.reset_debounce_state()
+    assert consumer.consecutive_buffer_size == 0
diff --git a/tests/kora_cli/promote/_shared/__init__.py b/tests/kora_cli/promote/_shared/__init__.py
new file mode 100644
index 000000000000..e69de29bb2d1
diff --git a/tests/kora_cli/promote/_shared/test_proposal_store.py b/tests/kora_cli/promote/_shared/test_proposal_store.py
new file mode 100644
index 000000000000..f92b1a38753c
--- /dev/null
+++ b/tests/kora_cli/promote/_shared/test_proposal_store.py
@@ -0,0 +1,107 @@
+"""Tests for kora_cli.promote._shared.proposal_store."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from kora_cli.promote._shared.proposal_store import (
+    ProposalNotFound,
+    expire_older_than,
+    list_by_status,
+    load,
+    save_pending,
+    transition,
+)
+
+
+@pytest.fixture(autouse=True)
+def _isolate(tmp_path, monkeypatch):
+    monkeypatch.setenv("KORA_HOME", str(tmp_path))
+    monkeypatch.setenv("KORA_PROMOTIONS_DIR", str(tmp_path / "promotions"))
+    return tmp_path
+
+
+def _payload(proposal_id: str = "prop-1", **extra) -> dict:
+    base = {
+        "proposal_id": proposal_id,
+        "status": "pending",
+        "created_at": "2026-05-23T12:00:00Z",
+    }
+    base.update(extra)
+    return base
+
+
+def test_save_pending_writes_atomically(tmp_path):
+    save_pending(loop_name="foo", proposal_id="p1", payload=_payload("p1"))
+    target = tmp_path / "promotions" / "foo" / "pending" / "p1.json"
+    assert target.is_file()
+    written = json.loads(target.read_text())
+    assert written["proposal_id"] == "p1"
+
+
+def test_list_by_status_returns_dicts():
+    save_pending(loop_name="foo", proposal_id="p1", payload=_payload("p1"))
+    save_pending(loop_name="foo", proposal_id="p2", payload=_payload("p2"))
+    items = list_by_status(loop_name="foo", status="pending")
+    assert sorted(p["proposal_id"] for p in items) == ["p1", "p2"]
+
+
+def test_load_returns_status_and_payload():
+    save_pending(loop_name="foo", proposal_id="p1", payload=_payload("p1"))
+    status, payload = load(loop_name="foo", proposal_id="p1")
+    assert status == "pending"
+    assert payload["proposal_id"] == "p1"
+
+
+def test_load_raises_when_missing():
+    with pytest.raises(ProposalNotFound):
+        load(loop_name="foo", proposal_id="ghost")
+
+
+def test_transition_moves_file_between_dirs(tmp_path):
+    save_pending(loop_name="foo", proposal_id="p1", payload=_payload("p1"))
+    old_status, payload = transition(
+        loop_name="foo",
+        proposal_id="p1",
+        new_status="approved",
+        payload_mutator=lambda p: p.update({"review_notes": "shipping it"}),
+    )
+    assert old_status == "pending"
+    assert payload["review_notes"] == "shipping it"
+    # Old location gone, new location populated.
+    assert not (tmp_path / "promotions" / "foo" / "pending" / "p1.json").is_file()
+    assert (tmp_path / "promotions" / "foo" / "approved" / "p1.json").is_file()
+
+
+def test_transition_rejects_unknown_status():
+    save_pending(loop_name="foo", proposal_id="p1", payload=_payload("p1"))
+    with pytest.raises(ValueError):
+        transition(
+            loop_name="foo", proposal_id="p1", new_status="archived"
+        )
+
+
+def test_expire_older_than_moves_old_pending():
+    save_pending(
+        loop_name="foo",
+        proposal_id="old",
+        payload=_payload("old", created_at="2020-01-01T00:00:00Z"),
+    )
+    save_pending(loop_name="foo", proposal_id="new", payload=_payload("new"))
+    moved = expire_older_than(loop_name="foo", days=7)
+    assert moved == 1
+    # old now in expired/, new stays in pending/
+    expired = list_by_status(loop_name="foo", status="expired")
+    pending = list_by_status(loop_name="foo", status="pending")
+    assert [p["proposal_id"] for p in expired] == ["old"]
+    assert [p["proposal_id"] for p in pending] == ["new"]
+
+
+def test_per_loop_dirs_dont_collide():
+    save_pending(loop_name="alpha", proposal_id="x", payload=_payload("x"))
+    save_pending(loop_name="beta", proposal_id="x", payload=_payload("x"))
+    assert len(list_by_status(loop_name="alpha", status="pending")) == 1
+    assert len(list_by_status(loop_name="beta", status="pending")) == 1
diff --git a/tests/kora_cli/promote/probe_fix_envelopes/__init__.py b/tests/kora_cli/promote/probe_fix_envelopes/__init__.py
new file mode 100644
index 000000000000..e69de29bb2d1
diff --git a/tests/kora_cli/promote/probe_fix_envelopes/test_probe_fix_envelopes.py b/tests/kora_cli/promote/probe_fix_envelopes/test_probe_fix_envelopes.py
new file mode 100644
index 000000000000..ec0b0622ed31
--- /dev/null
+++ b/tests/kora_cli/promote/probe_fix_envelopes/test_probe_fix_envelopes.py
@@ -0,0 +1,214 @@
+"""Tests for kora_cli.promote.probe_fix_envelopes."""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import pytest
+
+from kora_cli.audit.jsonl_sink import BATCH_SIZE_ENV, _reset_batching_for_tests
+from kora_cli.promote.probe_fix_envelopes.observer import (
+    InvestigationObservation,
+    collect_recent_investigations,
+)
+from kora_cli.promote.probe_fix_envelopes.plugin import (
+    ENABLED_ENV,
+    run_probe_fix_envelopes_cycle,
+)
+from kora_cli.promote.probe_fix_envelopes.proposer import (
+    MIN_CLUSTER_SIZE_ENV,
+    generate_proposals,
+)
+
+
+@pytest.fixture(autouse=True)
+def _isolate(tmp_path, monkeypatch):
+    monkeypatch.setenv("KORA_HOME", str(tmp_path))
+    monkeypatch.setenv("KORA_PROMOTIONS_DIR", str(tmp_path / "promotions"))
+    monkeypatch.setenv(
+        "KORA_AUDIT_LOG_PATH", str(tmp_path / "kora_audit_log.jsonl")
+    )
+    monkeypatch.setenv(BATCH_SIZE_ENV, "0")
+    _reset_batching_for_tests()
+    yield
+    _reset_batching_for_tests()
+
+
+def _write_audit(tmp_path: Path, entries: list) -> None:
+    path = tmp_path / "kora_audit_log.jsonl"
+    path.write_text(
+        "\n".join(json.dumps(e) for e in entries) + "\n", encoding="utf-8"
+    )
+
+
+def _investigation_entry(
+    *,
+    probe: str = "fly",
+    issue_category: str = "machine_down",
+    summary: str = "Restart the fly machine to recover.",
+    autofix_attempted: bool = False,
+    caller_session_id: str = "probe:fly:machine_down",
+    emitted_at: datetime | None = None,
+) -> dict:
+    if emitted_at is None:
+        emitted_at = datetime.now(timezone.utc) - timedelta(hours=1)
+    return {
+        "emitted_at": emitted_at.isoformat(),
+        "seam": "probe.investigation_completed",
+        "details": {
+            "probe": probe,
+            "issue_category": issue_category,
+            "severity": "warning",
+            "investigation_summary_text": summary,
+            "autofix_attempted": autofix_attempted,
+        },
+        "caller_session_id": caller_session_id,
+        "source": "reasoning",
+    }
+
+
+# ---------------------------------------------------------------------------
+# Observer
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_observer_skips_autofix_attempted(tmp_path):
+    _write_audit(
+        tmp_path,
+        [
+            _investigation_entry(autofix_attempted=False),
+            _investigation_entry(autofix_attempted=True),
+        ],
+    )
+    out = await collect_recent_investigations(
+        since=datetime.now(timezone.utc) - timedelta(days=1)
+    )
+    assert len(out) == 1
+
+
+@pytest.mark.asyncio
+async def test_observer_skips_empty_summaries(tmp_path):
+    _write_audit(
+        tmp_path,
+        [
+            _investigation_entry(summary=""),
+            _investigation_entry(summary="real text"),
+        ],
+    )
+    out = await collect_recent_investigations(
+        since=datetime.now(timezone.utc) - timedelta(days=1)
+    )
+    assert len(out) == 1
+
+
+# ---------------------------------------------------------------------------
+# Proposer
+# ---------------------------------------------------------------------------
+
+
+def _obs(
+    probe: str = "fly",
+    cat: str = "machine_down",
+    summary: str = "Restart the fly machine.",
+    csid: str = "probe:fly:machine_down:1",
+) -> InvestigationObservation:
+    return InvestigationObservation(
+        probe=probe,
+        issue_category=cat,
+        severity="warning",
+        investigation_summary_text=summary,
+        caller_session_id=csid,
+        timestamp=datetime.now(timezone.utc),
+    )
+
+
+def test_proposer_clusters_by_probe_and_category(monkeypatch):
+    monkeypatch.setenv(MIN_CLUSTER_SIZE_ENV, "2")
+    obs = [
+        _obs(probe="fly", cat="machine_down", csid=f"c{i}") for i in range(3)
+    ] + [
+        _obs(probe="vercel", cat="deploy_fail", csid=f"v{i}") for i in range(3)
+    ]
+    out = generate_proposals(obs, now=datetime.now(timezone.utc))
+    by_probe = {p.probe: p for p in out}
+    assert set(by_probe.keys()) == {"fly", "vercel"}
+    assert by_probe["fly"].fix_name_suggestion == "proposed_fly_machine_down"
+    assert by_probe["fly"].cluster_size == 3
+
+
+def test_proposer_skips_under_min_cluster(monkeypatch):
+    monkeypatch.setenv(MIN_CLUSTER_SIZE_ENV, "3")
+    obs = [_obs(probe="fly", cat="machine_down", csid="c1")]
+    out = generate_proposals(obs, now=datetime.now(timezone.utc))
+    assert out == []
+
+
+def test_proposer_default_blast_radius_is_conservative(monkeypatch):
+    monkeypatch.setenv(MIN_CLUSTER_SIZE_ENV, "2")
+    obs = [_obs(csid=f"c{i}") for i in range(2)]
+    out = generate_proposals(obs, now=datetime.now(timezone.utc))
+    assert "operator must review" in out[0].blast_radius_summary
+
+
+def test_proposer_surfaces_recurring_text(monkeypatch):
+    monkeypatch.setenv(MIN_CLUSTER_SIZE_ENV, "2")
+    obs = [
+        _obs(summary="Restart the fly machine.", csid=f"c{i}") for i in range(3)
+    ] + [_obs(summary="something else entirely", csid="cother")]
+    out = generate_proposals(obs, now=datetime.now(timezone.utc))
+    assert out[0].recurring_recommendation_text.startswith("Restart")
+
+
+# ---------------------------------------------------------------------------
+# Cycle
+# ---------------------------------------------------------------------------
+
+
+def _read_audit(tmp_path) -> list:
+    path = tmp_path / "kora_audit_log.jsonl"
+    if not path.is_file():
+        return []
+    return [
+        json.loads(line)
+        for line in path.read_text(encoding="utf-8").splitlines()
+        if line.strip()
+    ]
+
+
+@pytest.mark.asyncio
+async def test_cycle_emits_audit_with_correct_seam(tmp_path, monkeypatch):
+    monkeypatch.setenv(MIN_CLUSTER_SIZE_ENV, "2")
+    _write_audit(
+        tmp_path,
+        [
+            _investigation_entry(
+                summary="Restart the fly machine.",
+                caller_session_id=f"probe:fly:machine_down:{i}",
+            )
+            for i in range(3)
+        ],
+    )
+    summary = await run_probe_fix_envelopes_cycle()
+    assert summary["proposals_generated"] == 1
+    # HARDCODED auto_apply_mode = False per safety posture.
+    assert summary["auto_apply_mode"] is False
+
+    rows = _read_audit(tmp_path)
+    promo = [
+        r
+        for r in rows
+        if r["seam"] == "promotion.probe_envelope_action_proposed"
+    ]
+    assert len(promo) == 1
+    assert promo[0]["details"]["probe"] == "fly"
+    assert promo[0]["details"]["fix_name_suggestion"] == "proposed_fly_machine_down"
+
+
+@pytest.mark.asyncio
+async def test_cycle_disabled_short_circuits(tmp_path, monkeypatch):
+    monkeypatch.setenv(ENABLED_ENV, "false")
+    summary = await run_probe_fix_envelopes_cycle()
+    assert summary["enabled"] is False
diff --git a/tests/kora_cli/promote/router_tuning/__init__.py b/tests/kora_cli/promote/router_tuning/__init__.py
new file mode 100644
index 000000000000..e69de29bb2d1
diff --git a/tests/kora_cli/promote/router_tuning/test_router_tuning.py b/tests/kora_cli/promote/router_tuning/test_router_tuning.py
new file mode 100644
index 000000000000..1419ae141516
--- /dev/null
+++ b/tests/kora_cli/promote/router_tuning/test_router_tuning.py
@@ -0,0 +1,221 @@
+"""Tests for kora_cli.promote.router_tuning (observer + proposer + cycle)."""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+from unittest.mock import MagicMock
+
+import pytest
+
+from kora_cli.audit.jsonl_sink import BATCH_SIZE_ENV, _reset_batching_for_tests
+from kora_cli.promote.router_tuning.observer import (
+    RouteEscalationRollup,
+    collect_route_rollups,
+)
+from kora_cli.promote.router_tuning.plugin import (
+    ENABLED_ENV,
+    run_router_tuning_cycle,
+)
+from kora_cli.promote.router_tuning.proposer import (
+    DEFAULT_MIN_CALLS,
+    DEFAULT_TIGHTEN_THRESHOLD,
+    MIN_CALLS_ENV,
+    TIGHTEN_THRESHOLD_ENV,
+    generate_proposals,
+)
+
+
+@pytest.fixture(autouse=True)
+def _isolate(tmp_path, monkeypatch):
+    monkeypatch.setenv("KORA_HOME", str(tmp_path))
+    monkeypatch.setenv("KORA_PROMOTIONS_DIR", str(tmp_path / "promotions"))
+    monkeypatch.setenv(
+        "KORA_AUDIT_LOG_PATH", str(tmp_path / "kora_audit_log.jsonl")
+    )
+    # Force sync audit writes so tests can assert the row landed
+    # without waiting on the background flusher.
+    monkeypatch.setenv(BATCH_SIZE_ENV, "0")
+    _reset_batching_for_tests()
+    yield
+    _reset_batching_for_tests()
+
+
+def _rollup(
+    route: str = "slack_dm",
+    calls: int = 50,
+    escs: int = 20,
+    cost: float = 0.10,
+) -> RouteEscalationRollup:
+    return RouteEscalationRollup(
+        route=route,
+        calls_count=calls,
+        escalation_count=escs,
+        escalation_rate=escs / calls if calls else 0.0,
+        cost_estimate_usd_total=cost,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Observer
+# ---------------------------------------------------------------------------
+
+
+def test_observer_reads_telemetry_singleton(monkeypatch):
+    """Mock the telemetry singleton + verify the rolling_24h projection."""
+    fake = MagicMock()
+    fake.snapshot.return_value = {
+        "rolling_24h": {
+            "slack_dm": {
+                "calls_count": 30,
+                "escalation_count": 12,
+                "cost_estimate_usd_total": 0.40,
+            },
+            "email_inbound": {
+                "calls_count": 0,
+                "escalation_count": 0,
+                "cost_estimate_usd_total": 0.0,
+            },
+        },
+        "monthly": {},
+    }
+    monkeypatch.setattr(
+        "kora_cli.telemetry.cost_telemetry.get_telemetry",
+        lambda: fake,
+    )
+    monkeypatch.setattr(
+        "kora_cli.telemetry.get_telemetry", lambda: fake
+    )
+
+    out = collect_route_rollups()
+    routes = {r.route: r for r in out}
+    assert routes["slack_dm"].escalation_rate == pytest.approx(12 / 30)
+    assert routes["email_inbound"].escalation_rate == 0.0
+    assert routes["slack_dm"].cost_estimate_usd_total == 0.4
+
+
+def test_observer_telemetry_failure_returns_empty(monkeypatch):
+    fake = MagicMock()
+    fake.snapshot.side_effect = RuntimeError("telemetry dead")
+    monkeypatch.setattr(
+        "kora_cli.telemetry.cost_telemetry.get_telemetry", lambda: fake
+    )
+    monkeypatch.setattr(
+        "kora_cli.telemetry.get_telemetry", lambda: fake
+    )
+    out = collect_route_rollups()
+    assert out == []
+
+
+# ---------------------------------------------------------------------------
+# Proposer
+# ---------------------------------------------------------------------------
+
+
+def test_proposer_skips_routes_below_min_calls(monkeypatch):
+    """Sample-size cap: a tiny route can't produce a proposal."""
+    monkeypatch.delenv(MIN_CALLS_ENV, raising=False)
+    rollups = [_rollup(route="rare", calls=5, escs=5)]
+    out = generate_proposals(rollups, now=datetime.now(timezone.utc))
+    assert out == []
+
+
+def test_proposer_skips_routes_below_threshold():
+    rollups = [_rollup(route="slack_dm", calls=100, escs=10)]
+    # 10% escalation < default 40% threshold → no proposal.
+    out = generate_proposals(rollups, now=datetime.now(timezone.utc))
+    assert out == []
+
+
+def test_proposer_emits_tighten_review_when_threshold_crossed():
+    rollups = [_rollup(route="slack_dm", calls=100, escs=50)]
+    out = generate_proposals(rollups, now=datetime.now(timezone.utc))
+    assert len(out) == 1
+    assert out[0].recommendation_kind == "tighten_review"
+    assert out[0].escalation_rate == pytest.approx(0.5)
+    assert "slack_dm" in out[0].rationale
+
+
+def test_proposer_sorted_by_confidence_desc():
+    rollups = [
+        _rollup(route="low", calls=20, escs=10),
+        _rollup(route="high", calls=200, escs=100),
+    ]
+    out = generate_proposals(rollups, now=datetime.now(timezone.utc))
+    assert [p.route for p in out] == ["high", "low"]
+
+
+def test_proposer_env_overrides(monkeypatch):
+    monkeypatch.setenv(MIN_CALLS_ENV, "5")
+    monkeypatch.setenv(TIGHTEN_THRESHOLD_ENV, "0.10")
+    rollups = [_rollup(route="slack_dm", calls=8, escs=2)]
+    out = generate_proposals(rollups, now=datetime.now(timezone.utc))
+    assert len(out) == 1
+
+
+# ---------------------------------------------------------------------------
+# Cycle
+# ---------------------------------------------------------------------------
+
+
+def _read_audit(tmp_path) -> list:
+    path = tmp_path / "kora_audit_log.jsonl"
+    if not path.is_file():
+        return []
+    return [
+        json.loads(line)
+        for line in path.read_text(encoding="utf-8").splitlines()
+        if line.strip()
+    ]
+
+
+@pytest.mark.asyncio
+async def test_cycle_disabled_short_circuits(tmp_path, monkeypatch):
+    monkeypatch.setenv(ENABLED_ENV, "false")
+    summary = await run_router_tuning_cycle()
+    assert summary["enabled"] is False
+    assert summary["proposals_generated"] == 0
+
+
+@pytest.mark.asyncio
+async def test_cycle_generates_persists_and_audits(tmp_path, monkeypatch):
+    """End-to-end: synthetic telemetry → 1 proposal persisted +
+    1 audit row with the canonical seam name."""
+    fake = MagicMock()
+    fake.snapshot.return_value = {
+        "rolling_24h": {
+            "slack_dm": {
+                "calls_count": 100,
+                "escalation_count": 60,
+                "cost_estimate_usd_total": 1.20,
+            }
+        },
+        "monthly": {},
+    }
+    monkeypatch.setattr(
+        "kora_cli.telemetry.cost_telemetry.get_telemetry", lambda: fake
+    )
+    monkeypatch.setattr(
+        "kora_cli.telemetry.get_telemetry", lambda: fake
+    )
+
+    summary = await run_router_tuning_cycle()
+    assert summary["enabled"] is True
+    assert summary["proposals_generated"] == 1
+    assert summary["proposals_persisted"] == 1
+
+    rows = _read_audit(tmp_path)
+    promo = [
+        r for r in rows if r["seam"] == "promotion.router_trigger_proposed"
+    ]
+    assert len(promo) == 1
+    assert promo[0]["details"]["route"] == "slack_dm"
+    assert promo[0]["details"]["recommendation_kind"] == "tighten_review"
+
+    # Pending proposal landed on disk.
+    pending_dir = (
+        tmp_path / "promotions" / "router_tuning" / "pending"
+    )
+    assert pending_dir.is_dir()
+    files = list(pending_dir.iterdir())
+    assert len(files) == 1
diff --git a/tests/kora_cli/promote/tool_trimming/__init__.py b/tests/kora_cli/promote/tool_trimming/__init__.py
new file mode 100644
index 000000000000..e69de29bb2d1
diff --git a/tests/kora_cli/promote/tool_trimming/test_tool_trimming.py b/tests/kora_cli/promote/tool_trimming/test_tool_trimming.py
new file mode 100644
index 000000000000..289f5a5f8d64
--- /dev/null
+++ b/tests/kora_cli/promote/tool_trimming/test_tool_trimming.py
@@ -0,0 +1,216 @@
+"""Tests for kora_cli.promote.tool_trimming."""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import pytest
+
+from kora_cli.audit.jsonl_sink import BATCH_SIZE_ENV, _reset_batching_for_tests
+from kora_cli.promote.tool_trimming.observer import (
+    RouteToolUsage,
+    collect_route_tool_usage,
+)
+from kora_cli.promote.tool_trimming.plugin import (
+    ENABLED_ENV,
+    run_tool_trimming_cycle,
+)
+from kora_cli.promote.tool_trimming.proposer import (
+    MIN_CALLS_ENV,
+    generate_proposals,
+)
+
+
+@pytest.fixture(autouse=True)
+def _isolate(tmp_path, monkeypatch):
+    monkeypatch.setenv("KORA_HOME", str(tmp_path))
+    monkeypatch.setenv("KORA_PROMOTIONS_DIR", str(tmp_path / "promotions"))
+    monkeypatch.setenv(
+        "KORA_AUDIT_LOG_PATH", str(tmp_path / "kora_audit_log.jsonl")
+    )
+    monkeypatch.setenv(BATCH_SIZE_ENV, "0")
+    _reset_batching_for_tests()
+    yield
+    _reset_batching_for_tests()
+
+
+def _write_audit(tmp_path: Path, entries: list) -> None:
+    path = tmp_path / "kora_audit_log.jsonl"
+    path.write_text(
+        "\n".join(json.dumps(e) for e in entries) + "\n", encoding="utf-8"
+    )
+
+
+def _audit_entry(
+    *,
+    tool_name: str = "kora__get_operational_state",
+    route: str | None = "slack_dm",
+    caller_session_id: str = "D1JOSH:1",
+    emitted_at: datetime | None = None,
+) -> dict:
+    if emitted_at is None:
+        emitted_at = datetime.now(timezone.utc) - timedelta(hours=1)
+    details: dict = {"tool_name": tool_name}
+    if route is not None:
+        details["route"] = route
+    return {
+        "emitted_at": emitted_at.isoformat(),
+        "seam": "reasoning.tool_called",
+        "details": details,
+        "caller_session_id": caller_session_id,
+        "source": "reasoning",
+    }
+
+
+# ---------------------------------------------------------------------------
+# Observer
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_observer_groups_by_route(tmp_path):
+    _write_audit(
+        tmp_path,
+        [
+            _audit_entry(tool_name="kora__a", route="slack_dm"),
+            _audit_entry(tool_name="kora__a", route="slack_dm"),
+            _audit_entry(tool_name="kora__b", route="email_inbound"),
+        ],
+    )
+    out = await collect_route_tool_usage(
+        since=datetime.now(timezone.utc) - timedelta(days=1)
+    )
+    routes = {r.route: r for r in out}
+    assert routes["slack_dm"].total_calls == 2
+    assert routes["slack_dm"].per_tool_calls == {"kora__a": 2}
+    assert routes["email_inbound"].total_calls == 1
+
+
+@pytest.mark.asyncio
+async def test_observer_derives_route_from_csid_prefix(tmp_path):
+    """When details.route is absent, the observer falls back to the
+    caller_session_id prefix convention."""
+    _write_audit(
+        tmp_path,
+        [
+            _audit_entry(
+                tool_name="kora__probe_check",
+                route=None,
+                caller_session_id="probe:fly:machine_down",
+            ),
+        ],
+    )
+    out = await collect_route_tool_usage(
+        since=datetime.now(timezone.utc) - timedelta(days=1)
+    )
+    assert [r.route for r in out] == ["probe_investigation"]
+
+
+# ---------------------------------------------------------------------------
+# Proposer
+# ---------------------------------------------------------------------------
+
+
+def test_proposer_emits_unused_tools_per_route(monkeypatch):
+    """slack_dm called tool_a; email called tool_b; available = both.
+    Expect: slack_dm gets a proposal for tool_b; email for tool_a."""
+    monkeypatch.setenv(MIN_CALLS_ENV, "2")
+    rollups = [
+        RouteToolUsage(
+            route="slack_dm",
+            total_calls=5,
+            tools_called={"kora__a"},
+            per_tool_calls={"kora__a": 5},
+        ),
+        RouteToolUsage(
+            route="email_inbound",
+            total_calls=5,
+            tools_called={"kora__b"},
+            per_tool_calls={"kora__b": 5},
+        ),
+    ]
+    out = generate_proposals(rollups, now=datetime.now(timezone.utc))
+    proposals = {p.route: p for p in out}
+    assert proposals["slack_dm"].unused_tools == ["kora__b"]
+    assert proposals["email_inbound"].unused_tools == ["kora__a"]
+
+
+def test_proposer_skips_routes_with_no_unused_tools(monkeypatch):
+    """Single-route case: tools_called == available → nothing to drop."""
+    monkeypatch.setenv(MIN_CALLS_ENV, "2")
+    rollups = [
+        RouteToolUsage(
+            route="slack_dm",
+            total_calls=10,
+            tools_called={"kora__a", "kora__b"},
+            per_tool_calls={"kora__a": 5, "kora__b": 5},
+        )
+    ]
+    out = generate_proposals(rollups, now=datetime.now(timezone.utc))
+    assert out == []
+
+
+def test_proposer_respects_min_calls(monkeypatch):
+    monkeypatch.setenv(MIN_CALLS_ENV, "10")
+    rollups = [
+        RouteToolUsage(
+            route="slack_dm",
+            total_calls=3,
+            tools_called={"kora__a"},
+            per_tool_calls={"kora__a": 3},
+        )
+    ]
+    out = generate_proposals(
+        rollups,
+        now=datetime.now(timezone.utc),
+        available_tool_names={"kora__a", "kora__b"},
+    )
+    assert out == []
+
+
+# ---------------------------------------------------------------------------
+# Cycle
+# ---------------------------------------------------------------------------
+
+
+def _read_audit(tmp_path) -> list:
+    path = tmp_path / "kora_audit_log.jsonl"
+    if not path.is_file():
+        return []
+    return [
+        json.loads(line)
+        for line in path.read_text(encoding="utf-8").splitlines()
+        if line.strip()
+    ]
+
+
+@pytest.mark.asyncio
+async def test_cycle_emits_audit_for_each_proposal(tmp_path, monkeypatch):
+    monkeypatch.setenv(MIN_CALLS_ENV, "2")
+    _write_audit(
+        tmp_path,
+        [
+            _audit_entry(tool_name="kora__a", route="slack_dm"),
+            _audit_entry(tool_name="kora__a", route="slack_dm"),
+            _audit_entry(tool_name="kora__b", route="email_inbound"),
+            _audit_entry(tool_name="kora__b", route="email_inbound"),
+        ],
+    )
+    summary = await run_tool_trimming_cycle()
+    assert summary["proposals_generated"] == 2
+    assert summary["proposals_persisted"] == 2
+
+    rows = _read_audit(tmp_path)
+    promo = [r for r in rows if r["seam"] == "promotion.tool_trim_proposed"]
+    assert len(promo) == 2
+    routes = sorted(r["details"]["route"] for r in promo)
+    assert routes == ["email_inbound", "slack_dm"]
+
+
+@pytest.mark.asyncio
+async def test_cycle_disabled_short_circuits(tmp_path, monkeypatch):
+    monkeypatch.setenv(ENABLED_ENV, "false")
+    summary = await run_tool_trimming_cycle()
+    assert summary["enabled"] is False