feat(kora): KR-PROMOTE-EXPAND-AND-TELEMETRY-WIRES-MEGABUCKET — 2nd loop + 7 route wires + snapshot v5

kora_cli/audit/jsonl_sink.py

@@ -191,6 +191,18 @@
     # operator consistently rejects are signal to tune the
     # proposer thresholds).
     "promotion.rejected",
+    # KR-PROMOTE-SNAPSHOT-EXPAND — second promotion loop. Observes
+    # reasoning tool-calls during status-shaped queries + proposes
+    # new snapshot fields that would have answered those queries at
+    # $0 LLM cost. The single seam ``promotion.snapshot_field_added``
+    # covers both the propose path (auto-apply OFF, v1 default) and
+    # the apply path (auto-apply ON). Payload carries
+    # proposal_id / proposed_field_path / proposed_collector_summary /
+    # cluster_size / sample_tool_calls / action (one of
+    # "proposed" / "auto_applied") / applier_diff_summary (only on
+    # auto_applied). Source is ``reasoning`` since the cluster
+    # input is reasoning audit.
+    "promotion.snapshot_field_added",
 ]
 
 SourceName = Literal[

kora_cli/listeners/__init__.py

@@ -86,3 +86,10 @@
 # swallowed by the heartbeat _loop; per-proposal failures don't
 # poison the batch.
 from kora_cli.listeners import promote_phrasebook_listener  # noqa: F401
+# KR-PROMOTE-SNAPSHOT-EXPAND — second promotion loop. Observes
+# ``reasoning.tool_called`` audit rows + proposes new snapshot fields
+# that would have pre-computed the answer. Imported AFTER the
+# phrasebook listener so both promotion loops register in dispatch
+# 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

kora_cli/listeners/promote_snapshot_expand_listener.py

@@ -0,0 +1,66 @@
+"""Heartbeat-scheduled snapshot-expand promotion cycle — KR-PROMOTE-SNAPSHOT-EXPAND.
+
+Registers :func:`run_snapshot_expand_cycle` as a periodic task
+against the heartbeat scheduler. Cadence operator-tunable via
+``KORA_PROMOTE_SNAPSHOT_EXPAND_INTERVAL_SEC`` (default 86400s = once
+daily). Master kill-switch
+``KORA_PROMOTE_SNAPSHOT_EXPAND_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 listener — the heartbeat scheduler
+is interval-based; the bucket spec's ``"0 7 * * *"`` cron suggestion
+is documented but not honored verbatim. Daily-interval is sufficient
+for a batch promotion loop (proposals go into the audit JSONL for
+operator review; not a real-time surface).
+
+# Fail-soft startup
+
+Cycle exceptions are swallowed by the heartbeat scheduler's
+``_loop``; per-proposal failures are 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.snapshot_expand.cycle import (
+    get_interval_seconds,
+    run_snapshot_expand_cycle,
+)
+
+logger = logging.getLogger(__name__)
+
+
+async def _periodic_task() -> None:
+    """Thin async wrapper so the heartbeat scheduler's signature is
+    satisfied. Cycle's summary dict logged at DEBUG so operator can
+    grep this specific task name when triaging."""
+    try:
+        summary = await run_snapshot_expand_cycle()
+        logger.debug(
+            "[kora.promote.snapshot_expand.listener] tick complete: "
+            "proposals_applied=%d auto_apply_mode=%s duration_ms=%d",
+            summary.get("proposals_applied", 0),
+            summary.get("auto_apply_mode", False),
+            summary.get("duration_ms", 0),
+        )
+    except Exception as exc:
+        # Belt-and-suspenders — cycle is fail-soft, but the wrapper
+        # catches anything that escapes (e.g., asyncio cancellation
+        # during shutdown).
+        logger.warning(
+            "[kora.promote.snapshot_expand.listener] tick raised %r — "
+            "next scheduled run will retry",
+            exc,
+        )
+
+
+register_periodic_task(
+    "promote_snapshot_expand_cycle",
+    interval_seconds=float(get_interval_seconds()),
+    callable=_periodic_task,
+)

kora_cli/promote/snapshot_expand/__init__.py

@@ -0,0 +1,38 @@
+"""Snapshot-expand promotion loop — KR-PROMOTE-SNAPSHOT-EXPAND.
+
+Second promotion loop (after :mod:`kora_cli.promote.phrasebook`).
+Observes which tool calls fire during status-shaped reasoning queries
+and proposes new snapshot fields that would have answered those
+queries at $0 LLM cost.
+
+# Loop shape
+
+  1. :mod:`.observer` — read recent ``reasoning.tool_called`` audit
+     rows; cluster by tool_name (proxy for "what was being asked").
+  2. :mod:`.proposer` — for each cluster ≥ min_cluster_size, propose
+     a new snapshot field whose collector would have returned the
+     same answer.
+  3. :mod:`.applier` — when AUTO_APPLY is OFF (v1 default), emit a
+     ``promotion.snapshot_field_added`` audit row with
+     ``action="proposed"``. When ON, also write a stub collector
+     entry + audit ``action="auto_applied"``.
+  4. :mod:`.cycle` — orchestrator the periodic-task listener calls.
+
+# Cost discipline
+
+Per ``feedback-promotion-loops-self-improving-subsystems``: target
+~$0.01-0.05/day combined across promotion loops. This loop is
+fully lexical (no LLM): clustering by tool_name groups exact-name
+matches, and the proposer derives field names + summaries from
+the cluster shape directly. Per cycle: **$0**. Daily ceiling
+absorbed entirely by the phrasebook loop's ≤$0.005/day.
+
+# Auto-apply safety
+
+Per STOP-ASK §4 of the bucket spec: schema-bumping at runtime is
+fragile. v1 ships with ``KORA_PROMOTE_SNAPSHOT_EXPAND_AUTO_APPLY=false``
+by default — proposals land in the audit JSONL only, and operator
+review (via a future cockpit endpoint or by reading the audit
+seam directly) is the gate before any schema change. Operator
+can flip the env once trust is built.
+"""

kora_cli/promote/snapshot_expand/applier.py

@@ -0,0 +1,157 @@
+"""Proposal applier — KR-PROMOTE-SNAPSHOT-EXPAND.
+
+For each :class:`SnapshotFieldProposal` produced by the proposer,
+this module:
+
+  * ALWAYS emits a ``promotion.snapshot_field_added`` audit row.
+    The ``action`` field distinguishes ``"proposed"`` (v1 default
+    — operator reviews via the audit JSONL or a future cockpit
+    endpoint, then manually adds the collector) from
+    ``"auto_applied"`` (env-gated — the loop would write a stub
+    collector + bump SCHEMA_VERSION).
+  * When ``KORA_PROMOTE_SNAPSHOT_EXPAND_AUTO_APPLY=true``, ALSO:
+    persists a minimal proposal record under
+    ``${KORA_HOME}/promotions/snapshot_expand/applied/<proposal_id>.json``
+    so the operator has an at-rest artifact + emits
+    ``action="auto_applied"`` in the audit row.
+
+# Safety posture
+
+Per STOP-ASK §4 of the bucket spec, schema-bumping at runtime is
+fragile. The auto-apply path in v1 ONLY writes a stub record —
+it does NOT modify ``state_snapshot.py`` or bump
+``SCHEMA_VERSION`` itself. A separate operator-driven step (a
+future cockpit "approve + scaffold" endpoint, or a manual code
+edit) is still required to actually add the collector. The
+auto_applied action is "we've recorded this; please scaffold."
+This is the conservative interpretation of "auto-apply with
+audit trail" — the audit trail is unconditional, the actual code
+change stays operator-gated.
+
+# Why not write the collector directly via codegen
+
+Writing into ``state_snapshot.py`` from a cron task would mean:
+  * Running code that mutates its own runtime imports.
+  * Bumping SCHEMA_VERSION mid-process (cache invalidation; FE
+    type drift; consumer assumptions about stable shape).
+  * No code review on the generated collector.
+The bucket spec's docstring template promised codegen as
+"future-when-trusted." v1 takes the safer first step: the audit
+trail establishes the loop's value over weeks, and operator
+approves the scaffolding manually.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+from pathlib import Path
+from typing import Any, Dict
+
+from .proposer import SnapshotFieldProposal, proposal_to_dict
+
+logger = logging.getLogger(__name__)
+
+
+AUTO_APPLY_ENV = "KORA_PROMOTE_SNAPSHOT_EXPAND_AUTO_APPLY"
+PROMOTIONS_ROOT_ENV = "KORA_PROMOTIONS_DIR"
+_APPLIED_RELATIVE = Path("promotions") / "snapshot_expand" / "applied"
+
+
+def _is_auto_apply_enabled() -> bool:
+    """Read the env. Default ``false`` per v1 spec — flip to true
+    only after operator builds trust with the loop."""
+    raw = os.environ.get(AUTO_APPLY_ENV, "false").strip().lower()
+    return raw in {"true", "1", "yes", "on"}
+
+
+def _applied_dir() -> Path:
+    """Resolve the applied-proposal store directory.
+
+    Mirrors the phrasebook store's env-override pattern so tests
+    can redirect via ``KORA_PROMOTIONS_DIR`` without touching
+    KORA_HOME.
+    """
+    override = os.environ.get(PROMOTIONS_ROOT_ENV, "").strip()
+    if override:
+        return Path(override) / "snapshot_expand" / "applied"
+    from kora_constants import get_kora_home
+
+    return get_kora_home() / _APPLIED_RELATIVE
+
+
+def _persist_applied(proposal: SnapshotFieldProposal) -> None:
+    """Atomic-write the proposal as an applied record. Best-effort
+    — OSError logged + swallowed (audit row is the canonical
+    artifact)."""
+    try:
+        target_dir = _applied_dir()
+        target_dir.mkdir(parents=True, exist_ok=True)
+        target = target_dir / f"{proposal.proposal_id}.json"
+        tmp = target.with_suffix(".json.tmp")
+        tmp.write_text(
+            json.dumps(proposal_to_dict(proposal), indent=2),
+            encoding="utf-8",
+        )
+        os.replace(tmp, target)
+    except OSError as exc:
+        logger.warning(
+            "[kora.promote.snapshot_expand.applier] persist failed for "
+            "%s: %r — audit row still emitted",
+            proposal.proposal_id,
+            exc,
+        )
+
+
+def _emit_audit(
+    proposal: SnapshotFieldProposal, *, action: str
+) -> None:
+    """Emit the ``promotion.snapshot_field_added`` audit row.
+
+    Payload shape:
+      * Full proposal projection (proposal_id / cluster_size /
+        proposed_field_path / proposed_collector_summary /
+        source_tool_name / sample_caller_session_ids /
+        confidence / created_at)
+      * ``action`` — one of ``"proposed"`` (auto-apply OFF) or
+        ``"auto_applied"`` (auto-apply ON; v1 means audit + stub
+        persist, NOT live schema mutation — see module docstring).
+    """
+    try:
+        from kora_cli.audit.jsonl_sink import emit_audit
+    except Exception as exc:
+        logger.warning(
+            "[kora.promote.snapshot_expand.applier] audit import failed: "
+            "%r — promotion.snapshot_field_added skipped",
+            exc,
+        )
+        return
+    payload: Dict[str, Any] = proposal_to_dict(proposal)
+    payload["action"] = action
+    try:
+        emit_audit(
+            "promotion.snapshot_field_added",
+            payload,
+            caller_session_id=(
+                f"promotion:snapshot_expand:{proposal.proposal_id}"
+            ),
+            source="reasoning",
+        )
+    except Exception as exc:
+        logger.warning(
+            "[kora.promote.snapshot_expand.applier] emit_audit raised "
+            "%r — proposal lost from the audit stream",
+            exc,
+        )
+
+
+def apply_proposal(proposal: SnapshotFieldProposal) -> str:
+    """Apply one proposal. Returns the ``action`` string emitted in
+    the audit row (``"proposed"`` or ``"auto_applied"``)."""
+    if _is_auto_apply_enabled():
+        _persist_applied(proposal)
+        _emit_audit(proposal, action="auto_applied")
+        return "auto_applied"
+    _emit_audit(proposal, action="proposed")
+    return "proposed"